spring boot + kafka 使用详细步骤[通俗易懂]

spring boot + kafka 使用详细步骤[通俗易懂]一,Kafka的安装以及配置1.下载文件wgethttp://mirror.bit.edu.cn/apache/kafka/0.11.0.0/kafka_2.11-0.11.0.0.tgz2.安装tarxzvfkafka_2.11-0.11.0.0.tgz-C/usr/local/3.配置(服务器在阿里云ECS上)vi%kafka_home%/server…

大家好,又见面了,我是你们的朋友全栈君。

一,Kafka的安装以及配置

1.下载文件

wget http://mirror.bit.edu.cn/apache/kafka/0.11.0.0/kafka_2.11-0.11.0.0.tgz

2.安装

tar xzvf kafka_2.11-0.11.0.0.tgz -C /usr/local/

3.配置(服务器在阿里云ECS上)

vi %kafka_home%/server.properties

############################# Server Basics #############################
broker.id=0
port=9092
host.name=阿里云内网ip
advertised.host.name=阿里云外网ip
delete.topic.enable=true

....其他的配置不变

二,Spring boot操作Kafka

1.Spring boot pom依赖

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>1.5.6.RELEASE</version>
    <relativePath/> <!-- lookup parent from repository -->
</parent>

2.kafka依赖

 <!--kafka支持-->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

3.kafka配置

#kafka相关配置
spring.kafka.bootstrap-servers=阿里云外网ip:9092
#设置一个默认组
spring.kafka.consumer.group-id=defaultGroup
#key-value序列化反序列化
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.batch-size=65536
spring.kafka.producer.buffer-memory=524288

4.kafka发送代码

@Component
public class KafkaSender {
    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * 发送消息到kafka,主题为test
     */
    public void sendTest(){
        kafkaTemplate.send("test","hello,kafka  "  + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS")));
    }
}

5.Kafka消费代码

public class KafkaConsumer {

    /**
     * 监听test主题,有消息就读取
     * @param message
     */
    @KafkaListener(topics = {"test"})
    public void consumer(String message){
        log.info("test topic message : {}", message);
    }
}

6,启动测试代码

@SpringBootApplication
@EnableScheduling
public class ServerApplication {
    @Autowired
    private KafkaSender kafkaSender;


    public static void main(String[] args) {
        SpringApplication.run(ServerApplication.class, args);
    }

    //然后每隔1分钟执行一次
    @Scheduled(fixedRate = 1000 * 60)
    public void testKafka() throws Exception {
        kafkaSender.sendTest();
    }
}

三,以上就是完整的spring boot整合Kafka消息队列的步骤.

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/141676.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号