SpringBoot集成Kafka

36271 软件 | 2022-6-20

Kafka 是一个由 LinkedIn 开发的分布式消息系统，它于2011年年初开源，现在由著名的 Apache 基金会维护与开发。 Kafka 使用 Scala 实现，被用作 LinkedIn 的活动流和运营数据处理的管道，现在也被诸多互联网企业广泛地用作数据流管道和消息系统， Kafka 是基于消息发布﹣订阅模式实现的消息系统。

之前为了测试方便，已经说了如何在Windows上安装Kafka

Kafka安装Windows版 - Java小强技术博客 (javacui.com)

安装之后，我们使用SpringBoot开发一个简单消息发送和接收示例

首先是POM引入

<dependency>
	<groupId>org.springframework.kafka</groupId>
	<artifactId>spring-kafka</artifactId>
	<version>2.8.6</version>
</dependency>
<dependency>
	<groupId>com.fasterxml.jackson.core</groupId>
	<artifactId>jackson-databind</artifactId>
	<version>2.13.2.2</version>
</dependency>

 

然后编辑SpringBoot配置文件

spring:
  main:
    # 允许循环依赖，高版本SpringBoot禁用了该选项
    allow-circular-references: true
    # 设置是否允许具有相同名称的Bean来覆盖之前的Bean
    allow-bean-definition-overriding: true
  jackson:
    default-property-inclusion: non_null # JSON处理时忽略非空字段
  # kafka 配置
  kafka:
    consumer:
      enable-auto-commit: true # 是否自动提交offset
      auto-commit-interval: 100  # 提交offset延时(接收到消息后多久提交offset)
      # earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
      # latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
      # none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
      auto-offset-reset: latest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      # 表示接受反序列化任意的类，也可限定包路径
      properties:
        spring:
          json:
            trusted:
              packages: '*'
    producer:
      retries: 0 # 重试次数
      # 0：producer不等待broker的ack，broker一接收到还没有写入磁盘就已经返回，当broker故障时有可能丢失数据；
      # 1：producer等待broker的ack，partition的leader落盘成功后返回ack，如果在follower同步成功之前leader故障，那么将会丢失数据；（数据要求快，重要性不高时用）
      # -1：producer等待broker的ack，partition的leader和follower全部落盘成功后才返回ack，数据一般不会丢失，延迟时间长但是可靠性高。（数据重要时用）
      acks: 1 # 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    bootstrap-servers: 127.0.0.1:9092

特别注意的是要指定序列化信任的类，否则传递的消息是对象时会报错。

可以看到对于值的序列化，使用的是Json，这是因为我们将使用对象来传递消息内容。

开发一个Controller开发送消息，那么发送消息将变得非常简单，只需要注入一个KafkaTemplate对象即可。

package com.example.springboot.controller;
import cn.hutool.core.date.DateUtil;
import com.example.springboot.model.Blog;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Date;
@RestController
@RequestMapping("/kafka")
public class MessSendController {
    @Autowired
    private KafkaTemplate kafkaTemplate;
    private static final String messTopic = "test";
    @RequestMapping("/send")
    public String sendMess(){
        Blog blog = Blog.builder().id(1).name("测试").isDel(false).birthday(new Date()).build();
        kafkaTemplate.send(messTopic, blog);
        System.out.println("客户端 消息发送完成");
        return DateUtil.now();
    }
}

发送消息的主题就是test，编写消息接收端

package com.example.springboot.listener;
import com.alibaba.fastjson.JSON;
import com.example.springboot.model.Blog;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
public class KafkaMessListener {
    private static final String messTopic = "test";
    @KafkaListener(topics = messTopic, groupId = "javagroup")
    public void messListener(Blog blog){
        System.out.println("消费端 收到消息：" + JSON.toJSONString(blog));
    }
}

启动程序，控制台会打印一下对于Kafka的基础配置参数

ConsumerConfig values: 

allow.auto.create.topics = true
auto.commit.interval.ms = 100
auto.offset.reset = latest
bootstrap.servers = [127.0.0.1:9092]
check.crcs = true
client.dns.lookup = use_all_dns_ips
client.id = consumer-javagroup-1
client.rack = 
connections.max.idle.ms = 540000
default.api.timeout.ms = 60000
enable.auto.commit = true
exclude.internal.topics = true
fetch.max.bytes = 52428800
fetch.max.wait.ms = 500
fetch.min.bytes = 1
group.id = javagroup
group.instance.id = null
heartbeat.interval.ms = 3000
interceptor.classes = []
internal.leave.group.on.close = true
internal.throw.on.fetch.stable.offset.unsupported = false
isolation.level = read_uncommitted
key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
max.partition.fetch.bytes = 1048576
max.poll.interval.ms = 300000
max.poll.records = 500
metadata.max.age.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.recording.level = INFO
metrics.sample.window.ms = 30000
partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor, class org.apache.kafka.clients.consumer.CooperativeStickyAssignor]
receive.buffer.bytes = 65536
reconnect.backoff.max.ms = 1000
reconnect.backoff.ms = 50
request.timeout.ms = 30000
retry.backoff.ms = 100
sasl.client.callback.handler.class = null
sasl.jaas.config = null
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.min.time.before.relogin = 60000
sasl.kerberos.service.name = null
sasl.kerberos.ticket.renew.jitter = 0.05
sasl.kerberos.ticket.renew.window.factor = 0.8
sasl.login.callback.handler.class = null
sasl.login.class = null
sasl.login.refresh.buffer.seconds = 300
sasl.login.refresh.min.period.seconds = 60
sasl.login.refresh.window.factor = 0.8
sasl.login.refresh.window.jitter = 0.05
sasl.mechanism = GSSAPI
security.protocol = PLAINTEXT
security.providers = null
send.buffer.bytes = 131072
session.timeout.ms = 45000
socket.connection.setup.timeout.max.ms = 30000
socket.connection.setup.timeout.ms = 10000
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2]
ssl.endpoint.identification.algorithm = https
ssl.engine.factory.class = null
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.certificate.chain = null
ssl.keystore.key = null
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLSv1.2
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.certificates = null
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
value.deserializer = class org.springframework.kafka.support.serializer.JsonDeserializer

可以看到，默认是允许自动创建主题的，这是本地线上的主题都是指定好的。

然后序列化已经是我们指定的组件了。

启动调用Controller就可以发送一个消息，由于在一个程序内，发送后会立马打印该消息已经收到。

END

推荐您阅读更多有关于“ 开源 springboot Kafka 消息队列 ”的文章

上一篇：SpringBoot集成Kafka之SASL_SSL 下一篇：Elasticsearch配置IK分词器-扩展词库和热更新

猜你喜欢

• 6-20HTTP POST数据和数据接收
• 6-20常用SpringJPA主键生成策略，UUID和自增
• 6-20Apache Commons Jar包下载
• 6-20IIS服务器返回Bad Request (Invalid Hostname)解决办法
• 6-20SpringJPA 慎重 findAll() 切记分页

评论：