一、为什么使用消息队列

1.使用同步的通信方式来解决多个服务之间的通信

同步的通信方式会存在性能和稳定性的问题。

2.使用异步的通信方式

针对于同步的通信方式来说，异步的方式，可以让上游快速成功，极大提高了系统的吞吐量。而且在分布式系统中，通过下游多个服务的分布式事务的保障，也能保障业务执行之后的最终一致性。

消息队列解决具体的是什么问题——通信问题。

二、消息队列的流派

1.什么是MQ

Message Queue（MQ），消息队列中间件。很多⼈都说：MQ 通过将消息的发送和接收分离来实现应⽤程序的异步和解偶，这个给⼈的直觉是——MQ 是异步的，⽤来解耦的，但是这个只是 MQ 的效果⽽不是⽬的。MQ 真正的⽬的是为了通讯，屏蔽底层复杂的通讯协议，定义了⼀套应⽤层的、更加简单的通讯协议。⼀个分布式系统中两个模块之间通讯要么是HTTP，要么是⾃⼰开发（rpc） TCP，但是这两种协议其实都是原始的协议。HTTP 协议很难实现两端通讯——模块 A 可以调⽤ B，B 也可以主动调⽤ A，如果要做到这个两端都要背上WebServer，⽽且还不⽀持⻓连接（HTTP 2.0 的库根本找不到）。TCP 就更加原始了，粘包、⼼跳、私有的协议，想⼀想头⽪就发麻。MQ 所要做的就是在这些协议之上构建⼀个简单的“协议”——⽣产者/消费者模型。MQ 带给我的“协议”不是具体的通讯协议，⽽是更⾼层次通讯模型。它定义了两个对象——发送数据的叫⽣产者；接收数据的叫消费者，提供⼀个SDK 让我们可以定义⾃⼰的⽣产者和消费者实现消息通讯⽽⽆视底层通讯协议

2.MQ的分类

2.1有 Broker 的 MQ

这个流派通常有⼀台服务器作为 Broker，所有的消息都通过它中转。⽣产者把消息发送给它就结束⾃⼰的任务了，Broker 则把消息主动推送给消费者（或者消费者主动轮询）

2.1.1重 Topic

kafka、JMS（ActiveMQ）就属于这个流派，⽣产者会发送 key 和数据到 Broker，由 Broker⽐较 key 之后决定给哪个消费者。这种模式是我们最常⻅的模式，是我们对 MQ 最多的印象。在这种模式下⼀个 topic 往往是⼀个⽐较⼤的概念，甚⾄⼀个系统中就可能只有⼀个topic，topic 某种意义就是 queue，⽣产者发送 key 相当于说：“hi，把数据放到 key 的队列中” 如上图所示，Broker 定义了三个队列，key1，key2，key3，⽣产者发送数据的时候会发送key1 和 data，Broker 在推送数据的时候则推送 data（也可能把 key 带上）。虽然架构⼀样但是 kafka 的性能要⽐ jms 的性能不知道⾼到多少倍，所以基本这种类型的MQ 只有 kafka ⼀种备选⽅案。如果你需要⼀条暴⼒的数据流（在乎性能⽽⾮灵活性）那么kafka 是最好的选择

2.1.2轻 Topic

这种的代表是 RabbitMQ（或者说是 AMQP）。⽣产者发送 key 和数据，消费者定义订阅的队列，Broker 收到数据之后会通过⼀定的逻辑计算出 key 对应的队列，然后把数据交给队列这种模式下解耦了 key 和 queue，在这种架构中 queue 是⾮常轻ᰁ级的（在 RabbitMQ 中它的上限取决于你的内存），消费者关⼼的只是⾃⼰的 queue；⽣产者不必关⼼数据最终给谁只要指定 key 就⾏了，中间的那层映射在 AMQP 中叫 exchange（交换机）。

AMQP 中有四种 exchange

