SpringBoot整合kafka(实现producer和consumer)

转载自 SpringBoot整合kafka(实现producer和consumer)

在Windows环境下安装运行Kafka：https://www.jianshu.com/p/d64798e81f3b

本文代码使用的是Spring Boot 2.1.1.RELEASE 版本

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

一、 pom.xml文件，引入依赖

<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><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>org.springframework.kafka</groupId><artifactId>spring-kafka-test</artifactId><scope>test</scope></dependency>
</dependencies>

采用Kafka提供的StringSerializer和StringDeserializer进行序列化和反序列化

1、在application-dev.properties配置生产者

#============== kafka ===================
# 指定kafka server的地址，集群配多个，中间，逗号隔开
spring.kafka.bootstrap-servers=127.0.0.1:9092#=============== provider  =======================
# 写入失败时，重试次数。当leader节点失效，一个repli节点会替代成为leader节点，此时可能出现写入失败，
# 当retris为0时，produce不会重复。retirs重发，此时repli节点完全成为leader节点，不会产生消息丢失。
spring.kafka.producer.retries=0
# 每次批量发送消息的数量,produce积累到一定数据，一次发送
spring.kafka.producer.batch-size=16384
# produce积累数据一次发送，缓存大小达到buffer.memory就发送数据
spring.kafka.producer.buffer-memory=33554432#procedure要求leader在考虑完成请求之前收到的确认数，用于控制发送记录在服务端的持久化，其值可以为如下：
#acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为-1。
#acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
#acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1的设置。
#可以设置的值为：all, -1, 0, 1
spring.kafka.producer.acks=1# 指定消息key和消息体的编解码方式
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

  • bootstrap.servers:kafka server的地址

  • acks:写入kafka时，leader负责一个该partion读写，当写入partition时，需要将记录同步到repli节点，all是全部同步节点都返回成功，leader才返回ack。

  • retris:写入失败时，重试次数。当leader节点失效，一个repli节点会替代成为leader节点，此时可能出现写入失败，当retris为0时，produce不会重复。retirs重发，此时repli节点完全成为leader节点，不会产生消息丢失。

  • batch.size:produce积累到一定数据，一次发送。

buffer.memory: produce积累数据一次发送，缓存大小达到buffer.memory就发送数据。

  • linger.ms :当设置了缓冲区，消息就不会即时发送，如果消息总不够条数、或者消息不够buffer大小就不发送了吗？当消息超过linger时间，也会发送。

  • key/value serializer：序列化类。

2、生产者向kafka发送消息

@RestController
public class KafkaController {@Autowiredprivate KafkaTemplate<String,Object> kafkaTemplate;@GetMapping("/message/send")public boolean send(@RequestParam String message){kafkaTemplate.send("testTopic",message);return true;}}

3、在application-dev.properties配置消费者

#=============== consumer  =======================
# 指定默认消费者group id --> 由于在kafka中，同一组中的consumer不会读取到同一个消息，依靠groud.id设置组名
spring.kafka.consumer.group-id=testGroup
# smallest和largest才有效，如果smallest重新0开始读取，如果是largest从logfile的offset读取。一般情况下我们都是设置smallest
spring.kafka.consumer.auto-offset-reset=earliest
# enable.auto.commit:true --> 设置自动提交offset
spring.kafka.consumer.enable-auto-commit=true
#如果'enable.auto.commit'为true，则消费者偏移自动提交给Kafka的频率（以毫秒为单位），默认值为5000。
spring.kafka.consumer.auto-commit-interval=100# 指定消息key和消息体的编解码方式
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

  • Producer是一个接口，声明了同步send和异步send两个重要方法。

  • ProducerRecord 消息实体类，每条消息由（topic,key,value,timestamp)四元组封装。一条消息key可以为空和timestamp可以设置当前时间为默认值。

4、消费者监听topic=testTopic的消息

