0604 - 分布式消息中间件

集群部署流程图

- NameServer是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。
- Broker部署相对复杂，Broker分为Master与Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master与Slave的对应关系通过指定相同的BrokerName，不同的BrokerId来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。
- Producer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master建立长连接，且定时向Master发送心跳。Producer完全无状态，可集群部署。
- Consumer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，订阅规则由Broker配置决定。

单Master模式

多Master模式

多Master多Slave模式（异步）

多Master多Slave模式（同步）

消息高可用采用2m-2s（同步双写）方式

git clone https://github.com/apache/rocketmq-externals
cd rocketmq-console
mvn clean package -Dmaven.test.skip=true

发送同步消息

发送异步消息

单向发送消息

负载均衡模式

广播模式

顺序消息

发送延时消息

如果您每次只发送不超过4MB的消息，则很容易使用批处理，样例如上

1）事务消息发送及提交

(1) 发送消息（half消息）。

(2) 服务端响应消息写入结果。

(3) 根据发送结果执行本地事务（如果写入失败，此时half消息对业务不可见，本地逻辑不执行）。

(4) 根据本地事务状态执行Commit或者Rollback（Commit操作生成消息索引，消息对消费者可见）

实现事务的监听接口：当发送半消息成功时，我们使用 executeLocalTransaction 方法来执行本地事务。它返回前一节中提到的三个事务状态之一。checkLocalTranscation 方法用于检查本地事务状态，并回应消息队列的检查请求。

分布式队列因为有高可靠性的要求，所以数据要进行持久化存储。

关系型数据库

文件系统

消息存储

消息发送

消息存储结构

刷盘机制

RocketMQ分布式集群是通过Master和Slave的配合达到高可用性的。

在Consumer的配置文件中，并不需要设置是从Master读还是从Slave 读，当Master不可用或者繁忙的时候，Consumer会被自动切换到从Slave 读。有了自动切换Consumer这种机制，当一个Master角色的机器出现故障后，Consumer仍然可以从Slave读取消息，不影响Consumer程序。这就达到了消费端的高可用性。

如果一个Broker组有Master和Slave，消息需要从Master复制到Slave 上，
有同步和异步两种复制方式。

Producer端，每个实例在发消息的时候，默认会轮询所有的message queue发送，以达到让消息平均
落在不同的queue上。而由于queue可以散落在不同的broker，所以消息就发送到不同的broker下，如
下图：

集群模式

广播模式

对于顺序消息，当消费者消费消息失败后，消息队列 RocketMQ 会自动不断进行消息重试（每次间隔时间为 1 秒），这时，应用会
出现消息消费被阻塞的情况。因此，在使用顺序消息时，务必保证应用能够及时监控并处理消费失败的情况，避免阻塞现象的发生。

对于无序消息（普通、定时、延时、事务消息），当消费者消费消息失败时，您可以通过设置返回状态达到消息重试的结果。
无序消息的重试只针对集群消费方式生效；广播方式不提供失败重试特性，即消费失败后，失败消息不再重试，继续消费新的消息。

特性

消费幂等的必要性

处理方式

源码目录结构：
broker: broker 模块（broke 启动进程）
client ：消息客户端，包含消息生产者、消息消费者相关类
common ：公共包
dev ：开发者信息（非源代码）
distribution ：部署实例文件夹（非源代码）
example: RocketMQ 例代码
filter ：消息过滤相关基础类
filtersrv：消息过滤服务器实现相关类（Filter启动进程）
logappender：日志实现相关类
namesrv：NameServer实现相关类（NameServer启动进程）
openmessageing：消息开放标准
remoting：远程通信模块，给予Netty
srcutil：服务工具类
store：消息存储实现相关类
style：checkstyle相关实现
test：测试相关类
tools：工具类，监控命令相关实现类