Direct exchange：key 就等于 queue
Fanout exchange：⽆视 key，给所有的 queue 都来⼀份
Topic exchange：key 可以⽤“宽字符”模糊匹配 queue
Headers exchange：⽆视 key，通过查看消息的头部元数据来决定发给那个
queue（AMQP 头部元数据⾮常丰富⽽且可以⾃定义）这种结构的架构给通讯带来了很⼤的灵活性，我们能想到的通讯⽅式都可以⽤这四种exchange 表达出来。如果你需要⼀个企业数据总线（在乎灵活性）那么 RabbitMQ 绝对的值得⼀⽤

2.2无 Broker 的 MQ

⽆ Broker 的 MQ 的代表是 ZeroMQ。该作者⾮常睿智，他⾮常敏锐的意识到——MQ 是更⾼级的 Socket，它是解决通讯问题的。所以 ZeroMQ 被设计成了⼀个“库”⽽不是⼀个中间件，这种实现也可以达到——没有 Broker 的⽬的节点之间通讯的消息都是发送到彼此的队列中，每个节点都既是⽣产者⼜是消费者。ZeroMQ做的事情就是封装出⼀套类似于 Socket 的 API 可以完成发送数据，读取数据

ZeroMQ 其实就是⼀个跨语⾔的、᯿ᰁ级的 Actor 模型邮箱库。你可以把⾃⼰的程序想象成⼀个 Actor，ZeroMQ 就是提供邮箱功能的库；ZeroMQ 可以实现同⼀台机器的 RPC 通讯也可以实现不同机器的 TCP、UDP 通讯，如果你需要⼀个强⼤的、灵活、野蛮的通讯能⼒，别犹豫 ZeroMQ

三、Kafka的基本知识

1.Kafka的安装

部署一台zookeeper服务器
安装jdk
下载kafka的安装包：http://kafka.apache.org/downloads
上传到kafka服务器上：/usr/local/kafka
解压缩压缩包

进入到config目录内，修改server.properties

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

进入到bin目录内，执行以下命令来启动kafka服务器（带着配置文件）

# 先启动zookeeper，使用自带的zookeeper
bin/zookeeper-server-start.sh -daemon   config/zookeeper.properties 
# 启动kafka
bin/kafka-server-start.sh -daemon  config/server.properties
# 这里的-daemon为可选参数，加上就是后台启动，不加就是前台启动会打印到控制台。

校验kafka是否启动成功：

# 进入到zk内查看是否有kafka的节点：
bin/zookeeper-shell.sh 127.0.0.1:2181
ls /brokers/ids/

2.kafka中的一些基本概念

名称	解释
Broker	消息中间件处理节点，⼀个Kafka节点就是⼀个broker，⼀个或者多个Broker可以组成⼀个Kafka集群
Topic	Kafka根据topic对消息进⾏归类，发布到Kafka集群的每条消息都需要指定⼀个topic
Producer	消息⽣产者，向Broker发送消息的客户端
Consumer	消息消费者，从Broker读取消息的客户端
ConsumerGroup	每个Consumer属于⼀个特定的Consumer Group，⼀条消息可以被多个不同的Consumer Group消费，但是⼀个Consumer Group 中只能有⼀个Consumer能够消费该消息
Partition	物理上的概念，⼀个topic可以分为多个partition，每个partition内部消息是有序的

3.创建topic

通过kafka命令向zk中创建⼀个主题

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

查看当前zk中所有的主题

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

4.发送消息

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

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

5.消费消息

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

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

# 创建消费者时，若不指定group.id,则该消费者属于默认消费组
./kafka-console-consumer.sh --bootstrap-server 192.168.245.21:9092 -topic test

⽅式⼆：从头开始消费

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

6.关于消息的细节

⽣产者将消息发送给broker，broker会将消息保存在本地的⽇志⽂件中
```
/usr/local/kafka/data/kafka-logs/主题-分区/00000000.log
```
消息的保存是有序的，通过offset偏移量来描述消息的有序性
消费者消费消息时也是通过offset来描述当前要消费的那条消息的位置

7.单播消息

在⼀个kafka的topic中，启动两个消费者，⼀个⽣产者，问：⽣产者发送消息，这条消息是否同时会被两个消费者消费？

