关于java:springbootroute十四整合Kafka

在上一章中SpringBoot整合RabbitMQ，曾经具体介绍了音讯队列的作用，这一种咱们间接来学习SpringBoot如何整合kafka发送音讯。

kafka简介

kafka是用Scala和Java语言开发的，高吞吐量的分布式消息中间件。高吞吐量使它在大数据畛域具备人造的劣势，被宽泛用来记录日志。

kafka架构剖析

注1：图中的红色箭头示意音讯的流动过程，蓝色示意分区备份，绿色示意kafka集群注册到zookeeper。

注2：在kafka0.9版本之前，消费者生产音讯的地位记录在zookeeper中，在0.9版本之后，生产音讯的地位记录在kafka的一个topic上。

kafka名词简介：

Producer：音讯生产者
Consumer：音讯消费者
Consumer Group（CG）：消费者组，一个topic能够有多个CG，每个Partition只会把音讯发送给GG中的一个Consumer
Broker：一台kafka服务器就是一个broker，一个broker有多个topic
Topic：音讯主题，音讯分类，可看作队列
Partition：分区，为了实现扩大，一个大的topic可能散布到多个broker上，一个topic能够分为多个partition，partition中的每条音讯都会被调配一个有序的id（offset），每个partiton中的音讯是有序的。
Offset：kafka的存储文件都是依照offset.kafka来命名的，不便查找，第一个offset为0000000000.kafka。
Leader：分区具备被备份，主分区
Follower：从分区

1. 生产者分区策略

指定分区。
没有指定分区但有key值，将key的hash值与以后topic的分区个数进行取余失去分区。
如果既没有指定分区又没有指定key，第一次调用时随机生成一个整数（当前调用每次在这个整数上自增），将这个随机数与该topic的分区数取余失去分区。

2. 音讯可靠性问题

采纳ack确认机制来保障音讯的可靠性。

kafka在发送音讯后会同步到其余分区正本，等所有正本都接管到音讯后，kafka才会发送ack进行确认。采纳这种模式的劣势就是当其中一个正本宕机后，则音讯生产者就不会收到kafka的ack。

kafka采纳ISR来解决这个问题。

ISR：Leader保护的一个和leader放弃同步的follower汇合。

当ISR中的folower实现数据同步之后，leader就会向follower发送ack，如果follower长时间未向leader同步数据，则该follower就会被踢出ISR，该工夫阀值的设置参数为replica.lag.time.max.ms，默认工夫为10s，leader产生故障后，就会从ISR中选举新的leader。

注：本文所讲的kafka版本为0.11，在0.9版本以前成为ISR还有一个条件，就是同步音讯的条数。

ack参数配置

0：生产者不期待broker的ack。

1：leader分区接管到音讯向生产者发送ack。

-1（all）：ISR中的leader和follower同步胜利后，向生产者发送ack。

3. 音讯一致性问题

如果leader中有10条音讯，向两个follower同步数据，follower A同步了8条，follower B同步了9条。这时候leader宕机了，follower A和follower B中的音讯是不统一的，剩下两个follower就会从新选举出一个leader。

LEO（log end offset）：每个正本的最初一个offset
HW（high watermark）：所有正本中最小的offset

为了保证数据的一致性，所有的follower会将各自的log文件高出HW的局部截掉，而后再从新的leader中同步数据。

4. 音讯重复性问题

在kafka0.11版本中引入了一个新个性：幂等性。启用幂等性后，ack默认为-1。将生产者中的enable.idompotence设置为true，即启用了幂等性。

开启幂等性的Producer在初始化的时候会被调配一个PID，发往同一Partition的音讯会附带Sequence Number。Broker端会对<PID,Partition,SeqNumber>做缓存，当具备雷同主键的音讯提交时，Broker只会缓存一条。然而每次重启PID就会发生变化，因而只能保障一次会话同一分区的音讯不反复。

5. 消费者组分区调配策略

kafka有两种调配策略，一种是RoundRobin，另一种是Range

RoundRobin是依照消费者组以轮询的形式去给消费者调配分区的形式，前提条件是消费者组中的消费者须要订阅同一个topic。

Range是kafka默认的调配策略，它是通过以后的topic依照肯定范畴来调配的，如果有3个分区，消费者组有两个消费者，则消费者A去生产1和2分区，消费者B去生产3分区。

6. 消费者offset保护

Kafka 0.9 版本之前，consumer默认将offset保留在zookeeper中，0.9 版本开始，offset保留在kafka的一个内置topic中，该topic为_consumer_offsets。

7. 生产者事务

为了实现跨分区会话的事务，须要引入一个全局惟一的Tracscation ID，并将Producer 取得的PID与之绑定。这样当Producer重启后就能够通过正在进行的Transaction ID取得原来的PID。

为了治理Transcation ID，kafka引入了一个新的组件Transcation Coordinator。Producer就是通过和Transcation Coordinator交互取得Transction ID对应的工作状态。

Spring Boot 整合kafka

1. 引入kafka依赖

<dependency>    <groupId>org.springframework.kafka</groupId>    <artifactId>spring-kafka</artifactId></dependency>

2. 配置kafka服务信息

spring:  kafka:    # kafka服务地址    bootstrap-servers: 47.104.155.182:9092    producer:      # 生产者音讯key序列化形式      key-serializer: org.apache.kafka.common.serialization.StringSerializer      # 生产者音讯value序列化形式      value-serializer: org.apache.kafka.common.serialization.StringSerializer    consumer:      # 消费者组      group-id: test-consumer-group      # 消费者音讯value反序列化形式      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer      # 消费者音讯value反序列化形式      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

3. 消费者

@Component@Slf4j@KafkaListener(topics = {"first-topic"},groupId = "test-consumer-group")public class Consumer {    @KafkaHandler    public void receive(String message){                log.info("我是消费者，我接管到的音讯是："+message);    }}

4. 生产者

@RestControllerpublic class Producer {    @Autowired    private KafkaTemplate kafkaTemplate;    @GetMapping("send")    public void send(){        String message = "你好，我是Java旅途";        // 第一个参数 topic        // 第二个参数 音讯        kafkaTemplate.send("first-topic",message);    }}

此是spring-boot-route系列的第十四篇文章，这个系列的文章都比较简单，次要目标就是为了帮忙首次接触Spring Boot 的同学有一个零碎的意识。本文已收录至我的github，欢送各位小伙伴star！

github：https://github.com/binzh303/s...

点关注、不迷路

如果感觉文章不错，欢送关注、点赞、珍藏，你们的反对是我创作的能源，感激大家。

如果文章写的有问题，请不要悭吝，欢送留言指出，我会及时核查批改。

如果你还想更加深刻的理解我，能够微信搜寻「Java旅途」进行关注。回复「1024」即可取得学习视频及精美电子书。每天7:30准时推送技术文章，让你的下班路不在孤单，而且每月还有送书流动，助你晋升硬实力！