消息中间件的设计思路一般是基于主题订阅发布的机制，消息生产者（Producer）发送某一个主题到消
息服务器，消息服务器负责将消息持久化存储，消息消费者（Consumer）订阅该兴趣的主题，消息服
务器根据订阅信息（路由信息）将消息推送到消费者（Push模式）或者消费者主动向消息服务器拉去
（Pull模式），从而实现消息生产者与消息消费者解耦。为了避免消息服务器的单点故障导致的整个系
统瘫痪，通常会部署多台消息服务器共同承担消息的存储。那消息生产者如何知道消息要发送到哪台消
息服务器呢？如果某一台消息服务器宕机了，那么消息生产者如何在不重启服务情况下感知呢？
NameServer就是为了解决以上问题设计的。

在项目启动之初来预测将来项目会碰到什么需求，是极其困难的。消息系统在处理过程中间插入了一个隐含的、基于数据的接口层，两边的处理过程都要实现这一接口。这允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。

系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。

在任何重要的系统中，都会有需要不同的处理时间的元素。例如，加载一张图片比应用过滤器花费更少的时间。消息队列通过一个缓冲层来帮助任务最高效率的执行———写入队列的处理会尽可能的快速。有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。

在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见；如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。

很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

有些情况下，处理数据的过程会失败。除非数据被持久化，否则将造成丢失。消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的"插入-获取-删除"范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。

因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。不需要改变代码、不需要调节参数。扩展就像调大电力按钮一样简单。

在大多使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。Kafka保证一个Partition内的消息的有序性。

Producer

Consumer

Broker

Consumer Group

Topic

Partition

Replica

Leader

Follower

首先 Kafka 消息队列由三个角色组成，左面的是消息的生产方 Producer；中间是 Kafka 集群， Kafka 集群由多台 Kafka server 组成，每个 Server 称为一个 Broker，也就是消息代理；右面的是消息的消费方 Consumer。

Kafka 中消息是按照 Topic 进行划分的，一个 Topic 就是一个 Queue。在实际应用中，不同业务数据就可以设置为不同的 Topic。一个 Topic 可以有多个消费方，当生产方在某个 Topic 发出一条消息后，所有订阅了这个 Topic 的消费方都可以收到这条消息。

为了提高并行能力，Kafka 为每个 Topic 维护了多个 Partition 分区，每个分区可以看作一份追加类型的日志。 每个分区中的消息保证 ID 唯一且有序，新消息不断追加到尾部。Partition 实际存储数据时，会对按大小进行分段（Segment），来保证总是对较小的文件进行写操作，提高性能，方便管理。

如图中间部分，Partition 分布于多个 Broker 上。图中绿色的模块表示 Topic1 被分为了 3 个 Partition。每个 Partition 会被复制多份存在于不同的 Broker 上，如图中红色的模块，这样可以保证主分区出现问题时进行容灾。每个 Broker 可以保存多个 Topic 的多个 Partition。

Kafka 只保证一个分区内的消息有序，不能保证一个 Topic 的不同分区之间的消息有序。为了保证较高的处理效率，所有的消息读写都是在主 Partition 中进行，其他副本分区只会从主分区复制数据。Kafka 会在 ZooKeeper 上针对每个 Topic 维护一个称为 ISR（in-sync replica），就是已同步的副本集。如果某个主分区不可用了，Kafka 就会从 ISR 集合中选择一个副本作为新的主分区。

来看图中的例子，这个 Topic 分为 4 个 Partition，就是图中绿色的 P1到 P4，上部的生产方根据规则选择一个 Partition 进行写入，默认规则是轮询策略。也可以由生产方指定 Partition 或者指定 key 来根据 Hash 值选择 Partition。

消息的发送有三种方式：同步、异步以及 oneway。
（1）同步模式下后台线程中发送消息时同步获取结果，这也是默认模式。
（2）异步的模式允许生产者批量发送数据，可以极大的提高性能，但是会增加丢失数据的风险。
（3）oneway 模式只发送消息不需要返回发送结果，消息可靠性最低，但是低延迟、高吞吐，适用于对可靠性要求不高的场景。

