文章目录

  • Spring Kafka概述
  • Code
    • 依赖
    • 配置文件
    • 消息
    • 生产者
    • 消费者
    • 单元测试： 同步发送
    • 测试
    • 单元测试： 异步发送
  • 源码地址

Spring Kafka概述

Spring提供了 Spring-Kafka 项目来操作 Kafka。

https://spring.io/projects/spring-kafka

Code

我们先对 Kafka-Spring 做个快速入门，实现 Producer发送消息 ，同时Consumer 消费消息。

依赖

 <dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- 引入 Spring-Kafka 依赖 --> <dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><scope>test</scope></dependency></dependencies>

Spring Boot 已经提供了 Kafka 的自动化配置的支持，但没有提供 spring-boot-kafka-starter 包…

配置文件

spring:# Kafka 配置项，对应 KafkaProperties 配置类kafka:bootstrap-servers: 192.168.126.140:9092 # 指定 Kafka Broker 地址，可以设置多个，以逗号分隔# Kafka Producer 配置项producer:acks: 1 # 0-不应答。1-leader 应答。all-所有 leader 和 follower 应答。retries: 3 # 发送失败时，重试发送的次数key-serializer: org.apache.kafka.common.serialization.StringSerializer # 消息的 key 的序列化value-serializer: org.springframework.kafka.support.serializer.JsonSerializer # 消息的 value 的序列化# Kafka Consumer 配置项consumer:auto-offset-reset: earliest # 设置消费者分组最初的消费进度为 earliestkey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.springframework.kafka.support.serializer.JsonDeserializerproperties:spring:json:trusted:packages: com.artisan.springkafka.domain# Kafka Consumer Listener 监听器配置listener:missing-topics-fatal: false # 消费监听接口监听的主题不存在时，默认会报错。所以通过设置为 false ，解决报错logging:level:org:springframework:kafka: ERROR # spring-kafka  apache:kafka: ERROR # kafka

spring.kafka 配置项， 对应 KafkaProperties 配置类 。Spring Boot 提供的 KafkaAutoConfiguration 自动化配置类，实现 Kafka 的自动配置，创建相应的 Producer 和 Consumer 。

特别说明一下： 生产者 的value-serializer 配置了 Spring-Kafka 提供的 JsonSerializer 序列化类， 使用 JSON 的方式，序列化复杂的 Message 消息。

消费者的 value-serializer 配置，同样使用了 JsonDeserializer 反序列化类，因为稍后我们要使用 JSON 的方式，反序列化复杂的 Message 消息。

properties.spring.json.trusted.packages 需要配置com.artisan.springkafka.domain 包下的 Message 类们。因为 JsonDeserializer 在反序列化消息时，考虑到安全性，只反序列化成信任的 Message 类。 务必配置

在序列化时，使用了 JsonSerializer 序列化 Message 消息对象，它会在 Kafka 消息 Headers 的 TypeId 上，值为 Message 消息对应的类全名。

在反序列化时，使用了 JsonDeserializer 序列化出 Message 消息对象，它会根据 Kafka 消息 Headers 的 TypeId 的值，反序列化消息内容成该 Message 对象。

消息

package com.artisan.springkafka.domain;/*** @author 小工匠* @version 1.0* @description: TODO* @date 2021/2/17 22:27* @mark: show me the code , change the world*/
public class MessageMock {private Integer id ;private String name ;public MessageMock() {}public MessageMock(Integer id, String name) {this.id = id;this.name = name;}public Integer getId() {return id;}public void setId(Integer id) {this.id = id;}public String getName() {return name;}public void setName(String name) {this.name = name;}@Overridepublic String toString() {return "MessageMock{" +"id=" + id +", name='" + name + '\'' +'}';}
}

生产者