如果多个消费者在同⼀个消费组，那么只有⼀个消费者可以收到订阅的topic中的消息。换⾔之，同⼀个消费组中只能有⼀个消费者收到⼀个topic中的消息。

./kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --consumer-property group.id=testGroup --topic test

8.多播消息

不同的消费组订阅同⼀个topic，那么不同的消费组中只有⼀个消费者能收到消息。实际上也是多个消费组中的多个消费者收到了同⼀个消息。

./kafka-console-consumer.sh --bootstrap-server 192.168.0.4:9092 --consumer-property group.id=testGroup1 --topic test

./kafka-console-consumer.sh --bootstrap-server 192.168.245.21:9092 --consumer-property group.id=testGroup2 --topic test

下图就是描述多播和单播消息的区别

9. 查看消费组的详细信息

# 查看当前主题下有哪些消费组
./kafka-consumer-groups.sh --bootstrap-server 192.168.245.21:9092 --list
# 通过以下命令可以查看到消费组的详细信息：
./kafka-consumer-groups.sh --bootstrap-server 192.168.245.21:9092 --describe --group testGroup

重点关注以下⼏个信息：

current-offset: 最后被消费的消息的偏移量
Log-end-offset: 消息总量（最后⼀条消息的偏移量）
Lag：积压了多少条消息

四、Kafka中主题和分区的概念

1.主题Topic

主题-topic在kafka中是⼀个逻辑的概念，kafka通过topic将消息进⾏分类。不同的topic会被订阅该topic的消费者消费。

但是有⼀个问题，如果说这个topic中的消息⾮常⾮常多，多到需要⼏T来存，因为消息是会被保存到log⽇志⽂件中的。为了解决这个⽂件过⼤的问题，kafka提出了Partition分区的概念

2.分区Partition

2.1分区的概念

通过partition将⼀个topic中的消息分区来存储。这样的好处有多个：

分区存储，可以解决统⼀存储⽂件过⼤的问题
提供了读写的吞吐量：读和写可以同时在多个分区中进⾏

2.2创建多分区的主题

# 为⼀个主题创建多个分区
./kafka-topics.sh --create --zookeeper 192.168.245.21:2181 --replicationfactor 1 --partitions 2 --topic test1

# 可以通过这样的命令查看topic的分区信息
./kafka-topics.sh --describe --zookeeper 192.168.245.21:2181 --topic test1

3.kafka中消息日志文件保存的内容

00000.log：这个文件中保存的就是消息
consumer_offsets-49: kafka内部⾃⼰创建了consumer_offsets主题包含了50个分区。这个主题⽤来存放消费者消费某个主题的偏移量。因为每个消费者都会⾃⼰维护着消费的主题的偏移量，也就是说每个消费者会把消费的主题的偏移量⾃主上报给kafka中的默认主题： consumer_offsets。因此kafka为了提升这个主题的并发性，默认设置了50个分区。
- 提交到哪个分区：通过hash函数：hash(consumerGroupId) % __consumer_offsets主题的分区数
- 提交到该主题中的内容是：key是consumerGroupId+topic+分区号，value就是当前offset的值
⽂件中保存的消息，默认保存7天。七天到后消息会被删除。

五、Kafka集群操作

1.搭建kafka集群（三个broker）

创建三个server.properties⽂件

# 0 1 2
broker.id=2
// 9092 9093 9094
listeners=PLAINTEXT://192.168.245.21:9094
//kafka-logs kafka-logs-1 kafka-logs-2
log.dir=/usr/local/data/kafka-logs-2

通过命令来启动三台broker

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

校验是否启动成功

# 进入到zk内查看是否有kafka的节点：
bin/zookeeper-shell.sh 127.0.0.1:2181
ls /brokers/ids/

2.副本的概念

在创建主题时，除了指明了主题的分区数以外，还指明了副本数，那么副本是⼀个什么概念呢？

