Spring Boot 整合 Kafka

Kafka 环境搭建

1. kafka 安装、配置、启动、测试说明：
3. 1. 安装：直接官网下载安装包，解压到指定位置即可（kafka 依赖的 Zookeeper 在文件中已包含）
4. 下载地址：https://kafka.apache.org/downloads
5. 示例版本：kafka_2.13-2.8.0.tgz
6. 下载后可本地解压安装，解压位置自选，如 D:\Java 下
7. 解压命令：tar -zxvf kafka_2.13-2.8.0.tgz
8. PS：可在 idea 命令行窗口或 git 提供的命令窗口中进行命令操作
9. 使用 git 提供的命令窗口：空白文件夹中右键——》Git Bash Here 即可打开
11. 2. 添加地址配置
12. 在 D:\Java\kafka_2.13-2.8.0\config\server.properties 中搜索添加以下两行配置：
13. listeners=PLAINTEXT://localhost:9092
14. advertised.listeners=PLAINTEXT://localhost:9092
15. 说明：以上配置默认是注释掉的，可搜索找到，根据需求进行自定义地址配置
17. 重要说明：以下命令操作默认都是在 D:\Java\kafka_2.13-2.8.0\ 即 kafaka 根目录下进行！
19. 3. 使用配置文件方式后台启动/关闭 Zookeeper 服务
20. 启动：bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
21. 关闭【自选】：bin/zookeeper-server-stop.sh -daemon config/zookeeper.properties
23. 4. 使用配置文件方式后台启动/关闭 kafka 服务
24. 启动：bin/kafka-server-start.sh -daemon config/server.properties
25. 关闭【自选】：bin/kafka-server-stop.sh -daemon config/server.properties
27. 5. 服务测试
29. 5.1 创建主题
30. bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic Hello-Kafka
32. 5.2 查看主题（可能需要查一会儿）
33. bin/kafka-topics.sh --list --zookeeper localhost:2181
35. 说明：发送消息和监听消息需要打开两个窗口进行测试！
37. 5.3 发送消息（kafka 根目录下新建窗口）
38. bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Hello-Kafka
39. 输入以上命令回车后，可继续输入内容测试消息发送
41. 5.4 监听消息（kafka 根目录下新建窗口）
42. bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic Hello-Kafka --from-beginning
43. 输入以上命令后，可观察消息接收情况，并且可在消息发送窗口继续发送消息测试此监听窗口的接收情况，正常接收，则服务环境搭建成功。

复制代码

Spring Boot 整合 Kafka

环境：自行创建 Spring Boot 项目，添加测试依赖，并启动 Zookeeper 和 kafka 服务。
注意：Zookeeper 默认好像占用 8080 端口，自己注意端口占用问题。

1. 添加依赖

1. <dependency>
2.     <groupId>org.springframework.kafka</groupId>
3.     <artifactId>spring-kafka</artifactId>
4. </dependency>

复制代码

2. 添加配置

1. # kafka 配置
2. spring:
3.   kafka:
4.     bootstrap-servers: localhost:9092
5.     producer:
6.       # 发生错误后，消息重发的次数。
7.       retries: 1
8.       #当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。
9.       batch-size: 16384
10.       # 设置生产者内存缓冲区的大小。
11.       buffer-memory: 33554432
12.       # 键的序列化方式
13.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
14.       # 值的序列化方式
15.       value-serializer: org.apache.kafka.common.serialization.StringSerializer
16.       # acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。
17.       # acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。
18.       # acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。
19.       acks: 1
20.     consumer:
21.       # 自动提交的时间间隔 在spring boot 2.X 版本中这里采用的是值的类型为Duration 需要符合特定的格式，如1S,1M,2H,5D
22.       auto-commit-interval: 1S
23.       # 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：
24.       # latest（默认值）在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）
25.       # earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录
26.       auto-offset-reset: earliest
27.       # 是否自动提交偏移量，默认值是true,为了避免出现重复数据和数据丢失，可以把它设置为false,然后手动提交偏移量
28.       enable-auto-commit: false
29.       # 键的反序列化方式
30.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
31.       # 值的反序列化方式
32.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
33.     listener:
34.       # 在侦听器容器中运行的线程数。
35.       concurrency: 5
36.       # listner负责ack，每调用一次，就立即commit
37.       ack-mode: manual_immediate
38.       missing-topics-fatal: false

复制代码

3. 创建消息生产者

1. @Component
2. public class KafkaProducer {
4.     private Logger logger = LoggerFactory.getLogger(KafkaProducer.class);
6.     @Resource
7.     private KafkaTemplate<String, Object> kafkaTemplate;
9.     public static final String TOPIC_TEST = "Hello-Kafka";
11.     public static final String TOPIC_GROUP = "test-consumer-group";
13.     public void send(Object obj) {
14.         String obj2String = JSON.toJSONString(obj);
15.         logger.info("准备发送消息为：{}", obj2String);
17.         // 发送消息
18.         ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC_TEST, obj);
19.         future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
20.             @Override
21.             public void onFailure(Throwable throwable) {
22.                 //发送失败的处理
23.                 logger.info(TOPIC_TEST + " - 生产者 发送消息失败：" + throwable.getMessage());
24.             }
26.             @Override
27.             public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
28.                 //成功的处理
29.                 logger.info(TOPIC_TEST + " - 生产者 发送消息成功：" + stringObjectSendResult.toString());
30.             }
31.         });
32.     }
34. }

复制代码

4. 创建消息消费者

1. @Component
2. public class KafkaConsumer {
4.     private Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);
6.     @KafkaListener(topics = KafkaProducer.TOPIC_TEST, groupId = KafkaProducer.TOPIC_GROUP)
7.     public void topicTest(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
8.         Optional<?> message = Optional.ofNullable(record.value());
9.         if (message.isPresent()) { // 包含非空值，则执行
10.             Object msg = message.get();
11.             logger.info("topic_test 消费了： Topic:" + topic + ",Message:" + msg);
12.             ack.acknowledge(); // 确认成功消费一个消息
13.         }
14.     }
16. }

复制代码

5. 消息发送测试

1. @RunWith(SpringRunner.class)
2. @SpringBootTest
3. public class KafkaProducerTest {
5.     private Logger logger = LoggerFactory.getLogger(KafkaProducerTest.class);
7.     @Resource
8.     private KafkaProducer kafkaProducer; // 注意使用自己创建的，看清楚！
10.     /*
11.       测试之前需要开启 Kafka 服务
12.       启动 Zookeeper：bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
13.       启动 Kafka：bin/kafka-server-start.sh -daemon config/server.properties
15.       测试结果数据：
17.       准备发送消息为："你好，我是Lottery 001"
18.       Hello-Kafka - 生产者 发送消息成功：SendResult [producerRecord=ProducerRecord(topic=Hello-Kafka, partition=null,
19.       headers=RecordHeaders(headers = [], isReadOnly = true), key=null, value=你好，我是Lottery 001, timestamp=null),
20.       recordMetadata=Hello-Kafka-0@47]
22.       topic_test 消费了： Topic:Hello-Kafka,Message:你好，我是Lottery 001
23.      */
24.     @Test
25.     public void test_send() throws InterruptedException {
26.         // 循环发送消息
27.         while (true) {
28.             kafkaProducer.send("你好，我是Lottery 001");
29.             Thread.sleep(3500);
30.         }
31.     }
33. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！