一、概述

1、什么是MQ

消息中间件，通过消息的发送和消息的接收分离实现应用程序的解耦，但是这个是MQ的效果并非目的，真正的目的是为了通讯，屏蔽底层复杂的通讯协议，定义一套更加简单的通讯，一个分布式系统中两个模块之间通讯要么是HTTP，要么是自己开发的（rpc） TCP，但是这两种协议其实都是原始的协议。HTTP 协议很难实现两端通讯——模块 A 可以调用 B，B 也可以主动调用 A，如果要做到这个两端都要背上WebServer，而且还不支持⻓连接（HTTP 2.0 的库根本找不到）。TCP 就更加原始了，粘包、心跳、私有的协议，想一想头皮就发麻。MQ 所要做的就是在这些协议之上构建一个简单的“协议”——生产者/消费者模型。MQ 带给我的“协议”不是具体的通讯协议，而是更高层次通讯模型。它定义了两个对象——发送数据的叫生产者；接收数据的叫消费者， 提供一个SDK 让我们可以定义自己的生产者和消费者实现消息通讯而无视底层通讯协议

2、MQ分类

2.1、有broker

又分为中topic和轻topic

kafka、JMS（ActiveMQ）就属于这个流派，生产者会发送 key 和数据到 Broker，由 Broker比较 key 之后决定给哪个消费者。这种模式是我们最常⻅的模式，是我们对 MQ 最多的印象。在这种模式下一个 topic 往往是一个比较大的概念，甚至一个系统中就可能只有一个topic，topic 某种意义上就是 queue，生产者发送 key 相当于说：“hi，把数据放到 key 的队列中”

2.2、无broker

这种的代表是 RabbitMQ（或者说是 AMQP）。生产者发送 key 和数据，消费者定义订阅的队列，Broker 收到数据之后会通过一定的逻辑计算出 key 对应的队列，然后把数据交给队列

3、kafka

基于分布式的分布/订阅消息队列，主要用于数据实时处理领域

官网：Apache Kafka

4、消息队列的两种模式

点对点模式，一对一，消费者主动拉取数据，消息收到后消息清除

消息生产者生产消息发送到Queue中，然后消息消费者从Queue中取出并且消费消息。消息被消费以后， queue 中不再有存储，所以消息消费者不可能消费到已经被消费的消息。Queue 支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费。

发布订阅模式，一对多，消费者消费数据之后不会清除消息

消息生产者（发布）将消息发布到 topic 中，同时有多个消息消费者（订阅）消费该消息。和点对点方式不同，发布到 topic 的消息会被所有订阅者消费。

二、基本架构和概念

 

1. Producer ： 消息生产者，就是向 Kafka发消息 ；
2. Consumer ： 消息消费者，向 Kafka 取消息的客户端；
3. Consumer Group （CG）： 消费者组，由多个 consumer 组成。 消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费；消费者组之间互不影响。 所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。
4. Broker ：经纪人 一台 Kafka 服务器就是一个 broker。一个集群由多个 broker 组成。一个 broker可以容纳多个 topic。
5. Topic ： 话题，可以理解为一个队列， 生产者和消费者面向的都是一个 topic；
6. Partition： 为了实现扩展性，一个非常大的 topic 可以分布到多个 broker（即服务器）上，一个 topic 可以分为多个 partition，每个 partition 是一个有序的队列；
7. Replica： 副本（Replication），为保证集群中的某个节点发生故障时， 该节点上的 partition 数据不丢失，且 Kafka仍然能够继续工作， Kafka 提供了副本机制，一个 topic 的每个分区都有若干个副本，一个 leader 和若干个 follower。
8. Leader： 每个分区多个副本的“主”，生产者发送数据的对象，以及消费者消费数据的对象都是 leader。
9. Follower： 每个分区多个副本中的“从”，实时从 leader 中同步数据，保持和 leader 数据的同步。 leader 发生故障时，某个 Follower 会成为新的 leader。

 

三、安装

1、window上安装

kafka(3版本之后不需要安装zk) 

在Windows系统上安装消息队列kafka

2、Linux安装

2.1、安装

需要提前安装好java和zk

下载kafka的压缩包:http://kafka.apache.org/downloads

解压到路径 /usr/local/kafka

修改配置文件/usr/local/kafka/kafka2.11-2.4/config/server.properties

