SpringBoot开发案例之整合Kafka实现消息队列
Posted go1188
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了SpringBoot开发案例之整合Kafka实现消息队列相关的知识,希望对你有一定的参考价值。
本文代码使用的是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 {
@Autowired
private 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;
}
@Override
public String toString() {
return "User{" +
"id=" + id +
", name=‘" + name + ‘‘‘ +
", age=" + age +
", desc=‘" + desc + ‘‘‘ +
‘}‘;
}
}
2、创建User序列化器
public class UserSerializable implements Serializer<User> {
@Override
public void configure(Map<String, ?> map, boolean b) {
}
@Override
public 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;
}
@Override
public void close() {
}
}
3、创建User反序列化器
public class UserDeserializer implements Deserializer<User> {
@Override
public void configure(Map<String, ?> map, boolean b) {
}
@Override
public 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;
}
@Override
public 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 {
@Autowired
private 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实现消息队列的主要内容,如果未能解决你的问题,请参考以下文章