# 副本是对分区的备份。在集群中，不同的副本会被部署在不同的broker上。下⾯例⼦：创建1个主题，2个分区、3个副本。
./kafka-topics.sh --create --zookeeper 192.168.245.21:2181 --replication-factor 3 --partitions 2 --topic my-replicated-topic

副本是为了为主题中的分区创建多个备份，多个副本在kafka集群的多个broker中，会有⼀个副本作为leader，其他是follower。查看topic情况：

# 查看topic情况
./kafka-topics.sh --describe --zookeeper 192.168.245.21:2181 --topic my-replicated-topic

leade: kafka的写和读的操作，都发⽣在leader上。leader负责把数据同步给follower。当leader挂了，经过主从选举，从多个follower中选举产⽣⼀个新的leader
follower: 接收leader的同步的数据
isr：可以同步和已同步的节点会被存⼊到isr集合中。这⾥有⼀个细节：如果isr中的节点性能较差，会被提出isr集合。

（重点～！）此时，broker、主题、分区、副本这些概念就全部展现了，⼤家需要把这些概念梳理清楚：集群中有多个broker，创建主题时可以指明主题有多个分区（把消息拆分到不同的分区中存储），可以为分区创建多个副本，不同的副本存放在不同的broker⾥。但是__consumer_offsets副本只有一个，不会保存在其他的brokers里面

3.关于集群消费

3.1向集群发送消息：

./kafka-console-producer.sh --broker-list 192.168.245.21:9092,192.168.245.21:9093,192.168.245.21:9094 --topic my-replicated-topic

3.2从集群中消费消息

./kafka-console-producer.sh --broker-list
192.168.245.21:9092,192.168.245.21:9093,192.168.245.21:9094 --topic my-replicated-topic

3.3指定消费组来消费消息

./kafka-console-consumer.sh --bootstrap-server
192.168.245.21:9092,192.168.245.21:9093,192.168.245.21:9094 --frombeginning --consumer-property group.id=testGroup1 --topic my-replicated-topic

3.4分区分消费组的集群消费中的细节

⼀个partition只能被⼀个消费组中的⼀个消费者消费，⽬的是为了保证消费的顺序性，但是多个partion的多个消费者消费的总的顺序性是得不到保证的，那怎么做到消费的总顺序性呢？
partition的数量决定了消费组中消费者的数量，建议同⼀个消费组中消费者的数量不要超过partition的数量，否则多的消费者消费不到消息
如果消费者挂了，那么会触发rebalance机制（后⾯介绍），会让其他消费者来消费该分区

六、kafka集群中的controller、rebalance、HW

1.controller

集群中谁来充当controller 每个broker启动时会向zk创建⼀个临时序号节点，获得的序号最⼩的那个broker将会作为集群中的controller，负责这么⼏件事：
当集群中有⼀个副本的leader挂掉，需要在集群中选举出⼀个新的leader，选举的规则是从isr集合中最左边获得。
当集群中有broker新增或减少，controller会同步信息给其他broker
当集群中有分区新增或减少，controller会同步信息给其他broker

2.rebalance机制

前提：消费组中的消费者没有指明分区来消费
触发的条件：当消费组中的消费者和分区的关系发⽣变化的时候
分区分配的策略：在rebalance之前，分区怎么分配会有这么三种策略
- range：根据公示计算得到每个消费消费哪⼏个分区：前⾯的消费者是分区总数/消费者数量+1,之后的消费者是分区总数/消费者数量
- 轮询：⼤家轮着来
- sticky：粘合策略，如果需要rebalance，会在之前已分配的基础上调整，不会改变之前的分配情况。如果这个策略没有开，那么就要进⾏全部的重新分配。建议开启。

3.HW和LEO

LEO是某个副本最后消息的消息位置（log-end-offset）

HW是已完成同步的位置。消息在写⼊broker时，且每个broker完成这条消息的同步后，hw才会变化。在这之前消费者是消费不到这条消息的。在等待Isr同步完成之后，HW更新之后，消费者才能消费到这条消息，这样的⽬的是防⽌消息的丢失。