来看消息的消费，Consumer 按照 Group 来消费消息，Topic 中的每一条消息可以被多个 Consumer Group 消费，如上图中的 GroupA 和 GroupB。Kafka 确保每个 Partition 在一个 Group 中只能由一个 Consumer 消费。Kafka 通过 Group Coordinator 来管理 Consumer 实际负责消费哪个 Partition，默认支持 Range 和轮询分配。

Kafka 在 ZK 中保存了每个 Topic 中每个 Partition 在不同 Group 的消费偏移量 offset，通过更新偏移量保证每条消息都被消费。

注意：用多线程来读取消息时，一个线程相当于一个 Consumer 实例。当 Consumer 的数量大于分区的数量的时候，有的 Consumer 线程会读取不到数据。 

hadoop102   hadoop103   hadoop104
zk                  zk                  zk
kafka             kafka             kafka

#broker 的全局唯一编号，不能重复
broker.id=0
#删除 topic 功能使能
delete.topic.enable=true
#处理网络请求的线程数量
num.network.threads=3
#用来处理磁盘 IO 的现成数量
num.io.threads=8
#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
#接收套接字的缓冲区大小
socket.receive.buffer.bytes=102400
#请求套接字的缓冲区大小
socket.request.max.bytes=104857600
#kafka 运行日志存放的路径
log.dirs=/opt/module/kafka/logs
#topic 在当前 broker 上的分区个数
num.partitions=1
#用来恢复和清理 data 下数据的线程数量
num.recovery.threads.per.data.dir=1
#segment 文件保留的最长时间，超时将被删除
log.retention.hours=168
#配置连接 Zookeeper 集群地址
zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181

#KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka
export PATH=$PATH:$KAFKA_HOME/bin

注意：分发之后记得配置其他机器的环境变量

依次在hadoop102、 hadoop103、 hadoop104节点上启动 kafka

$ bin/kafka server stop.s h stop

for i in hadoop102 hadoop 103 hadoop104
do
echo "========== $i =========="
ssh $i '/opt/module/kafka/bin/kafka server start.sh daemon
/opt/module/kafka/config/server.
done

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh zookeeper  hadoop102:2181 list

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --zookeeper hadoop102:2181 --create --replication-factor 3 --partitions 1 --topic first

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --zookeeper hadoop10 2:2181 --delete --topic first

[atguigu@hadoop102 kafka]$ bin/kafka-console-producer.sh --broker-list hadoop102:9092 --topic first

[atguigu@hadoop10 2 kafka]$ bin/kafka-console-consumer.sh \--zookeeper hadoop102: 2181 --topic first

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --zookeeper hadoop102:2181 --describe --topic first

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --zookeeper hadoop102:2181 --alter --topic first --partitions 6

由于生产者生产的消息会不断追加到log 文件末尾，为防止log 文件过大导致数据定位
效率低下，Kafka 采取了分片和索引机制，将每个partition 分为多个segment。
每个segment对应两个文件--“.index”文件和“.log”文件。这些文件位于一个文件
夹下，该文件夹的命名规则为：topic 名称+分区序号。
例如，first 这个topic 有三个分区，则其对应的文件夹为first-0,first-1,first-2

分区的原因

分区的原则

原理

1、副本数据同步策略

2、ISR

3、ack应答机制

4、故障处理细节

将服务器的ACK级别设置为 -1，可以保证Producer到Server之间不会丢失数据，即At Least Once语义 。
相对的将服务器ACK级别设置为 0，可以保证生产者每条消息只会被发送一次，即At Most Once语义。

At Least Once可以保证数据不丢失，但是不能保证数据不重复；相对的， At Least Once可以保证数据不重复，但是不能保证数据不丢失。 但是，对于一些非常重要的息，比如说交易数据，下游数据消费者要求数据既不重复也不丢失，即 Exactly Once语义。

在 0.11版本以前的 Kafka，对此是无能为力的，只能保证数据不丢失，再在下游消费者对数据做全局去重。对于多个下游应用的情况，每个都需要单独做全局去重，这就对性能造成了很大影响。