package com.artisan.springkafka.producer;import com.artisan.springkafka.constants.TOPIC;
import com.artisan.springkafka.domain.MessageMock;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;import java.util.Random;
import java.util.concurrent.ExecutionException;/*** @author 小工匠* @version 1.0* @description: TODO* @date 2021/2/17 22:25* @mark: show me the code , change the world*/@Component
public class ArtisanProducerMock {@Autowiredprivate KafkaTemplate<Object,Object> kafkaTemplate ;/*** 同步发送* @return* @throws ExecutionException* @throws InterruptedException*/public SendResult sendMsgSync() throws ExecutionException, InterruptedException {// 模拟发送的消息Integer id = new Random().nextInt(100);MessageMock messageMock = new MessageMock(id,"artisanTestMessage-" + id);// 同步等待return  kafkaTemplate.send(TOPIC.TOPIC, messageMock).get();}}

消费者

模拟两个不同消费组下的消费者 ，测试消费情况

【消费者A 】

package com.artisan.springkafka.consumer;import com.artisan.springkafka.domain.MessageMock;
import com.artisan.springkafka.constants.TOPIC;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;/*** @author 小工匠* @version 1.0* @description: TODO* @date 2021/2/17 22:33* @mark: show me the code , change the world*/@Component
public class ArtisanCosumerMock {private Logger logger = LoggerFactory.getLogger(getClass());private static final String CONSUMER_GROUP_PREFIX = "MOCK-A" ;@KafkaListener(topics = TOPIC.TOPIC ,groupId = CONSUMER_GROUP_PREFIX + TOPIC.TOPIC)public void onMessage(MessageMock messageMock){logger.info("【接受到消息][线程:{} 消息内容：{}]", Thread.currentThread().getName(), messageMock);}}

在方法上添加了 @KafkaListener 注解，指定了消费的 Topic 和 消费者分组 。

建议：建 一个类，对应一个方法。一个消费者分组，仅消费一个 Topic ，确保每个消费者分组职责单一。

【消费者B 】

package com.artisan.springkafka.consumer;import com.artisan.springkafka.domain.MessageMock;
import com.artisan.springkafka.constants.TOPIC;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;/*** @author 小工匠* @version 1.0* @description: TODO* @date 2021/2/17 22:33* @mark: show me the code , change the world*/@Component
public class ArtisanCosumerMockDiffConsumeGroup {private Logger logger = LoggerFactory.getLogger(getClass());private static final String CONSUMER_GROUP_PREFIX = "MOCK-B" ;@KafkaListener(topics = TOPIC.TOPIC ,groupId = CONSUMER_GROUP_PREFIX + TOPIC.TOPIC)public void onMessage(MessageMock messageMock){logger.info("【接受到消息][线程:{} 消息内容：{}]", Thread.currentThread().getName(), messageMock);}}

在方法上添加了 @KafkaListener 注解，指定了消费的 Topic 和 消费者分组 。

消费组和第一个消费者属于不同的消费组，请注意。

单元测试： 同步发送

package com.artisan.springkafka.produceTest;import com.artisan.springkafka.SpringkafkaApplication;
import com.artisan.springkafka.producer.ArtisanProducerMock;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.support.SendResult;
import org.springframework.test.context.junit4.SpringRunner;import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;/*** @author 小工匠* @version 1.0* @description: TODO* @date 2021/2/17 22:40* @mark: show me the code , change the world*/@RunWith(SpringRunner.class)
@SpringBootTest(classes = SpringkafkaApplication.class)
public class ProduceMockTest {private Logger logger = LoggerFactory.getLogger(getClass());@Autowiredprivate ArtisanProducerMock artisanProducerMock;@Testpublic void testSyncSend() throws ExecutionException, InterruptedException {SendResult sendResult = artisanProducerMock.sendMsgSync();logger.info("testSyncSend Result =  topic:[{}] , partition:[{}], offset:[{}]",sendResult.getRecordMetadata().topic(),sendResult.getRecordMetadata().partition(),sendResult.getRecordMetadata().offset());// 阻塞等待，保证消费new CountDownLatch(1).await();}}

在方法内部，调用 KafkaTemplate#send(topic, data) 方法，异步发送消息。不过，因为后面调用了 ListenableFuture#get() 方法，阻塞等待发送结果，实现了同步的效果。

测试

运行上面的单元测试 ，

2021-02-18 00:13:50.789  INFO 24768 --- [           main] c.a.s.produceTest.ProduceMockTest        : testSyncSend Result =  topic:[MOCK_TOPIC] , partition:[0], offset:[12]
2021-02-18 00:13:50.849  INFO 24768 --- [ntainer#1-0-C-1] a.s.c.ArtisanCosumerMockDiffConsumeGroup : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#1-0-C-1 消息内容：MessageMock{id=51, name='artisanTestMessage-51'}]
2021-02-18 00:13:50.849  INFO 24768 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock        : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容：MessageMock{id=51, name='artisanTestMessage-51'}]

可以看到我们发送了一个消息到MOCK_TOPIC上， 两个消费者属于不同的消费者组，均订阅了该TOPIC， 从结果上可以看到 该消息 可以分别被消费者组 “MOCK-ATOPIC” 和消费者组 “MOCK-BTOPIC” 都消费一次。

但是，如果我们启动多个该示例的实例，则消费者分组 “MOCK-ATOPIC” 和 “MOCK-BTOPIC” 都会有多个 Consumer实例， 结果会怎样呢？

我们再运行一个， 上次的不要关哈

此时，我们再发送一条消息到MOCK_TOPIC，只会被 "MOCK-ATOPIC"的一个 Consumer 消费一次，也同样只会被 “MOCK-BTOPIC” 的一个 Consumer 消费一次。

这个有啥用呢？ 举个例子

通过集群消费的机制，可以实现针对相同 Topic ，不同消费者分组实现各自的业务逻辑。

比如说用户注册成功时，发送一条 Topic 为 “XXXX” 的消息。 不同模块使用不同的消费者分组，订阅该 Topic ，实现各自的拓展逻辑：