7.Kafka中的优化问题

1.如何防⽌消息丢失

acks=0 acks = 0如果设置为零，那么生产者将完全不会管服务器是否收到消息。该记录将立即添加到套接字缓冲区中并视为已发送。并且重试配置不会生效（因为客户端通常不会知道任何故障）。返回值的偏移量将始终等于 -1。该方式具有最大的吞吐量，一般建议直接配合 send(msg)使用。
acks=1 当leader接受到消息就会直接给客户端返回成功，一般情况下这种模式都能很好的保证数据的不丢失，只有在laeder接受到数据，然后还没来得及同步到follower，就挂掉了才会导致数据的丢失，这种概率还是比较小的。这也是默认的选择方式，兼具较好的吞吐和较高的可靠性
acks=all 或者 acks=-1 当leader接受到消息，并同步到了一定数量的follower，才向生产者发生成功的消息，同步到的follower数量由 broker 端的 min.insync.replicas 决定除非一些不可抗力因素，这种方式基本可以确保数据的完全不丢失。
⽣产者：1）使⽤同步发送 2）把ack设成1或者all，并且设置同步的分区数>=2
消费者：把⾃动提交改成⼿动提交

2.如何防⽌重复消费

在防⽌消息丢失的⽅案中，如果⽣产者发送完消息后，因为⽹络抖动，没有收到ack，但实际上broker已经收到了。

此时⽣产者会进⾏重试，于是broker就会收到多条相同的消息，⽽造成消费者的重复消费。

怎么解决：

⽣产者关闭重试：会造成丢消息（不建议）
消费者解决⾮幂等性消费问题：所谓的幂等性：多次访问的结果是⼀样的。对于rest的请求（get（幂等）、post（⾮幂等）、put（幂等）、delete（幂等））解决⽅案：
- 在数据库中创建联合主键，防⽌相同的主键创建出多条记录
- 使⽤分布式锁，以业务id为锁。保证只有⼀条记录能够创建成功

3.如何做到消息的顺序消费

⽣产者：保证消息按顺序消费，且消息不丢失——使⽤同步的发送，ack设置成⾮0的值。
消费者：主题只能设置⼀个分区，消费组中只能有⼀个消费者

kafka的顺序消费使⽤场景不多，因为牺牲掉了性能，但是⽐如rocketmq在这⼀块有专⻔的功能已设计好。

4.如何解决消息积压问题

4.1消息积压问题的出现

消息的消费者的消费速度远赶不上⽣产者的⽣产消息的速度，导致kafka中有⼤量的数据没有被消费。随着没有被消费的数据堆积越多，消费者寻址的性能会越来越差，最后导致整个kafka对外提供的服务的性能很差，从⽽造成其他服务也访问速度变慢，造成服务雪崩。

4.2消息积压的解决⽅案

在这个消费者中，使⽤多线程，充分利⽤机器的性能进⾏消费消息。
通过业务的架构设计，提升业务层⾯消费的性能。
创建多个消费组，多个消费者，部署到其他机器上，⼀起消费，提⾼消费者的消费速度
创建⼀个消费者，该消费者在kafka另建⼀个主题，配上多个分区，多个分区再配上多个消费者。该消费者将poll下来的消息，不进⾏消费，直接转发到新建的主题上。此时，新的主题的多个分区的多个消费者就开始⼀起消费了。——不常⽤

5.实现延时队列的效果

5.1应⽤场景

订单创建后，超过30分钟没有⽀付，则需要取消订单，这种场景可以通过延时队列来实现

5.2具体⽅案

kafka中创建创建相应的主题
消费者消费该主题的消息（轮询）
消费者消费消息时判断消息的创建时间和当前时间是否超过30分钟（前提是订单没⽀付）
- 如果是：去数据库中修改订单状态为已取消
- 如果否：记录当前消息的offset，并不再继续消费之后的消息。等待1分钟后，再次向kafka拉取该offset及之后的消息，继续进⾏判断，以此反复。

1.kafka笔记.md 22 KB Előzmények Nyers