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作为标准的数据传输格式。站在巨人的肩膀上，事半功倍。
SpringBoot整合kafka(实现producer和consumer)相关推荐
- springboot 整合 kafka demo 顺便看一下源码
大家好,我是烤鸭: 今天分享下 springboot 整合 kafka. 1. 环境参数: windows + kafka_2.11-2.3.0 + zookeeper-3.5.6 + spring ...
- springboot 整合kafka 实现生产，消费数据
一 kafka集群的启动 1.1 机器说明 192.168.152.128 master 192.168.152.129 slaver01 192.168.152.130 slaver02 1.2 查 ...
- Kafka精品教学（入门，安装，Springboot整合Kafka）
ps:本文是博主结合视频和博客学习之后,自己实验总结编写的,如果侵权请联系删除. 要学习kafka首先要了解什么是消息队列,因为Kafka 是一个分布式的基于发布 / 订阅模式的消息队列(Messag ...
- kafka 安装使用 /springboot整合kafka /消息投递机制以及存储策略 /副本处理机制
一.背景 1.基本信息 Kafka是由Apache软件基金会开发的一个开源流处理平台,由Scala和Java编写.Kafka是一种高吞吐量的分布式发布订阅消息系统,它可以处理消费者在网站中的所有动作流 ...
- Kafka原理以及SpringBoot整合Kafka
1.Kafka原理 1. brokers有多个broker组成,broker是指Kafka服务器(192.168.223.140就是其中的一个broker),上面三台Kafka服务器组成了Kafka集 ...
- SpringBoot整合kafka之kafka分区实战
本文来说下SpringBoot整合kafka之kafka分区实战 文章目录 准备工作 程序代码 程序测试 本文小结 准备工作 当然我们也可以不手动创建topic,在执行代码kafkaTemplate. ...
- SpringBoot整合kafka实战之带回调的生产者
本文来说下SpringBoot整合kafka部分知识内容 文章目录 带回调的生产者 方式一 方式二 本文小结 带回调的生产者 前面我们说了简单的生产和消费,本文说下带回调的生产者.kafkaTempl ...
- SpringBoot整合 ActiveMQ、SpringBoot整合RabbitMQ、SpringBoot整合Kafka
1.概念:SpringBoot 整合消息服务2.具体内容对于异步消息组件在实际的应用之中会有两类:· JMS:代表作就是 ActiveMQ,但是其性能不高,因为其是用 java 程序实现的:· AMQ ...
- java连接kafka api_Kafka-JavaAPI(Producer And Consumer)
Kafka--JAVA API(Producer和Consumer) Kafka 版本2.11-0.9.0.0 producer package com.yzy.spark.kafka; import ...
最新文章
- python装饰器教学_Python装饰器学习（九步入门）
- 烂泥：ubuntu下配置msmtp+mutt发送邮件
- GIS项目中数据开源、工具开源、开发开源的解决方案
- Codeforces Round #494 (Div. 3)【未完结】
- (转)jQuery 基础
- CSS改变插入光标颜色
- Android开发之自定义菊花进度条对话框
- Silverlight MMORPG网页游戏开发课程(Game Lesson)：目录
- python http 返回json中文乱码
- 一些简单的例子让你在Java中能更好的学习并理解循环结构（1）！
- 假如 C++ 是一只箭，你会用它来射哪只雕？
- 常用jQuery代码
- 2018-08-06-Python全栈开发day35-day36-select
- 顺序表（C语言实现）
- 谭永霞电路分析第三版课后答案_《电路分析》谭永霞西南交通大学课后习题和其答案.pdf...
- IGBT工作原理,解析IGBT工作原理及作用
- CSS：光标跟随实现
- window.performance（监控网页与程序性能）
- Android学习之动画（二）
- 谷歌浏览器调试工具使用进阶（二）