0.11版本的 Kafka，引入了一项重大特性：幂等性。所谓的幂等性就是指 Producer不论向 Server发送多少次重复数据， Server端都只会持久化一条。幂等性结合 At Least Once语义，就构成了 Kafka的 Exactly Once语义。即：At Least Once + 幂等性 = Exactly Once

要启用幂等性，只需要将Producer的参数中 enable.idompotence设置为 true即可。 Kafka的幂等性实现其实就是将原来下游需要做的去重放在了数据上游。开启幂等性的 Producer在初始化的时候会被分配一个 PID，发往同一 Partition的消息会附带 Sequence Number。而Broker端会对 <PID, Partition, SeqNumber>做缓存，当具有相同主键的消息提交时， Broker只会持久化一条。

但是PID重启就会变化，同时不同的 Partition也具有不同主键，所以幂等性无法保证跨分区跨会话的 Exactly Once。

consumer采用pull（拉）模式从broker中读取数据

push（推）模式很适应消费速率不同的消费者，因为消息发送速率是由broker决定的

一个consumer group 中有多个consumer，一个 topic 有多个partition，所以必然会涉及
到partition 的分配问题，即确定那个partition 由哪个consumer 来消费。

Kafka 有两种分配策略，一是RoundRobin，一是Range。

Kafka0.9版本之前，consumer默认将offset保存在Zookeeper中，从0.9 版本开始，consumer默认
将offset保存在Kafka一个内置的topic 中，该topic 为consumer_offsets。

Kafka的producer生产数据，要写入到 log文件中，写的过程是一直追加到文件末端，为顺序写 。 官网有数据表明，同样的磁盘，顺序写能到 600M/s，而随机写只有 100K/s。这与磁盘的机械机构有关，顺序写之所以快，是因为其省去了大量磁头寻址的时间 。

为了实现跨分区跨会话的事务，需要引入一个全局唯一的Transaction ID，并将 Producer获得的 PID和 Transaction ID绑定。这样当 Producer重启后就可以通过正在进行的 Transaction ID获得原来的 PID。
为了管理Transaction Kafka引入了一个新的组件 Transaction Coordinator。 Producer就是通过和 Transaction Coordinator交互获得 Transaction ID对应的任务状态。 Transaction Coordinator还负责将事务所有写入 Kafka的一个内部 Topic，这样即使整个服务重启，由于事务状态得到保存，进行中的事务状态可以得到恢复，从而继续进行。

上述事务机制主要是从Producer方面考虑，对于 Consumer而言，事务的保证就会相对较弱，尤其时无法保证 Commit的信息被精确消费。这是由于 Consumer可以通过 offset访问任意信息，而且不同的 Segment File生命周期不同，同一事务的消息可能会出现重启后被删除的情况。

Kafka的 Producer发送消息采用的是 异步发送 的方式。在消息发送的过程中，涉及到了两个线程 main线程和 Sender线程 ，以及 一个线程共享变量 RecordAccumulator。main线程将消息发送给 RecordAccumulator Sender线程不断从 RecordAccumulator中拉取消息发送到 Kafka broker。

（1）导入依赖

代码举例

导入依赖

编写代码

数据漏消费和重复消费分析

Kafka 0.9版本之前， offset存储在zookeeper0.9版本及之后，默认将offset存储在Kafka的一个内置的topic中。除此之外， Kafka还可以选择自定义存储offset。
offset的维护是相当繁琐的， 因为需要考虑到消费者的 Rebalace。

当有新的消费者加入消费者组、已有的消费者推出消费者组 或者所订阅的主题的分区发生变化 ，就会触发到分区的重新分配，重新分配的过程叫做 Rebalance。
消费者发生Rebalance之后，每个消费者消费的分区就会发生变 化。 因此消费者要首先获取到自己被重新分配到的分区，并且定位到每个分区最近提交的
offset位置继续消费。

要实现自定义存储offset，需要借助 ConsumerRebalanceListener 以 下为 示例代码 ，其中提交和获取 offset的方法，需要根据所选的 offset存储系统自行实现。