  • 积分模块：给用户增加 积分
  • 优惠劵模块：发放新用户专享优惠
  • …

这样，就可以将注册成功后的业务拓展逻辑，实现业务上的解耦，未来也更加容易拓展。同时，也提高了注册接口的性能。

单元测试： 异步发送

com.artisan.springkafka.producer.ArtisanProducerMock

  public ListenableFuture<SendResult<Object, Object>> sendMsgASync() throws ExecutionException, InterruptedException {// 模拟发送的消息Integer id = new Random().nextInt(100);MessageMock messageMock = new MessageMock(id,"messageSendByAsync-" + id);// 异步发送消息ListenableFuture<SendResult<Object, Object>> result = kafkaTemplate.send(TOPIC.TOPIC, messageMock);return result ;}

单元测试

com.artisan.springkafka.produceTest.ProduceMockTest

@Testpublic void testAsynSend() throws ExecutionException, InterruptedException {artisanProducerMock.sendMsgASync().addCallback(new ListenableFutureCallback<SendResult<Object, Object>>() {@Overridepublic void onFailure(Throwable throwable) {logger.info(" 发送异常{}]]", throwable);}@Overridepublic void onSuccess(SendResult<Object, Object> objectObjectSendResult) {logger.info("回调结果 Result =  topic:[{}] , partition:[{}], offset:[{}]",objectObjectSendResult.getRecordMetadata().topic(),objectObjectSendResult.getRecordMetadata().partition(),objectObjectSendResult.getRecordMetadata().offset());}});// 阻塞等待，保证消费new CountDownLatch(1).await();}

测试结果 同上

2021-02-18 00:40:22.443  INFO 24056 --- [ad | producer-1] c.a.s.produceTest.ProduceMockTest        : 回调结果 Result =  topic:[MOCK_TOPIC] , partition:[0], offset:[17]
2021-02-18 00:40:22.504  INFO 24056 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock        : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容：MessageMock{id=37, name='messageSendByAsync-37'}]
2021-02-18 00:40:22.504  INFO 24056 --- [ntainer#1-0-C-1] a.s.c.ArtisanCosumerMockDiffConsumeGroup : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#1-0-C-1 消息内容：MessageMock{id=37, name='messageSendByAsync-37'}]

  • 通过日志可以看到，发送的消息，分别被 ArtisanCosumerMock 和 ArtisanCosumerMockDiffConsumeGroup 两个消费者（位于不同的消费者分组）均消费了一次。

  • 两个消费者在不同的线程中，消费了该条消息

源码地址

https://github.com/yangshangwei/boot2/tree/master/springkafka

Apache Kafka-SpringBoot整合Kafka发送复杂对象相关推荐

  1. springboot 整合kafka 实现生产，消费数据

    一 kafka集群的启动 1.1 机器说明 192.168.152.128 master 192.168.152.129 slaver01 192.168.152.130 slaver02 1.2 查 ...

  2. Kafka精品教学（入门，安装，Springboot整合Kafka）

    ps:本文是博主结合视频和博客学习之后,自己实验总结编写的,如果侵权请联系删除. 要学习kafka首先要了解什么是消息队列,因为Kafka 是一个分布式的基于发布 / 订阅模式的消息队列(Messag ...

  3. SpringBoot整合kafka(实现producer和consumer)

    转载自 SpringBoot整合kafka(实现producer和consumer) 在Windows环境下安装运行Kafka:https://www.jianshu.com/p/d64798e81f ...

  4. springboot 整合 kafka demo 顺便看一下源码

    大家好,我是烤鸭: 今天分享下 springboot 整合 kafka. 1.  环境参数: windows + kafka_2.11-2.3.0 + zookeeper-3.5.6 + spring ...

  5. SpringBoot整合kafka之kafka分区实战

    本文来说下SpringBoot整合kafka之kafka分区实战 文章目录 准备工作 程序代码 程序测试 本文小结 准备工作 当然我们也可以不手动创建topic,在执行代码kafkaTemplate. ...

  6. Kafka原理以及SpringBoot整合Kafka

    1.Kafka原理 1. brokers有多个broker组成,broker是指Kafka服务器(192.168.223.140就是其中的一个broker),上面三台Kafka服务器组成了Kafka集 ...

  7. kafka 安装使用 /springboot整合kafka /消息投递机制以及存储策略 /副本处理机制

    一.背景 1.基本信息 Kafka是由Apache软件基金会开发的一个开源流处理平台,由Scala和Java编写.Kafka是一种高吞吐量的分布式发布订阅消息系统,它可以处理消费者在网站中的所有动作流 ...

  8. SpringBoot整合kafka实战之带回调的生产者

    本文来说下SpringBoot整合kafka部分知识内容 文章目录 带回调的生产者 方式一 方式二 本文小结 带回调的生产者 前面我们说了简单的生产和消费,本文说下带回调的生产者.kafkaTempl ...

  9. 玩转Kafka—SpringGo整合Kafka

    玩转Kafka-Spring整合Kafka 1 新建Spring Boot项目,增加依赖 <dependencies><dependency><groupId>or ...

  10. SpringBoot整合 ActiveMQ、SpringBoot整合RabbitMQ、SpringBoot整合Kafka

    1.概念:SpringBoot 整合消息服务2.具体内容对于异步消息组件在实际的应用之中会有两类:· JMS:代表作就是 ActiveMQ,但是其性能不高,因为其是用 java 程序实现的:· AMQ ...

最新文章

  1. 1.6的锁优化（适应性自旋/锁粗化/锁削除/轻量级锁/偏向锁）
  2. 会赚钱的人共有的特征，你是吗？
  3. JSP中Request属性范围
  4. 夏季晨练不得早于6点 - 生活至上，美容至尚！
  5. 赚钱只要找到方法，就如吸空气
  6. ASP.NET 2.0 Web Part编程之定制Web Part
  7. Java Web开发入门 - 第3章 Tomcat单元测试
  8. P4778 Counting swaps （组合数学，打表推通项公式，OEIS）
  9. cad批量打印_CAD批量打印PDF软件CAD转PDF自动选纸SmartBatchPlo
  10. Alice and the List of Presents CodeForces - 1236B 数学推导
  11. EasyUI项目驱动学习
  12. 【无标题】我们的三周年————致我的小宝贝
  13. 毫秒转化天小时分钟秒
  14. MOOS-ivp简介
  15. 四年级下册用计算机做的题,四年级下册数学用计算器计算练习题
  16. es查询-统计总数以及深度分页
  17. 如何快速记忆C语言运算符,C语言运算符快速记忆法
  18. nect计算机报名入口,教师资格考试如何报名？
  19. DSCTF2022 fuzzerinstrospector-Wp
  20. 解决开放平台的证书错误：ValidatorException：PKIX path validation failed

热门文章

  1. oracle 计划中的view,执行计划里的view
  2. android XML解析详解(封装好的工具类)
  3. PercentFrameLayout(百分比布局)的基本使用
  4. C++虚继承下的内存模型(一)
  5. 基于xgboost 的贷款风险预测
  6. 组态王怎么做超级曲线_鱼怎么做才好吃？试试这个方法，吃着过瘾，还超级下饭！（收藏）...
  7. 知识图谱(历史回顾及技术挑战)
  8. tensorflow就该这么学－－3
  9. MATLAB从入门到精通：MATLAB 图形操作
  10. 强化学习（二）马尔科夫决策过程(MDP)