首页 / 消息队列 / 【消息队列高手课】- 进阶篇
【消息队列高手课】- 进阶篇
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了【消息队列高手课】- 进阶篇，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含32347字，纯文字阅读大概需要47分钟。
内容图文
Kafka的官方文档：http://kafka.apache.org/documentation/#quickstart
Kafka的核心论文：https://www.kancloud.cn/kancloud/log-real-time-datas-unifying/58708
Kafka原理介绍文档：https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Improvement+Proposals、http://kafka.apache.org/documentation/#design、http://kafka.apache.org/documentation/#implementation
--代码的异步化处理
CompletableFuture简单介绍：
- runAsync 和 supplyAsync方法
- 没有指定Executor的方法会使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码。如果指定线程池，则使用指定的线程池运行。
- runAsync 没有返回值
- supplyAsync有返回值
whenComplete和exceptionally- whenComplete 在执行完成之后顺序执行的内容
whenCompleteAsync把这个任务提交给线程池执行exceptionally 就是执行失败了
- thenApply
handle是任务执行完成对结果的处理，同thenApply不同的是，handle是在线程执行完之后再执行，如果异常也会执行，但是thenApply只会执行成功的- 当一个线程依赖另一个线程时，可以使用 thenApply 方法来把这两个线程串行化。
thenAccept 根据返回结果执行后续步骤- thenRun 没有返回结果执行后续步骤
--异步收发网络数据
Netty:
NIO:
关于JAVA的网络，之前有个比喻形式的总结，分享给大家：
例子：有一个养鸡的农场，里面养着来自各个农户（Thread）的鸡（Socket），每家农户都在农场中建立了自己的鸡舍（SocketChannel）
1、BIO：Block IO，每个农户盯着自己的鸡舍，一旦有鸡下蛋，就去做捡蛋处理；
2、NIO：No-Block IO-单Selector，农户们花钱请了一个饲养员（Selector），并告诉饲养员（register）如果哪家的鸡有任何情况（下蛋）均要向这家农户报告（select keys）；
3、NIO：No-Block IO-多Selector，当农场中的鸡舍逐渐增多时，一个饲养员巡视（轮询）一次所需时间就会不断地加长，这样农户知道自己家的鸡有下蛋的情况就会发生较大的延迟。怎么解决呢？没错，多请几个饲养员（多Selector），每个饲养员分配管理鸡舍，这样就可以减轻一个饲养员的工作量，同时农户们可以更快的知晓自己家的鸡是否下蛋了；
4、Epoll模式：如果采用Epoll方式，农场问题应该如何改进呢？其实就是饲养员不需要再巡视鸡舍，而是听到哪间鸡舍的鸡打鸣了（活跃连接），就知道哪家农户的鸡下蛋了；
5、AIO：Asynchronous I/O, 鸡下蛋后，以前的NIO方式要求饲养员通知农户去取蛋，AIO模式出现以后，事情变得更加简单了，取蛋工作由饲养员自己负责，然后取完后，直接通知农户来拿即可，而不需要农户自己到鸡舍去取蛋。
--应用程序之间网络传输的数据的形式
TCP连接传输的是二进制流，也就是一段一段的101010...，而在一般的网络框架API中，传输形式的是字节(Byte)，一个字节为8个二进制位（bit），所以在这里，二进制流和字节流本质上是一样的。
序列化实现需要考虑的因素：
- 序列化后的数据最好是易于人类阅读的；
- 实现的复杂度是否足够低；
- 序列化和反序列化的速度越快越好；
- 序列化后的信息密度越大越好，也就是说，同样的一个结构化数据，序列化之后占用的存储空间越小越好；
为什么不能直接把内存中，对象对应的二进制数据直接通过网络发送出去，或者保存在文件中呢？为什么还需要序列化和反序列化呢？
答案很简单：因为对象中不只是包含想要传输的数据，还包含这个对象的一些额外的信息，具有语言独特性，如果直接发送过去，有可能接收端语言不通无法识别。而序列化之后，根据约定的序列化格式解析就可以读取数据了。
--传输协议
如何断句呢？给每句话前面加一个表示这句话长度的数字，收到数据的时候，我们按照长度来读取就可以了。
单工协议：
双工协议：
Kafka 有哪些“独门绝技”呢？
1.使用批量消息提升服务端消息处理能力
发送端：Producer只提供了单条send的能力，但是在实际发送时，使用了异步批量发送的机制；攒一波再发送。
Broker端：不会把批消息还原为多条消息，再一条条处理，而是直接处理一个批消息
Consumer端：Consumer 从 Broker 拉到一批消息后，在客户端把批消息解开，再一条一条交给用户代码处理
2.使用顺序读写提升磁盘IO性能
对于磁盘来说它有一个特性，顺序读写的性能要远远好于随机读写，在SSD（固态硬盘）能够达到几倍，在机械硬盘上能够达到几十倍
因为操作系统每次从磁盘读写数据的时候都会寻址，也就是要找到数据在磁盘上的物理位置。
Kafka就是利用了这个特性：
对于每个分区，它把从Producer收到的消息顺序的写入到对应的log文件中，一个文件写满了，就开启一个新文件接着顺序写下去。
消费的时候也是从某个全局的位置开始，也就是从某个log文件的位置开始，顺序的读取消息。
3.实现缓存 - PageCache加速消息读写
PageCache是现代操作系统具备的一种特性，就是操作系统在内存中给磁盘的文件建立缓存。无论我们使用什么语言，在调用操作系统API读写文件时，并不是直接去读写磁盘上的文件，而是操作的PageCache，也就是文件在内存中的副本。
应用程序在写入文件时，操作系统会先写入到PageCache中，然后再一批批的写入到磁盘
应用程序在读取文件时，也是先从PageCache中读取，如果PageCache中有，就直接读取，这样就省去了读取磁盘的时间；若是没有，这时操作系统会引发一个缺页中断，应用程序的读取线程会被阻塞，操作系统把文件复制到PageCache中，再从PageCache中读取。
PageCache的保存策略：
应用程序在使用完某块PageCache后，并不会立即清除而是尽可能地利用空闲的物理内存保存这些 PageCache，除非系统内存不够用，操作系统才会清理掉一部分 PageCache。清理的策略一般是 LRU 或它的变种算法，它保留 PageCache 的逻辑是：优先保留最近一段时间最长使用的PageCache。
Kafka读写消息重新利用了这个特性，一般来说，消息刚刚写到服务端就会被消费。消息队列它的读写比例大致是 1：1
好处：1.读取速度非常快，2.为写入让出磁盘IO资源，间接的提升写入的性能。
4.Zero Copy ：零拷贝技术 - 提升消费性能
在服务端处理消息的逻辑为：
1.从磁盘文件中读取数据到PageCache
2.从PageCahce中复制到应用程序的内存中
3.从应用程序的内存空间复制到Socket的缓冲区，这个过程就是我们调用网络框架API的过程。
使用零拷贝技术可以将上面的复制减少一次，上面的2、3步骤两次复制合并为1次复制，直接将PageCache中数据复制到Socket缓冲区中。这样减少一次数据复制，由于不用把数据复制到用户内存空间，DMA 控制器可以直接完成数据复制，不需要 CPU 参与，速度更快。
--硬件同步原语
硬件同步原语（Atomic Hardware Primitives）是由计算机硬件提供的一组原子操作，我们比较常用的原语主要是 CAS 和 FAA 这两种。
CAS(Compare And Swap):先比较，再交换
FAA(Fetch and Add):先获取，然后给变量 增加 值，返回变量之前的值
/**
* 使用锁实现并发转账
*/
private static int balance = 0;
public static void main(String[] args) {
int count = 10000000;
Lock lock = new ReentrantLock();
final CountDownLatch latch = new CountDownLatch(count);
long l = System.currentTimeMillis();
for (int i = 0; i < count; i++) {
//这里未指定线程池，用到的是ForkJoinPool
CompletableFuture.runAsync(() -> transfer(1, lock, latch));
}
System.out.println(balance);
System.out.println("1消耗时间：" + String.valueOf(System.currentTimeMillis() - l));
try {
//假如countDownLatch，等待所有的线程都执行完成再处理结果
latch.await();
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(balance);
System.out.println("2消耗时间：" + String.valueOf(System.currentTimeMillis() - l));
}
public static void transfer(int add, Lock lock, CountDownLatch latch) {
lock.lock();
try {
balance += add;
latch.countDown();
} finally {
lock.unlock();
}
}
/**
* AtomicInteger CAS
*/
private static AtomicInteger balance = new AtomicInteger(0);
public static void main(String[] args) {
int count = 10000000;
final CountDownLatch latch = new CountDownLatch(count);
long l = System.currentTimeMillis();
for (int i = 0; i < count; i++) {
//这里未指定线程池，用到的是ForkJoinPool
CompletableFuture.runAsync(() -> transfer2(1, latch));
}
System.out.println(balance);
System.out.println("1消耗时间：" + String.valueOf(System.currentTimeMillis() - l));
try {
//假如countDownLatch，等待所有的线程都执行完成再处理结果
latch.await();
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(balance);
System.out.println("2消耗时间：" + String.valueOf(System.currentTimeMillis() - l));
}
/**
* cas
*/
public static void transfer(int add, CountDownLatch latch) {
while(true){
int old = balance.get();
if (balance.compareAndSet(old,old + add)){
break;
}
}
latch.countDown();
}
/**
* ffa
*/
public static void transfer2(int add, CountDownLatch latch) {
balance.getAndAdd(add);
latch.countDown();
}
数据压缩：数据压缩不仅能节省存储空间，还可以用于提升网络传输性能
什么时候做数据压缩？
- 不压缩直接传输需要的时间： 传输未压缩数据的耗时。
- 使用数据压缩需要的时间： 压缩耗时 + 传输压缩数据耗时 + 解压耗时。
故：数据压缩是一个时间换空间或者说CPU资源换存储资源的游戏，所以要考量系统的瓶颈，是磁盘IO、还是网络带宽、还是CPU；再决定是否需要数据压缩。
常见的压缩算法：ZIP，GZIP（更高的压缩比），SNAPPY，LZ4（压缩速度快） 等
如果要对流数据进行压缩，那必须把流数据划分成多个帧，一帧一帧的分段压缩。一般是对出现频率高、占用内容大的内容使用较短的编码。被压缩的数据长度越大，重码率会更高，压缩比也就越高，但是分段也不是越大越好，容易造成解压浪费。
结论：根据你的业务，选择合适的压缩分段，在压缩率、压缩速度和解压浪费之间找到一个合适的平衡。
Kafka如何处理数据压缩？
Kafka默认不开启压缩。可以配置开启压缩及压缩算法
压缩策略是：
Kafka 选择一批消息一起压缩，每一个批消息就是一个压缩分段。也就是在消息发送方压缩，Broker不解压，在消费方解压。
RocketMQ的压缩代码：
//DefaultMQProducerImpl
private boolean tryToCompressMessage(final Message msg) { //批次消息不支持 if (msg instanceof MessageBatch) { //batch dose not support compressing right now return false; } byte[] body = msg.getBody(); if (body != null) { //如果消息体的长度大于4K if (body.length >= this.defaultMQProducer.getCompressMsgBodyOverHowmuch()) { try { //使用zip算法，级别为5 byte[] data = UtilAll.compress(body, zipCompressLevel); if (data != null) { msg.setBody(data); return true; } } catch (IOException e) { log.error("tryToCompressMessage exception", e); log.warn(msg.toString()); } } } return false; }
rocketmq release-4.5.1源码阅读
Rocket消息发送
init - 启动过程 - 实现：org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#start(boolean)
send - 消息发送 - 实现：org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#send(org.apache.rocketmq.common.message.Message)
在 Producer 的接口 MQProducer 中，按照发送方式不同可以分成三类：
- 单向发送（Oneway）：发送消息后立即返回，不处理响应，不关心是否发送成功；
- 同步发送（Sync）：发送消息后等待响应；
- 异步发送（Async）：发送消息后立即返回，在提供的回调方法中处理响应。
switch (communicationMode) {
case ONEWAY:
this.remotingClient.invokeOneway(addr, request, timeoutMillis);
return null;
case ASYNC:
final AtomicInteger times = new AtomicInteger();
long costTimeAsync = System.currentTimeMillis() - beginStartTime;
if (timeoutMillis < costTimeAsync) {
throw new RemotingTooMuchRequestException("sendMessage call timeout");
}
this.sendMessageAsync(addr, brokerName, msg, timeoutMillis - costTimeAsync, request, sendCallback, topicPublishInfo, instance,
retryTimesWhenSendFailed, times, context, producer);
return null;
case SYNC:
long costTimeSync = System.currentTimeMillis() - beginStartTime;
if (timeoutMillis < costTimeSync) {
throw new RemotingTooMuchRequestException("sendMessage call timeout");
}
return this.sendMessageSync(addr, brokerName, msg, timeoutMillis - costTimeSync, request);
default:
assert false;
break;
}
Kafka消息消费源码
订阅：org.apache.kafka.clients.consumer.KafkaConsumer#subscribe(java.util.Collection<java.lang.String>, org.apache.kafka.clients.consumer.ConsumerRebalanceListener)
怎么保证消息不会丢失：
生产和消费两端：基于“请求和确认”机制和业务代码，来确保消息不会丢失。
在服务端（Broker）：一般采用持久化和复制的方式
复制的方式：把消息复制到多个节点，不仅可以解决丢消息的问题，还可以保证消息服务的高可用，即使某一个节点宕机了，也可以使用其他节点来收发消息，所以大部分生产系统都会把消息队列配置成集群模式，开启消息复制。
消息复制面临的问题:
高性能：消息复制写入的节点越多，可用性和数据可靠性就越好，但是写入性能就越低，这是一个天然的矛盾。对消息消费不影响。
一致性：采用“主-从”的复制方式
高可用：需要解决当某个主节点宕机了，尽快选出一个节点接替主节点。
1.节点自选举：还存活的节点通过投票，选择一个主节点；优点是：没有外部依赖，可以实现自我管理。缺点是：投票的实现都比较复杂，并且选举的过程是比较慢的，选举过程一直不可用。
2.大部分复制的实现，都不会选择把消息写入全部副本再返回确认
RocketMQ实现复制的方式：
18年的时候迎来了一次重大的更新：引入 Deldger，增加了一种全新的复制方式。
传统的复制方式：
复制的基本单位为Broker，复制也是采用的主从方式，通常情况下配置成一主一从，也可以配置成一主多从。
两种复制方式：
异步复制：消息先发到主节点上，就返回“写入成功”，然后消息再复制到从节点
RocketMQ中的主从关系是配置固定的，不支持动态切换。如果主节点宕机了，生产者就不能生产消息了，消费者可以自动切换到从节点消费消息；即使有一些消息没来及复制到从节点，依然在主节点的磁盘上，除非主节点的磁盘坏了，否则主节点恢复福的时候，就能把这些消息同步到从节点，再由消费者消费。 - 结果是：不会丢消息，消息的顺序也是没有问题的。缺点是牺牲了一部分的可用性。换取了比较好的性能和数据一致性。
同步双写：消息同步写到主从节点上，都写入成功在返回成功。
本质区别是：写入多少副本再返回成功的问题。在返回成功前，如果写入的副本数不够多，那就回丢消息。
RocketMQ如何保证可用性：
既然一对主从可能会出现不可用的问题，那就再来多个主从节点，同时支持同一个主题的不同队列分布在多对主从节点上去，每对主从节点承担主题中的一部分队列。如果某个主节点宕机了，会自动切换到其他主节点上发消息。这样既解决了部分可用性问题，还通过水平扩容提升Topic性能。
还有一些问题：如果是保证消息顺序一致性的前提下，在主题层面无法保证顺序，只能执行队列来发送消息，这样当这个主节点宕机了，就不能切换到其他主节点。
基于上诉问题：RocketMQ 引入 Dledger，使用新的复制方式
Dledger复制
Dledger在写消息的时候，要求消息复制到半数以上的节点之后，才给客户端返回写入成功，并且它是支持通过选举来动态切换主节点的。
demo：假如有3个节点，主节点宕机了，2个从节点会通过投票选出一个新的主节点来提供服务，相比主从的复制模式，解决了可用性的问题。由于消息至少写入到两个节点才会返回成功，即使主节点宕机了，也至少有另一个从节点的消息跟主节点是一样的，在选举时，总是把数据跟主节点一样的从节点选举为主节点，这样就保证了数据的一致性。
缺点：
选举过程不能提供服务
至少需要3个节点才能保证数据的一致性，3个节点，只能保证一个节点宕机时可用，一旦两个节点同时宕机，即使还有一个节点也无法提供服务，造成部分资源浪费。
另外由于需要至少写入2个节点才返回成功，所以性能上并没有主从异步复制快。
Kafka实现复制的方式
Kafka复制的基本单位是分区。每个分区的几个副本之间，构成一个小的复制集群，Broker是这些分区集群的容器。Kafka的Broker是不分主从的。
分区的多个副本也是采用一主多从的方式，在写入消息的时候使用的也是异步复制，消息在写入到主节点之后，并不会马上返回成功，而是等待足够多的节点都返回成功后再返回
“足够多”的名词为：ISR：保持数据同步的副本，数量是可配置的，注意：这个数量是包含主节点的。
Kafka 使用 ZooKeeper 来监控每个分区的多个节点，如果发现某个分区的主节点宕机了，Kafka 会利用 ZooKeeper 来选出一个新的主节点。
默认情况下，如果所有的ISR都宕机，那么分区就无法同步服务了，也可以配置成让分区继续提供服务，这样只要有一个节点存活就可以提供服务，缺点是会丢失消息。
问题：
假设我们有一个 5 节点的 RocketMQ 集群，采用 Dledger5 副本的复制方式，集群中只有一个主题，50 个队列均匀地分布到 5 个 Broker 上。如果需要你来配置一套 Kafka 集群，要求达到和这个 RocketMQ 集群一样的性能（不考虑 Kafka 和 RocketMQ 本身的性能差异）、可用性和数据一致性，该如何配置？
答：5副本，10个分区，至少保持isr集合中有三个broker
NameServer
在RocketMQ使用NameServer来作为协调器，使Broker、生产者、消费者之间能够通信。
NameServer的作用：作为一个独立的进程，为Broker、生产者和消费者提供服务，为客户端提供寻址服务，帮助客户端找到topic对应的broker地址；监控每个broker的存活状态。
NameServer的部署：可以单节点，可以集群式部署，集群式部署各个节点之间并不通信，每个节点可以独立提供全部的服务。
NameServer的工作过程：
1.路由注册：每个broker启动的时候都要和所有的NameServer通信，来保证NameServer信息的完整性。
2.当Broker保存的topic信息发生变化时，也会主动通知NameServer更新路由信息
3.心跳：Broker定时（每隔30秒）给所有的Broker节点上报路由信息，1是为了保证数据的一致性，另外也是为了监控Broker的健康状态。
4.路由剔除：若是Broker没有定时上报路由信息（默认2分钟），则认为Broker下线，NameServer会移除这个Broker的信息，避免客户端连接一个不可用的Broker
5.路由发现：主要是针对生产者和消费者-统称客户端，客户端会定时（30秒）挨个topic从NameServer拉取Broker的信息。当Broker与客户端通信失败之后，会重新从NameServer拉取信息，然后连接到其他Broker继续生产或者消费，这样实现了自动切换（依赖于重试机制）
稍微介绍下RocketMQ的消费类型：
1. 普通消息：可以重试（默认重试2次），发送到哪个队列都可以，无序
2.普通有序消息：不能重试 动态决定需要发送到哪个队列，除非出现异常情况才会发送到其他队列
3.严格有序消息：不能重试，明确指定到发送的队列
问题来了，为什么RocketMQ选择了NameServer，而不是使用Kafka中类似的Zookeeper？
Zk具备NameServer的所有功能，同时还能够进行master选举、服务治理等，功能完备，为什么还要被放弃呢，
放弃的原因:
1.zk功能太齐全了，RocketMQ只需要轻量级的数据服务，保证最终一致性，而不是强一致性。
2.减少中间件依赖，减少维护成本
结论：根据CAP理论，Rocket在NameServer设计上选择了AP
C（Consistency）:一致性 NameServer实例间不通信，某个时间，实例中的数据会不同
A（Availability）:有效性 有一个节点存活就能提供服务
P（Partiton Tolerance）:分区容错性 NameServer节点多机房部署等等.
源码解读：
NameServer 的总体结构
- NamesrvStartup：程序入口。
- NamesrvController：NameServer 的总控制器，负责所有服务的生命周期管理。
- RouteInfoManager：NameServer 最核心的实现类，负责保存和管理集群路由信息。
- BrokerHousekeepingService：监控 Broker 连接状态的代理类。
- DefaultRequestProcessor：负责处理客户端和 Broker 发送过来的 RPC 请求的处理器。
- ClusterTestRequestProcessor：用于测试的请求处理器。
NameServer中主要存储的信息主要是几个map：在 RouteInfoManager 中
//这里存储的是topic和队列的信息 ，其中每个队列信息对应的QueueData还保存了对应的brokerName - 这里的broker是一组broker，比如一个主节点+多个从节点
private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
//保存了集群每一个brokerName对应的broker的信息
private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
//集群名称 和 broker名称的对应关系
private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
//每个broker当前的动态信息，心跳的更新时间，路由数据版本等
private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
//每个broker对应的消息过滤服务的地址，用于服务端消息过滤
private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
Broker如何注册到NameServer
入口：DefaultRequestProcessor#processRequest
实际处理方法：RouteInfoManager#registerBroker
public RegisterBrokerResult registerBroker(
final String clusterName,
final String brokerAddr,
final String brokerName,
final long brokerId,
final String haServerAddr,
final TopicConfigSerializeWrapper topicConfigWrapper,
final List<String> filterServerList,
final Channel channel) {
RegisterBrokerResult result = new RegisterBrokerResult();
try {
try {
//使用了读写锁，这里为写锁
this.lock.writeLock().lockInterruptibly();
//获取当前的集群名称对应的broker名称
Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
if (null == brokerNames) {
brokerNames = new HashSet<String>();
this.clusterAddrTable.put(clusterName, brokerNames);
}
//将当前的broker加入到set中
brokerNames.add(brokerName);
boolean registerFirst = false; //判断是否为第一次注册
BrokerData brokerData = this.brokerAddrTable.get(brokerName);
if (null == brokerData) { //若是原来的brokerMap中没有，就认为是第一次注册
registerFirst = true;
brokerData = new BrokerData(clusterName, brokerName, new HashMap<Long, String>());
this.brokerAddrTable.put(brokerName, brokerData);
}
Map<Long, String> brokerAddrsMap = brokerData.getBrokerAddrs(); //获取到broker的地址
//Switch slave to master: first remove <1, IP:PORT> in namesrv, then add <0, IP:PORT>
//The same IP:PORT must only have one record in brokerAddrTable
Iterator<Entry<Long, String>> it = brokerAddrsMap.entrySet().iterator();
while (it.hasNext()) {
Entry<Long, String> item = it.next();
//把不等于当前brokerId的节点都去掉
if (null != brokerAddr && brokerAddr.equals(item.getValue()) && brokerId != item.getKey()) {
it.remove();
}
}
//把新的broker地址放入map，得到老的地址
String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr);
registerFirst = registerFirst || (null == oldAddr);
if (null != topicConfigWrapper
&& MixAll.MASTER_ID == brokerId) {
//当前为brokerId的master，并且是第一次注册 ，或者是broker中的信息变更了
if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion())
|| registerFirst) {
ConcurrentMap<String, TopicConfig> tcTable =
topicConfigWrapper.getTopicConfigTable();
if (tcTable != null) {
//处理broker对应的topic对应的信息 进行更新
for (Map.Entry<String, TopicConfig> entry : tcTable.entrySet()) {
this.createAndUpdateQueueData(brokerName, entry.getValue());
}
}
}
}
//更新：brokerLiveTable
BrokerLiveInfo prevBrokerLiveInfo = this.brokerLiveTable.put(brokerAddr,
new BrokerLiveInfo(
System.currentTimeMillis(),
topicConfigWrapper.getDataVersion(),
channel,
haServerAddr));
if (null == prevBrokerLiveInfo) {
log.info("new broker registered, {} HAServer: {}", brokerAddr, haServerAddr);
}
//更新filterServerTable
if (filterServerList != null) {
if (filterServerList.isEmpty()) {
this.filterServerTable.remove(brokerAddr);
} else {
this.filterServerTable.put(brokerAddr, filterServerList);
}
}
//如果当前broker不是master，那么结果中需要带上master的地址
if (MixAll.MASTER_ID != brokerId) {
String masterAddr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
if (masterAddr != null) {
BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.get(masterAddr);
if (brokerLiveInfo != null) {
result.setHaServerAddr(brokerLiveInfo.getHaServerAddr());
result.setMasterAddr(masterAddr);
}
}
}
} finally {
//释放锁
this.lock.writeLock().unlock();
}
} catch (Exception e) {
log.error("registerBroker Exception", e);
}
return result;
}
View Code
客户端如何获取Broker地址
入口：RouteInfoManager#pickupTopicRouteData
/**
* 根据topic找到topic的路由信息
*/
public TopicRouteData pickupTopicRouteData(final String topic) {
//// 初始化返回数据 topicRouteData
TopicRouteData topicRouteData = new TopicRouteData();
boolean foundQueueData = false;
boolean foundBrokerData = false;
Set<String> brokerNameSet = new HashSet<String>();
List<BrokerData> brokerDataList = new LinkedList<BrokerData>();
topicRouteData.setBrokerDatas(brokerDataList);
HashMap<String, List<String>> filterServerMap = new HashMap<String, List<String>>();
topicRouteData.setFilterServerTable(filterServerMap);
try {
try {
//加读锁
this.lock.readLock().lockInterruptibly();
//获取这个主题对应的队列的信息
List<QueueData> queueDataList = this.topicQueueTable.get(topic);
if (queueDataList != null) {
//把队列信息放在topic的队列路由信息中
topicRouteData.setQueueDatas(queueDataList);
foundQueueData = true;
//遍历队列
Iterator<QueueData> it = queueDataList.iterator();
while (it.hasNext()) {
QueueData qd = it.next();
brokerNameSet.add(qd.getBrokerName());
}
//遍历broker信息
for (String brokerName : brokerNameSet) {
BrokerData brokerData = this.brokerAddrTable.get(brokerName);
if (null != brokerData) {
//复制一个broker信息
BrokerData brokerDataClone = new BrokerData(brokerData.getCluster(), brokerData.getBrokerName(), (HashMap<Long, String>) brokerData
.getBrokerAddrs().clone());
//放入到list中
brokerDataList.add(brokerDataClone);
foundBrokerData = true;
for (final String brokerAddr : brokerDataClone.getBrokerAddrs().values()) {
List<String> filterServerList = this.filterServerTable.get(brokerAddr);
filterServerMap.put(brokerAddr, filterServerList);
}
}
}
}
} finally {
//释放du'suo
this.lock.readLock().unlock();
}
} catch (Exception e) {
log.error("pickupTopicRouteData Exception", e);
}
log.debug("pickupTopicRouteData {} {}", topic, topicRouteData);
if (foundBrokerData && foundQueueData) {
return topicRouteData;
}
return null;
}
View Code
Kafka借助Zookeeper构建集群
简单说下Zookeeper：
定义：是一个分布式协调框架，主要解决分布式集群中的一致性的问题
功能：
1.分布式存储系统，提供了类似文件系统的树形结构。
2.zk在树形的存储结构中，每个节点都是一个“ZNode”，同时提供了临时节点。
临时节点的特性：如果创建临时节点的客户端与zk丢失了连接，临时节点也会自动消失。
3.ZNode状态变化通知机制“Watcher”，一旦ZNode和它的子节点状态发生变化时，客户端会收到通知。
Kafka在zk中保存了什么信息？
如图所示：
(注：圆角矩形为临时节点，直角矩形为持久化节点)
Kafka的客户端并不会直连ZK，ZK中的元数据信息通过Broker转发给每个客户端。
源码：
入口：客户端同Broker发生网络传输入口： org.apache.kafka.clients.NetworkClient#poll
调用链：NetworkClient#poll() -> DefaultMetadataUpdater#maybeUpdate(long) -> DefaultMetadataUpdater#maybeUpdate(long, Node)
? Broker接收rpc请求： 入口：kafka.server.KafkaApis#handleTopicMetadataRequest ? Kafka在Broker中维护了一份和Zk中一模一样的数据，只要zk中有变更， 就会通过watcher机制，同步所有的Broker。 ? Zk的缺点： 1.不要在zk中写入大量数据，它并不是真正意义上的存储系统，只适合存储少量数据 2.不要让业务集群的可用性依赖于zk。 ? Kafka主要使用zk保存元数据，监控broker和分区的存活状态， 并利用zk选举。集群的可用性严重依赖ZK。 建议：如果部署大规模kafka集群，可以拆分为多个kafka小集群，每个小集群对应一组独立的zk。- 防止整个kafka集群不可用。 ?RocketMQ 的事务源码
producer发送事务消息
入口：org.apache.rocketmq.client.producer.TransactionMQProducer#sendMessageInTransaction(org.apache.rocketmq.common.message.Message, java.lang.Object)
public TransactionSendResult sendMessageInTransaction(final Message msg,
final LocalTransactionExecuter localTransactionExecuter, final Object arg)
throws MQClientException {
TransactionListener transactionListener = getCheckListener();
if (null == localTransactionExecuter && null == transactionListener) {
throw new MQClientException("tranExecutor is null", null);
}
Validators.checkMessage(msg, this.defaultMQProducer);
SendResult sendResult = null;
//1.对消息添加属性，表明这是一个事务消息，也就是半消息。
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_PRODUCER_GROUP, this.defaultMQProducer.getProducerGroup());
//2.发送半消息
try {
sendResult = this.send(msg);
} catch (Exception e) {
throw new MQClientException("send message Exception", e);
}
LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
Throwable localException = null;
switch (sendResult.getSendStatus()) {
case SEND_OK: {
try {
if (sendResult.getTransactionId() != null) {
msg.putUserProperty("__transactionId__", sendResult.getTransactionId());
}
String transactionId = msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
if (null != transactionId && !"".equals(transactionId)) {
msg.setTransactionId(transactionId);
}
//3.执行本地事务
if (null != localTransactionExecuter) {
localTransactionState = localTransactionExecuter.executeLocalTransactionBranch(msg, arg);
} else if (transactionListener != null) {
log.debug("Used new transaction API");
// 3.1这里就是我们实现的执行本地事务的方法
localTransactionState = transactionListener.executeLocalTransaction(msg, arg);
}
if (null == localTransactionState) {
localTransactionState = LocalTransactionState.UNKNOW;
}
if (localTransactionState != LocalTransactionState.COMMIT_MESSAGE) {
log.info("executeLocalTransactionBranch return {}", localTransactionState);
log.info(msg.toString());
}
} catch (Throwable e) {
log.info("executeLocalTransactionBranch exception", e);
log.info(msg.toString());
localException = e;
}
}
break;
case FLUSH_DISK_TIMEOUT:
case FLUSH_SLAVE_TIMEOUT:
case SLAVE_NOT_AVAILABLE:
localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
break;
default:
break;
}
try {
//4.根据发送半消息的结果和执行本地事务的结果，来判断如何处理后续流程
//判断是提交还是回滚
this.endTransaction(sendResult, localTransactionState, localException);
} catch (Exception e) {
log.warn("local transaction execute " + localTransactionState + ", but end broker transaction failed", e);
}
TransactionSendResult transactionSendResult = new TransactionSendResult();
transactionSendResult.setSendStatus(sendResult.getSendStatus());
transactionSendResult.setMessageQueue(sendResult.getMessageQueue());
transactionSendResult.setMsgId(sendResult.getMsgId());
transactionSendResult.setQueueOffset(sendResult.getQueueOffset());
transactionSendResult.setTransactionId(sendResult.getTransactionId());
transactionSendResult.setLocalTransactionState(localTransactionState);
return transactionSendResult;
}
producer发送事务消息
Broker接收半消息 入口：org.apache.rocketmq.broker.processor.SendMessageProcessor#sendMessage 调用链：org.apache.rocketmq.broker.processor.SendMessageProcessor#sendMessage ->?org.apache.rocketmq.broker.transaction.queue.TransactionalMessageServiceImpl#prepareMessage ->?org.apache.rocketmq.broker.transaction.queue.TransactionalMessageBridge#putHalfMessage ->?org.apache.rocketmq.broker.transaction.queue.TransactionalMessageBridge#parseHalfMessageInner
private MessageExtBrokerInner parseHalfMessageInner(MessageExtBrokerInner msgInner) {
//记录当前消息的主题和队列到新的属性中
MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_TOPIC, msgInner.getTopic());
MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_QUEUE_ID,
String.valueOf(msgInner.getQueueId()));
msgInner.setSysFlag(
MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), MessageSysFlag.TRANSACTION_NOT_TYPE));
// 替换消息的主题和队列为：RMQ_SYS_TRANS_HALF_TOPIC，0
//这个主题和队列对于消费者来说是不可见的，所以消息里的数据不可能被消费，这样就保证，在事务成功提交之前，这个半消息对于消费者来说是消费不到的。
msgInner.setTopic(TransactionalMessageUtil.buildHalfTopic());
msgInner.setQueueId(0);
msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
return msgInner;
}
Broker事务反查 - 基于定时任务
入口：org.apache.rocketmq.broker.transaction.TransactionalMessageCheckService#run -> org.apache.rocketmq.broker.transaction.TransactionalMessageCheckService#onWaitEnd -> org.apache.rocketmq.broker.transaction.queue.TransactionalMessageServiceImpl#check
-> org.apache.rocketmq.broker.transaction.AbstractTransactionalMessageCheckListener#resolveHalfMsg -> org.apache.rocketmq.broker.transaction.AbstractTransactionalMessageCheckListener#sendCheckMessage
Broker提交事务或者回滚
入口：org.apache.rocketmq.broker.processor.EndTransactionProcessor
Kafka 的事务和 Exactly Once
Rocket MQ中的事务：
解决问题：确保本地事务和发消息两个操作，要么都成功，要么都失败。并且Rocket MQ增加了一个事务反查的机制，来尽量提高事务执行的成功率和事务一致性。
Kafka的事务：
解决问题：确认在一个事务中发送的多条消息，要么都发送成功，要么都发送失败。（使用场景不多）
通常的Exactly Once：
消息从生产者发送到Broker，消费者再从Broker拉取消息，再进行消费。在这个过程中，确保每一条消息确保恰好传输一次，不重不丢。
Kafka的 Exactly Once：
在流计算中，用Kafka作为数据源，并且将计算结果保存到Kafka的这种场景下，数据从Kafka的某个主题中消费，在计算集群中计算，再把计算结果保存到Kafka的其他主题中。保证每条消息恰好被计算一次，确保计算结果正确。
Kafka的事务如何实现？
实现流程：
1.开启事务：生产者给协调者发起请求开启事务，协调者在事务日志中记录下事务id
2.给协调者发送请求，告知发送的消息的主题和分区
3.生产者发送消息
注：这里Kafka同Rocket MQ不同，Rocket MQ是在Broker中设置了一个特殊的队列，不能让消费者消费；Kafka却是当做一个正常消息，能够被消费者拉取，只是拉取到客户端了，再由客户端暂时过滤。
4.消息发送完成：生产者给协调者发送消息，告知事务提交或者事务回滚，由协调者开启两阶段提交，完成事务。
5.第一阶段：协调者把事务状态写入到事务日志，并设置为“预提交”，到这一步，无论后续出现什么问题，事务最终都会被提交。
6.第二结算：协调者在跟事务相关的每个分区，写入一条“事务结束”的消息；当消费者读取到事务结束的消息后，会把之前暂时过滤的消息放行给业务代码消费。
7.协调者记录最后一条事务日志，标识事务结束。
（Kafka事务的时序图）
有状态：节点上存储了数据
无状态：节点上不保存数据
那么可知：Rocket MQ 和Kafka的Broker节点都是有节点，因为他们把数据存储在节点上。
新一代消息中间件：Pulsar
Pulsar的Broker是无状态
名词解释：
BookKeeper：分布式存储集群，存储消息数据
Zk：存储元数据
Ledger：一段WAL（Write Ahead Log），包含连续的若干条消息
Entry：消息在Ledger称为Entry
为了保证 Ledger 中的 Entry 的严格顺序，Pulsar 为 Ledger 增加一次性的写入限制，Broker 创建一个 Ledger 后，只有这个 Broker 可以往 Ledger 中写入 Entry，一旦 Ledger 关闭后，无论是 Broker 主动关闭，还是因为 Broker 宕机异常关闭，这个 Ledger 就永远只能读取不能写入
存储计算分离的优点：
1.对于计算节点，不需要进行数据存储，变为无状态节点，一个无状态节点组成的集群、管理、调度都变得非常简单；天然就支持水平扩展；任意一个请求都可以打到任意一个节点上，负载均衡非常灵活；故障转移也很简单，直接把请求转发到其他节点就可以了；
像ZK这种存储与计算不分离的，故障转移就十分麻烦，选举Leader就复杂。
2.对于存储节点，实现的功能十分专注就可以了
缺点：
1.Broker并没有解决问题，只是把问题抛给了BookKeeper，存储计算分离之后，原来一个集群变成了两个集群，整个系统其实变得更加复杂了
2.存储计算分离之后，性能上有一定的损失。比如消费一条消息，Broker还需要从BookKeeper读取数据，再返回客户端，增加了网络请求和内存拷贝。
不过：相对来说，存储计算分离是利用现有的技术，也是减少了复杂度，损失一些性能也是可以接受的。
Kafka和Rocket MQ的选举
Kafka的选举
Kafka使用ZK监控每个分区的多个副本，如果发现某个分区的主节点宕机了，Controller会受到ZK的通知，Controller会从ISR中重新选择一个主节点。
Controller本身也是通过ZK选举产生的。
在Broker启动后，都会尝试在ZK中创建一个临时节点，/controller，并把信息写到这个节点中，并且只有一个Broker能创建这个节点，那么这个Broker就是leader节点，其他没抢到的，就会Watch这个节点。
这个算法虽然不怎么优雅，但胜在简单直接，并且快速公平，是非常不错的选举方法。
RocketMQ的选举：
仅仅靠集群中的节点投票来选举Leader，自我选举的算法， 只会在集群内部进行，同事算法也会十分复杂，一般要进行多次选举。Dledger 采用的是Raft 一致性算法
内容总结
以上是互联网集市为您收集整理的【消息队列高手课】- 进阶篇全部内容，希望文章能够帮你解决【消息队列高手课】- 进阶篇所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。