#broker.id属性在kafka集群中必须要是唯一
broker.id= 0
#kafka部署的机器ip和提供服务的端口号
listeners=PLAINTEXT://192.168.65.60:9092
#kafka的消息存储文件
log.dir=/usr/local/data/kafka-logs
#kafka连接zookeeper的地址
zookeeper.connect= 192.168.65.60:2181

启动kafka

./kafka-server-start.sh -daemon../config/server.properties

验证是否启动成功：

进入到zk中的节点看id是 0 的broker有没有存在（上线）

ls /brokers/ids/

2.2、server.properties核心配置详解：

 四、基本知识

1、创建主题、生产消息、消费消息

执行以下命令创建名为“test”的topic，这个topic只有一个partition，并且备份因子也设置为1

./kafka-topics.sh --create --zookeeper 172.16.253.35:2181 --replication-factor 1 --partitions 1 --topic test

查看当前kafka内有哪些topic

./kafka-topics.sh --list --zookeeper 172.16.253.35:2181

kafka自带了一个producer命令客户端，可以从本地文件中读取内容，或者我们也可以以命令行中直接输入内容，并将这些内容以消息的形式发送到kafka集群中。在默认情况下，每一个行会被当做成一个独立的消息。使用kafka的发送消息的客户端，指定发送到的kafka服务器地址和topic 

./kafka-console-producer.sh --broker-list 172.16.253.38:9092 --topic test

 

对于consumer，kafka同样也携带了一个命令行客户端，会将获取到内容在命令中进行输 出， 默认是消费最新的消息 。使用kafka的消费者消息的客户端，从指定kafka服务器的指定 topic中消费消息

方式一：从最后一条消息的偏移量+1开始消费

./kafka-console-consumer.sh --bootstrap-server 172.16.253.38:9092 --topic testCopy to clipboardErrorCopied

方式二：从头开始消费

./kafka-console-consumer.sh --bootstrap-server 172.16.253.38:9092 --from-beginning --topic test

 2、单播消息、多播消息

单播场景：同一个消费组里的多个消费者只有一个消费者能消费到某一个topic消息

多播场景：一个topic中的一条消息要被多个消费者消费，需要让不同的消费者处于不同的消费组

3、消费组详细信息

# 查看当前主题下有哪些消费组
./kafka-consumer-groups.sh --bootstrap-server 10.31.167.10:9092 --list
# 查看消费组中的具体信息：比如当前偏移量、最后一条消息的偏移量、堆积的消息数量
./kafka-consumer-groups.sh --bootstrap-server 172.16.253.38:9092 --describe --group testGroup

● current-offset:最后被消费的消息的偏移量

● Log-end-offset:消息总量（最后一条消息的偏移量）

● Lag:积压了多少条消息 

4、小结

• 消息是被存储的，发送给broker保存至本地的日志文件中(配置文件中配置的目录)
• 消息的保存是有顺序的，通过offset偏移量来描述消息的有序性
• 消费者消费消息时也是通过offset描述当前要消费的那条消息的位置

五、主题、分区、_comsumer_offsets

一个topic中的文件可以非常多，多到几个g，存在log文件中，为了解决这个文件问题，引入了partition分区，分开存储一个topic中的消息，比如一个topic三个partition，，这个topic中的消息可以分三个地方部署，partition不仅解决了文件存储过大的问题，同时支持读写多个分区

给主题创建多个分区

./kafka-topics.sh --create --zookeeper localhost:2181 --partitions 2 --topic test1

查看topic主题的分区信息

./kafka-topics.sh --describe --zookeeper localhost:2181 --topic test1

实际上是存在data/kafka-logs/test1-0 和 test1-1中的0000000.log文件中

000.index:相当于一个稀疏索引文件

000.timeindex：相当于一个时间索引文件

000.log就是消息

kafka内部自己创建了一个_consumer_offsets主题，这个主题有49个分区，在data/log目录下可以查看，kafka会定期将分区的offset提交给这个主题，即这个主题主要用来保存消费者消费消息的偏移量，比如一个test主题有100条消息，一个消费者两给消费者a和b,a从1消费到50突然挂掉了，此时kafka会将偏移量50保存到_consumer_offsets主题中，b消费的时候从51消费

保存到_consumer_offsets时，key是consumerGroupId+topic+分区号，value就是当前的offset，

因为要处理高并发，所以默认有50个分区，可以通过offset.num.topic.partitions设置，这样通过加机器抗大并发，通过如下公式计算要提交到_consumer_offset中的哪个分区