Producer拦截器 (interceptor)是在 Kafka 0.10版本被引入的，主要用于实现 clients端的定制化控制逻辑。

对于producer而言， interceptor使得用户在消息发送前以及 producer回调逻辑前有机会对消息做一些定制化需求，比如 修改消息等。同时， producer允许用户指定多个 interceptor按序作用于同一条消息从而形成一个拦截链 (interceptor chain)。 Intercetpor的实现接口是org.apache.kafka.clients.producer.ProducerInterceptor，其 定义的方法包括：
（1）configure(configs)    获取配置信息 和 初始化数据时调用 。
（2）onSend(ProducerRecord)   该方法封装进KafkaProducer.send方法中，即它运行在用户主线程中。 Producer确保在消息被序列化以及计算分区前调用该方法。
用户可以在该方法中对消息做任何操作，但最好保证不要修改消息所属的 topic和分区， 否则会影响目标分区的计算。
（3）onAcknowledgement(RecordMetadata, Exception）   该方法会在消息从RecordAccumulator成功发送到Kafka Broker之后，或者在发送过程中失败时调用。 并且通常都是在 producer回调逻辑触发之前。 onAcknowledgement运行在producer的IO线程中，因此不要在该方法中放入很重的逻辑，否则会拖慢 producer的消息
发送效率 。  
（4） close    关闭interceptor，主要用于执行一些资源清理工作如前所述，interceptor可能被运行在多个线程中，因此在具体实现时用户需要自行确保线程安全。另外 倘若指定了多个 interceptor，则 producer将按照指定顺序调用它们 ，并仅仅是捕获每个 interceptor可能抛出的异常记录到错误日志中而非在向上传递。这在使用过程中要特别留意。

需求：实现一个简单的双interceptor组成的拦截链。第一个 interceptor会在消息发送前将时间戳信息加到
消息value的最前部；第二个 interceptor会在消息发送后更新成功发送消息数或失败发送消息数。

引入外部依赖越多，系统约容易挂掉

加个MQ进来，需要考虑消息重复、丢失、传输顺序问题

如果多个系统写入数据，其中某个失败了，怎么保证这些系统的一致性问题

普通集群模式

镜像集群模式

如何开启这个镜像集群模式

Kafka8.0后，提供了HA机制（副本机制）；

每个partition的数据都会同步到其他集群上，形成自己的多个副本（replica）；

副本（replica）会选举一个leader出来，其他副本就是follower；

当leader挂掉时，从follower中重新选举出leader，读和写只有Leader负责，follower只用来备份数据；

在写的时候leader会负责把数据同步到所有的follower上去，所有的follower都同步成功后返回ack。

消息传入MQ过程中丢失

解决方案

RabbitMQ

Kafka

消费者消费消息的过程中挂掉，导致消息丢失

解决方案

场景1：一个queue，有多个consumer去消费，consumer从MQ里面读取数据是有序的，每个consumer的执行时间是不固定的，造成数据顺序顺利错误。

场景2：一个queue对应一个consumer，但是consumer里面进行了多线程消费，这样也会造成消息消费顺序错误

Kafka一个topic，一个partiton，一个consumer，但是consumer内部进行多线程消费

在消费者处可以把消息写入多个内存queue，需要顺序执行的消息写入同一个queue然后N个线程分别消费一个内存queue

吧需要顺序执行的消息放入同一个queue中，然后一个queue对应一个消费者去消费

先修复consumer的问题，确保其恢复消费速度，然后将现有consumer都停掉；

新建一个topic，partition是原理的10倍，临时建立好原先10倍的queue数量；

然后写一个临时的分发数据的consumer程序，这个程序部署上去消费积压的数据，消费之后不做耗时的处理，直接轮询写入临时建立好的10倍数量的queue；

接着临时征用10倍的机器来部署consumer，每一批consumer消费一个临时queue的数据；

等到快速消费完积压数据之后，得恢复原先部署的架构，重新用原先的consumer集器来消费信息

写程序，将丢失的那批数据写个临时程序，一点一点的查出来，然后重新灌入mq里面去