spring boot集成kafka开发，接收kafka消息，Java（1）

男娘i 2023-10-01 19:51 84阅读 0赞

在这篇文章（1）基础上：

[kafka连接zookeeper配置搭建环境、安装和启动，Windows环境下\_Zhang Phil-CSDN博客kafka，分布式集群架构下，高性能的流式事件数据（主要是消息）集成、发布（生产）和订阅（分发、消费）组件（中间件）。kafka依赖zooeeper（数据后端），这里有Windows下安装配置启动zookeeper的 文章（1）：Windows 10环境zookeeper单机伪集群部署和配置\_Zhang Phil-CSDN博客Windows 10环境zookeeper单机伪集群部署和配置1，首先到zookeeper项目主页地址下载项目包，https://archive.apache.org/dist![favicon32.ico][]https://blog.csdn.net/zhangphil/article/details/123086790][kafka_zookeeper_Windows_Zhang Phil-CSDN_kafka_kafka_zooeeper_Windows_zookeeper_ _1_Windows 10_zookeeper_Zhang Phil-CSDN_Windows 10_zookeeper_1_zookeeper_https_archive.apache.org_dist_favicon32.ico_https_blog.csdn.net_zhangphil_article_details_123086790]

实现一个简单功能：通过kafka控制台发送消息，然后在spring boot编程实现的java代码程序里面接收消息。

按照文章（1）启动zookeeper和kafka，使用命令：

kafka-topics.bat --list --bootstrap-server 127.0.0.1:9092

查看Windows环境下，列出主机127.0.0.1上kafka的主题topic。本例选择其中一个topic主题，使用kafka命令在控制台发送消息，接收方则用spring框架里面的java程序实现。

下面开始实现spring框架的kafka消息接收java代码。

（1）在application.properties里面配置spring环境中的kafka参数：

server.port=7999
    spring.kafka.bootstrap-servers=127.0.0.1:9092
    spring.kafka.consumer.bootstrap-servers=127.0.0.1:9092
    spring.kafka.consumer.group-id=zhangphil-group
    spring.kafka.consumer.key-deserializer=org.springframework.kafka.support.serializer.StringOrBytesSerializer
    spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonSerializer
    spring.kafka.producer.bootstrap-servers=127.0.0.1:9092
    spring.kafka.producer.key-serializer=org.springframework.kafka.support.serializer.StringOrBytesSerializer
    spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

(2)添加注解@EnableKafka， 打开spring application的kafka配置：

import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.kafka.annotation.EnableKafka;
    
    @EnableKafka
    @SpringBootApplication
    public class SpringKafkaApplication {
        public static void main(String[] args) {
            SpringApplication.run(SpringKafkaApplication.class, args);
        }
    }

（3）写一个kafka消息接收器：

import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    
    @Component
    public class MyKafka {
        @KafkaListener(topics = "zhangphil_demo", groupId = "${spring.kafka.consumer.group-id}")
        public void msg_listener(ConsumerRecord<?, ?> record) {
            System.out.println(record.topic() + "," + record.offset() + "," + record.value());
            System.out.println(record);
        }
    }

以上java程序写完后，即可启动spring application。

然后在kafka控制台，用命令:

kafka-console-producer.bat --topic zhangphil_demo --bootstrap-server localhost:9092

在主题topic为zhangphil\_demo里面发送测试消息 hello,world!2022,运行输出如图：

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAemhhbmdwaGls_size_20_color_FFFFFF_t_70_g_se_x_16][]

[favicon32.ico]: https://g.csdnimg.cn/static/logo/favicon32.ico
[kafka_zookeeper_Windows_Zhang Phil-CSDN_kafka_kafka_zooeeper_Windows_zookeeper_ _1_Windows 10_zookeeper_Zhang Phil-CSDN_Windows 10_zookeeper_1_zookeeper_https_archive.apache.org_dist_favicon32.ico_https_blog.csdn.net_zhangphil_article_details_123086790]: https://blog.csdn.net/zhangphil/article/details/123086790
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBAemhhbmdwaGls_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/713b93e173f74bbb841b7b6c10725cc7.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBAemhhbmdwaGls,size_20,color_FFFFFF,t_70,g_se,x_16