@Component
public class ConsumerListener {@KafkaListener(topics = "testTopic")public void onMessage(String message){//insertIntoDb(buffer);//这里为插入数据库代码System.out.println(message);}}

到此，采用Kafka提供的StringSerializer和StringDeserializer进行序列化和反序列化，因为此种序列化方式无法序列化实体类，顾，下面为自定义序列化和反序列化器进行实体类的消息传递

采用自定义序列化和反序列化器进行实体类的序列化和反序列化

和内置的StringSerializer字符串序列化一样，如果要自定义序列化方式，需要实现接口Serializer。假设每个字段按照下图所示的方式自定义序列化：

 

1、创建User实体类

public class User implements Serializable {private Long id;private String name;private Integer age;/*** transient 关键字修饰的字段不会被序列化*/private transient String desc;public Long getId() {return id;}public void setId(Long id) {this.id = id;}public String getName() {return name;}public void setName(String name) {this.name = name;}public Integer getAge() {return age;}public void setAge(Integer age) {this.age = age;}public String getDesc() {return desc;}public void setDesc(String desc) {this.desc = desc;}@Overridepublic String toString() {return "User{" +"id=" + id +", name='" + name + '\'' +", age=" + age +", desc='" + desc + '\'' +'}';}
}

2、创建User序列化器

public class UserSerializable implements Serializer<User> {@Overridepublic void configure(Map<String, ?> map, boolean b) {}@Overridepublic byte[] serialize(String topic, User user) {System.out.println("topic : " + topic + ", user : " + user);byte[] dataArray = null;ByteArrayOutputStream outputStream = null;ObjectOutputStream objectOutputStream = null;try {outputStream = new ByteArrayOutputStream();objectOutputStream = new ObjectOutputStream(outputStream);objectOutputStream.writeObject(user);dataArray = outputStream.toByteArray();} catch (Exception e) {throw new RuntimeException(e);}finally {if(outputStream != null){try {outputStream.close();} catch (IOException e) {e.printStackTrace();}}if(objectOutputStream != null){try {objectOutputStream.close();} catch (IOException e) {e.printStackTrace();}}}return dataArray;}@Overridepublic void close() {}
}

3、创建User反序列化器

public class UserDeserializer implements Deserializer<User> {@Overridepublic void configure(Map<String, ?> map, boolean b) {}@Overridepublic User deserialize(String topic, byte[] bytes) {User user = null;ByteArrayInputStream inputStream = null;ObjectInputStream objectInputStream = null;try {inputStream = new ByteArrayInputStream(bytes);objectInputStream = new ObjectInputStream(inputStream);user = (User)objectInputStream.readObject();} catch (Exception e) {throw new RuntimeException(e);}finally {if(inputStream != null){try {inputStream.close();} catch (IOException e) {e.printStackTrace();}}if(objectInputStream != null){try {objectInputStream.close();} catch (IOException e) {e.printStackTrace();}}}return user;}@Overridepublic void close() {}
}

4、修改application-dev.properties配置

A、修改生产者配置的value-serializer

# 指定生产者消息key和消息体的编解码方式
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=com.yibo.springbootkafkademo.Serializable.UserSerializable

B、修改消费者配置的value-deserializer

# 指定消费者消息key和消息体的编解码方式
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=com.yibo.springbootkafkademo.Serializable.UserDeserializer

5、生产者向kafka发送消息

@RestController
public class KafkaController {@Autowiredprivate KafkaTemplate<String,Object> kafkaTemplate;@PostMapping("/user/save")public boolean saveUser(@RequestBody User user){kafkaTemplate.send("userTopic",user);return true;}
}

6、消费者监听topic=userTopic的消息

@Component
public class ConsumerListener {@KafkaListener(topics = "userTopic")public void onMessage(User user){//insertIntoDb(buffer);//这里为插入数据库代码System.out.println(user);}
}

总结

可以看到，自定义Serializer和Deserializer非常痛苦，还有很多类型不支持，非常脆弱。复杂类型的支持更是一件痛苦的事情，不同版本之间的兼容性问题更是一个极大的挑战。由于Serializer和Deserializer影响到上下游系统，导致牵一发而动全身。自定义序列化&反序列化实现不是能力的体现，而是逗比的体现。所以强烈不建议自定义实现序列化&反序列化，推荐直接使用StringSerializer和StringDeserializer，然后使用json作为标准的数据传输格式。站在巨人的肩膀上，事半功倍。


 

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/322780.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

ASP.NET Core 认证与授权[7]:动态授权

ASP.NET Core 认证与授权[7]:动态授权

基于资源的授权 有些场景下&#xff0c;授权需要依赖于要访问的资源&#xff0c;例如&#xff1a;每个资源通常会有一个创建者属性&#xff0c;我们只允许该资源的创建者才可以对其进行编辑&#xff0c;删除等操作&#xff0c;这就无法通过[Authorize]特性来指定授权了。因为授…
阅读更多...
H5的Websocket基本使用

H5的Websocket基本使用

前端代码 <!DOCTYPE html> <html lang"en"> <head><meta charset"UTF-8"><meta name"viewport" content"widthdevice-width, initial-scale1.0"><title>Document</title> </head> &…
阅读更多...
springboot手动提交kafka offset

springboot手动提交kafka offset

转载自 springboot手动提交kafka offset enable.auto.commit参数设置成了false 但是测试发现enable.auto.commit参数设置成了false&#xff0c;kafka的offset依然提交了&#xff08;也没有进行人工提交offset&#xff09;。 查看源码 如果我们enable.auto.commit设置为false…
阅读更多...
可观测性与原生云监控

可观测性与原生云监控

在近日发表的一篇文章中&#xff0c;Cindy Sridharan概括介绍了可观测性及其与原生云应用程序监控的关系。可观测性是一种理念&#xff0c;包括监控、日志聚合、指标和分布式跟踪&#xff0c;可以实时更深入地观察系统。 Sridharan的文章基于她就同一个主题所做的Velocity演讲。…
阅读更多...
使用Microsoft.AspNetCore.TestHost进行完整的功能测试

使用Microsoft.AspNetCore.TestHost进行完整的功能测试

简介 Microsoft.AspNetCore.TestHost是可以用于Asp.net Core 的功能测试工具。很多时候我们一个接口写好了&#xff0c;单元测试什么的也都ok了&#xff0c;需要完整调试一下&#xff0c;检查下单元测试未覆盖到的代码是否有bug。步骤为如下&#xff1a;程序打个断点->F5运行…
阅读更多...
MockJs案例

MockJs案例

有时候前端写好模板后&#xff0c;后端还完工&#xff0c;那么总不能一直让项目停滞吧&#xff0c;这里就用Mockjs来模拟后端接口的数据&#xff0c;让我们先人一步完成项目。 首先创建一个html&#xff0c;导入axios和mockjs 再用mock去拦截请求&#xff0c;如果后端接口写好了…
阅读更多...
Entity Framework Core 使用HiLo生成主键

Entity Framework Core 使用HiLo生成主键

HiLo是在NHibernate中生成主键的一种方式&#xff0c;不过现在我们可以在Entity Framework Core中使用。所以在这篇内容中&#xff0c;我将向您在介绍如何在Entity Framework Core中使用HiLo生成主键。 什么是Hilo&#xff1f; HiLo是High Low的简写&#xff0c;翻译成中文叫高…
阅读更多...
Echarts报错：Component series.lines not exists. Load it first.

Echarts报错：Component series.lines not exists. Load it first.

前几天用的echarts标签是bootcdn的 <script src"https://cdn.bootcdn.net/ajax/libs/echarts/4.7.0/echarts-en.common.js"></script>用着官方给的案例还可以&#xff0c;但是一用gallery社区里面的例子就报错 后来经过不断调试终于知道是需要换个cdn&…
阅读更多...
认识微软Visual Studio Tools for AI

认识微软Visual Studio Tools for AI

微软已经发布了其 Visual Studio Tools for AI 的测试版本&#xff0c;这是微软 Visual Studio 2017 IDE 的扩展&#xff0c;可以让开发人员和数据科学家将深度学习模型嵌入到应用程序中。Visual Studio Tools for AI 工具同时支持 Microsoft 的 Cognitive Toolkit 和 Google 的…
阅读更多...
mybatis源码阅读(一)：SqlSession和SqlSessionFactory

mybatis源码阅读(一)：SqlSession和SqlSessionFactory

转载自 mybatis源码阅读(一)&#xff1a;SqlSession和SqlSessionFactory 一、接口定义 听名字就知道这里使用了工厂方法模式&#xff0c;SqlSessionFactory负责创建SqlSession对象。其中开发人员最常用的就是DefaultSqlSession &#xff08;1&#xff09;SqlSession接口定义…
阅读更多...
开源纯C#工控网关+组态软件(六)图元组件

开源纯C#工控网关+组态软件(六)图元组件

一、 图元概述 图元是构成人机界面的基本单元。如一个个的电机、设备、数据显示、仪表盘&#xff0c;都是图元。构建人机界面的过程就是铺排、挪移、定位图元的过程。 图元设计是绘图和编码的结合。因为图元不仅有显示和动画&#xff0c;还有背后操纵动画的控制逻辑。 一个好…
阅读更多...
git合并分支的策略（赞）

git合并分支的策略（赞）

假设当前有两个分支 master和test&#xff0c;两个分支一模一样&#xff0c;都有这三个文件 现在test添加一个test4.txt&#xff0c;然后提交到本地&#xff08;git add . git commit&#xff09; 切换到master分支上&#xff0c;git checkout master git merge test 这样mase…
阅读更多...
改造独立部署(SCD)模式下.NET Core应用程序 dotnet的exe文件启动过程

改造独立部署(SCD)模式下.NET Core应用程序 dotnet的exe文件启动过程

设置一个小目标 改造前 改造后 独立部署SCD模式&#xff0c;是指在使用dotnet publish 命令时带上-r 参数运行时标识符&#xff08;RID&#xff09;。 目标提出原因&#xff1a;SCD模式下文件太乱了&#xff0c;很多文件在开发时大多又涉及不到&#xff0c;发布后如果能把文件…
阅读更多...
mybatis源码阅读(四)：mapper(dao)实例化

mybatis源码阅读(四)：mapper(dao)实例化

转载自 mybatis源码阅读(四)&#xff1a;mapper(dao)实例化 在开始分析之前&#xff0c;先来了解一下这个模块中的核心组件之间的关系&#xff0c;如图&#xff1a; 1.MapperRegistry&MapperProxyFactory MapperRegistry是Mapper接口及其对应的代理对象工程的注册中心&…
阅读更多...
自定义路由匹配和生成

自定义路由匹配和生成

前言 前两篇文章主要总结了CMS系统两个技术点在ASP.NET Core中的应用&#xff1a; 《ASP.NET Core 中的SEO优化&#xff08;1&#xff09;&#xff1a;中间件实现服务端静态化缓存》 《ASP.NET Core 中的SEO优化&#xff08;2&#xff09;&#xff1a;中间件中渲染Razor视图》…
阅读更多...
如何封装并发布一个属于自己的ui组件库

如何封装并发布一个属于自己的ui组件库

以前就一直有个想法自己能不能封装一个类似于elementui一样的组件库&#xff0c;然后发布到npm上去&#xff0c;毕竟前端说白了&#xff0c;将组件v上去&#xff0c;然后进行数据交互。借助这次端午&#xff0c;终于有机会&#xff0c;尝试自己去封装发布组件库了 我这里了只做…
阅读更多...
听云支持.NET Core的应用性能监控

听云支持.NET Core的应用性能监控

随着微软于2017年8月正式发布.NET Core 2.0&#xff0c; .NET Core 社区开始活跃&#xff0c;众多.NET开发者开始向跨平台转变。 听云于2017年11月推出了.NET Core应用监控工具&#xff0c;和听云其他语言的监控工具一样&#xff0c;.NET Core应用监控工具具有以下特征&#xf…
阅读更多...
mybatis源码阅读(五) ---执行器Executor

mybatis源码阅读(五) ---执行器Executor

转载自 mybatis源码阅读(五) ---执行器Executor 1. Executor接口设计与类结构图 public interface Executor {ResultHandler NO_RESULT_HANDLER null;// 执行update&#xff0c;delete&#xff0c;insert三种类型的sql语句int update(MappedStatement ms, Object parameter…
阅读更多...
[52ABP实战系列] .NET CORE实战入门第三章更新了

[52ABP实战系列] .NET CORE实战入门第三章更新了

早安 各位道友好&#xff0c;.NET CORE入门视频的第三章也算录制完毕了。欢迎大家上传课网进行学习。 更新速度 大家也知道最近的社会新闻比较多。频繁发生404、关键字打不出来&#xff0c;我个人也在关注这些事件。导致精力分散&#xff0c;没有做到稳定更新&#xff0c;现在呢…
阅读更多...
如何安装nuxt

如何安装nuxt

因为vue是单页面应用&#xff0c;所以不被Seo&#xff0c;如百度和Google抓取到&#xff0c;在Vue中如果想要爬虫爬到就必须使用nuxt 那么如何安装使用呢&#xff1f; yarn create nuxt-app <project-name> cd <project-name> yarn build yarn start必须先build&a…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023