公式：hash(consumerGroupId)%_consumer_offsets分区数

文件中保存的消息，默认保存7天，7天到期自动清除

六、kafka集群、副本

1、搭建

略

2、副本

在一个集群即3个broker，同一个主题创建2个分区 3个人副本

./kafka-topics.sh --create --zookeeper 172.16.253.35:2181 --replication-factor 3 --partitions 2 --topic my-replicated-topic
查看详情

./kafka-topics.sh --describe --zookeeper 172.16.253.35:2181 --topic my-replicated-topic

kafka集群中由多个broker组成

一个broker中存放一个topic的不同partition——副本

leader：副本里的概念

• 每个partition都有一个broker作为leader。
• 消息发送方要把消息发给哪个broker？就看副本的leader是在哪个broker上面。副本里的leader专⻔用来接收消息。
• 接收到消息，其他follower通过poll的方式来同步数据。

通过kill掉leader之后查看

# kill掉leader
ps -aux | grep server.properties
kill 17631
# 查看topic情况
./kafka-topics.sh --describe --zookeeper 172.16.253.35:2181 --topic my-replicated-topic

follower：leader处理所有针对这个partition的读写请求，而follower被动复制leader，不提供读写（主要是为了保证多副本数据与消费的一致性），如果leader所在的broker挂掉，那么就会进行新leader的选举，至于怎么选，在之后的controller的概念中介绍。

isr： 可以同步的broker节点和已同步的broker节点，存放在isr集合中。

3、消息的生产和发送

#生产
./kafka-console-producer.sh --broker-list 172.16.253.38:9092,172.16.253.38:9093,172.16.253.38:9094 --topic my-replicated-topic
#消费
./kafka-console-consumer.sh --bootstrap-server 172.16.253.38:9092,172.16.253.38:9093,172.16.253.38:9094 --from-beginning --topic my-replicated-topic

 

• 一个partitions只能被一个消费组里的一个消费者消费目的是为了保证消费的顺序性，但是多个partition的多个消费者消费的消息顺序性是没法保证的，如何保证了？后续
• partition的数量决定了消费者的数量，建议同一个消费组里的消费者数量不要大于partition数量，多的消费者消费不到消息
• 如果消费者挂了，会触发rebalance机制，让其他消费者消费该分区

七、kafka java生产者和消费者

 待补充

八、Springboot整合kafka待补充

待补充

九、深入理解&生产优化

1、controller

创建一个主题多个分区时候，Kafka集群中的broker在zk中创建临时序号节点，序号最小的节点（最先创建的节点）将作为集群的controller，负责管理整个集群中的所有分区和副本的状态：

• 当某个分区的leader副本出现故障时，由控制器负责为该分区选举新的leader副本,选举的规则是从ISR集合最左边获得。
• 当检测到某个分区的ISR集合发生变化时，由控制器负责通知所有broker更新其元数据信息。
• 当使用kafka-topics.sh脚本为某个topic增加分区数量时，同样还是由控制器负责让新分区被其他节点感知到。

2、rebalance机制

前提是：消费者没有指明分区消费。

触发条件：当消费组里消费者和分区的关系发生变化，那么就会触发rebalance机制。这个机制会重新调整消费者消费哪个分区。在触发rebalance机制之前，消费者消费哪个分区有三种策略：

• range：通过公示来计算某个消费者消费哪个分区，前面的消费者是分区总数/消费者数量+1 之后的消费者是分区策略/消费者数量
• 轮询：大家轮着消费
• sticky：在触发了rebalance后，在消费者消费的原分区不变的基础上进行调整。如果这个策略没有打开那么就要全部重新分配，建议开启

3、HW和LEO

HW:已完成同步的位置，消息在写入broker时，且每个broker完成这条消息的变化，hw才发生变化，在这之前消费者是消费不到这条消息的

4、如何防止消息丢失

• 发送方： ack是 1 或者-1/all 可以防止消息丢失，如果要做到99.9999%，ack设成all，把min.insync.replicas配置成分区备份数

• 消费方：把自动提交改为手动提交。

5、如何防止消息重复消费

一条消息被消费者消费多次。如果为了消息的不重复消费，而把生产端的重试机制关闭、消费端的手动提交改成自动提交，这样反而会出现消息丢失，那么可以直接在防治消息丢失的手段上再加上消费消息时的幂等性保证，就能解决消息的重复消费问题。

6、如何做到顺序消费