目录本章主要是讲解kafka基于SpringBoot集成的内容
包括一些在开发中用到的一些生产者和消费者特性
有需要的可以慢慢食用。
- Kafka之SpringBoot集成
- 背景
- 基本组成
- 主题(topic)
- 分区(partition)
- 副本(Replica)
- Broker
- 消息位移(offset)
- 消费者组(Consumers)
- topic推给消费者的规则
- 生产者组(Producers)
- 环境要求
- SpringBoot集成
- 基本环境
- maven：
- yml配置
- 生产者特性
- 异步发送消息
- 同步发送消息
- 指定分区发送
- 自定义分区规则
- 生产者幂等性
- 消费者特性
- 手动提交偏移量
- 多线程消费
- 指定分区消费
基本组成 主题(topic)Kafka最早是由LinkedIn公司开发的，作为其自身业务消息处理的基础，后LinkedIn公司将Kafka捐赠给Apache，现在已经成为Apache的一个顶级项目了，Kafka作为一个高吞吐的分布式的消息系统，目前已经被很多公司应用在实际的业务中了，并且与许多数据处理框架相结合，比如Hadoop，Spark等。
分区(partition)顾名思义Topics是一些主题的集合，更通俗的说Topic就像一个消息队列，生产者可以向其写入消息，消费者可以从中读取消息，一个Topic支持多个生产者或消费者同时订阅它，所以其扩展性很好。Topic又可以由一个或多个partition（分区）组成，比如下图：
副本(Replica)其中每个partition中的消息是有序的，但相互之间的顺序就不能保证了，若Topic有多个partition，生产者的消息可以指定或者由系统根据算法分配到指定分区，若你需要所有消息都是有序的，那么你最好只用一个分区。另外partition支持消息位移读取，消息位移有消费者自身管理，比如下图：
Broker为保证集群中的某个节点发生故障时，该节点上的partition数据不丢失，且kafka仍然能够继续工作，kafka提供了副本机制，一个topic的每个分区都有若干个副本，一个leader和若干个follower。
leader：每个分区多个副本的“主”，生产者发送数据的对象，以及消费者消费数据的对象都是leader
follower：每个分区多个副本中的“从”，实时从leader中同步数据，保持和leader数据的同步。leader发生故障时，某个follower会成为新的follower
在kafka中，每个主题可以有多个分区，每个分区又可以有多个副本。这多个副本中，只有一个是leader，而其他的都是follower副本。仅有leader副本可以对外提供服务
消息位移(offset)一台kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topic
消费者组(Consumers)Kafka就是用offset来表示消费者的消费进度到哪了，每个消费者会都有自己的offset。说白了offset就是表示消费者的消费进度。
在以前版本的Kafka，这个offset是由Zookeeper来管理的，后来Kafka开发者认为Zookeeper不合适大量的删改操作，于是把offset在broker以内部topic(__consumer_offsets)的方式来保存起来。
每次消费者消费的时候，都会提交这个offset，Kafka可以让你选择是自动提交还是手动提交。
topic推给消费者的规则Consumers是一群消费者的集合，可以称之为消费者组，是一种更高层次的的抽象，向Topic订阅消费消息的单位是Consumers，当然它其中也可以只有一个消费者（consumer）。
下面是关于consumer的两条原则：
假如所有消费者都在同一个消费者组中，那么它们将协同消费订阅Topic的部分消息（根据分区与消费者的数量分配），保存负载平衡；
假如所有消费者都在不同的消费者组中，并且订阅了同个Topic，那么它们将可以消费Topic的所有消息；
生产者组(Producers)若消费者数小于partition数，且消费者数为一个，那么它就消费所有消息；
若消费者数小于partition数，假设消费者数为N，partition数为M，那么每个消费者能消费的分区数为M/N或M/N+1；
若消费者数等于partition数，那么每个消费者都会均等分配到一个分区的消息；
若消费者数大于partition数，则将会出现部分消费者得不到消息分区，出现空闲
环境要求Producers作为消息的生产者，可以自己指定将消息发布到订阅Topic中的指定分区，策略可以自己指定，比如语义或者结构类似的消息发布在同一分区，当然也可以由系统循环发布在每一个分区上。
搭建一个kafka集群，3节点，kafka版本为2.5.0。
部署不是本章的内容
可视化工具：cmak（jdk11）
SpringBoot集成 基本环境springboot：2.2.6.RELEASE
maven：yml配置org.springframework.kafka spring-kafkaorg.projectlombok lombok1.18.12
spring:
kafka:
bootstrap-servers: ip:port
consumer:
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer #消费者键序列化方式
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer #消费者value序列化方式
group-id: test-group-id #组id
auto-offset-reset: earliest #offset重启设置 有三个属性 earliest latest none
enable-auto-commit: false #是否自动提交偏移量
producer:
retries: 3 #重试次数
batch-size: 16384 # 批量提交数据大小
buffer-memory: 33554432 #缓存大小
key-serializer: org.apache.kafka.common.serialization.StringSerializer #字符串序列化
value-serializer: org.apache.kafka.common.serialization.StringSerializer #字符串序列化
acks: all
properties:
enable.idempotence: true #开启幂等性 幂等性必须重试有次数
#partitioner.class: com.yy.config.CustomizePartitioner #自定义分区策略
listener:
# 在侦听器容器中运行的线程数。
concurrency: 5
#listner负责ack，每调用一次，就立即commit
ack-mode: manual_immediate
missing-topics-fatal: false # 消费监听接口监听的主题不存在时，默认会报错。所以通过设置为 false ，解决报错
生产者特性
异步发送消息
异步发送消息主要通过对回调来判断消息是否发送成功
public void send(Object obj) {
String obj2String = JSONObject.toJSONString(obj);
log.info("准备发送消息为：{}", obj2String);
//发送消息
ListenableFuture> future = kafkaTemplate.send(TOPIC_TEST, obj);
future.addCallback(new ListenableFutureCallback>() {
@Override
public void onFailure(Throwable throwable) {
//发送失败的处理
log.info(TOPIC_TEST + " - 生产者 发送消息失败：" + throwable.getMessage());
}
@Override
public void onSuccess(SendResult stringObjectSendResult) {
int partition = stringObjectSendResult.getRecordMetadata().partition();
log.info("partition==========:"+partition);
//成功的处理
log.info(TOPIC_TEST + " - 生产者 发送消息成功：" + stringObjectSendResult.toString());
}
});
}
同步发送消息
同步发送消息主要用get()方法来阻塞发送，实现同步。
public void sendSync(Object obj) {
String obj2String = JSONObject.toJSONString(obj);
log.info("准备发送消息为：{}", obj2String);
//发送消息
SendResult stringObjectSendResult = null;
try {
stringObjectSendResult = kafkaTemplate.send(TOPIC_TEST, obj).get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
log.info(stringObjectSendResult.toString());
}
指定分区发送
主要是使用：
ListenableFuture
partion即指明发送给哪个分区，当然key可以换算成某个分区，具体规则如下：
1.若发送消息时指定了分区（即自定义分区策略），则直接将消息append到指定分区；
2.若发送消息时未指定patition，但指定了key（kafka允许为每条消息设置一个key），则对key值进行hash计算，根据计算结果路由到指定分区，这种情况下可以保证同一个Key的所有消息都进入到相同的分区；
3.patition 和 key 都未指定，则使用kafka默认的分区策略，轮询选出一个 patition。
public void sendToDirctpartition(Object obj,Integer partition) {
String obj2String = JSONObject.toJSONString(obj);
log.info("准备发送消息为：{}", obj2String);
//发送消息
String key ="test";
ListenableFuture> future = kafkaTemplate.send(TOPIC_TEST,partition,key,obj);
future.addCallback(new ListenableFutureCallback>() {
@Override
public void onFailure(Throwable throwable) {
//发送失败的处理
log.info(TOPIC_TEST + " - 生产者 发送消息失败：" + throwable.getMessage());
}
@Override
public void onSuccess(SendResult stringObjectSendResult) {
int partition = stringObjectSendResult.getRecordMetadata().partition();
log.info("partition==========:"+partition);
//成功的处理
log.info(TOPIC_TEST + " - 生产者 发送消息成功：" + stringObjectSendResult.toString());
}
});
}
自定义分区规则
在上面的指定分区发送消息提到了分区规则，那用户是否可以自定义分区规则呢？是可以的，通过实现Partitioner接口可以配置用户自己的分区规则
@Component
public class CustomizePartitioner implements Partitioner {
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
// 获取topic的分区列表
List partitionInfoList = cluster.availablePartitionsForTopic(topic);
int partitionCount = partitionInfoList.size();
int auditPartition = partitionCount - 1;
return auditPartition;
}
@Override
public void close() {
System.out.println("colse ...");
}
@Override
public void configure(Map configs) {
System.out.println("init ...");
}
}
生产者幂等性
何为幂等性？
所谓的幂等性就是指Producer不论向server发送多少次重复数据，Server端都只会持久化一条
需要注意：
1.kafka的版本大于0.11.0
2.生成者开启尝试次数
properties:
enable.idempotence: true #开启幂等性 幂等性必须重试有次数
消费者特性
手动提交偏移量
- 设置消费者自动提交偏移量为false
- MANUAL_IMMEDIATE: listner负责ack，每调用一次，就立即commit
spring:
kafka:
consumer:
enable-auto-commit: false
listener:
ack-mode: manual_immediate
消费端在消费完信息后记得提交偏移量
ack.acknowledge()多线程消费
- 在某个消费端指明当前的线程数
@KafkaListener注解加上concurrency 属性配置线程数
@KafkaListener(topics = KafkaProvider.TOPIC_TEST,groupId = KafkaProvider.TOPIC_GROUP,concurrency = "5")
public void topic_test(ConsumerRecord record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, Consumer consumer) {
Optional message = Optional.ofNullable(record.value());
if (message.isPresent()) {
Object msg = message.get();
log.info("topic_test 消费了： Topic:" + topic + ",Message:" + msg);
ack.acknowledge();
consumer.commitAsync();
}
}
- 在yml配置直接配置全局的线程数
spring:
kafka:
listener:
# 在侦听器容器中运行的线程数。
concurrency: 5
指定分区消费
@KafkaListener(topicPartitions = {@TopicPartition(topic =KafkaProvider.TOPIC_TEST,partitions = {"0"})},groupId = KafkaProvider.TOPIC_GROUP1)
public void topic_test3(ConsumerRecord record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
Optional message = Optional.ofNullable(record.value());
if (message.isPresent()) {
Object msg = message.get();
int partition = record.partition();
log.info("topic_test3 消费了： Topic:" + topic + ",Message:" + msg+",partition:"+partition);
ack.acknowledge();
}
}
