1 RocketMQ基本理论

1.1 发展历史

阿里巴巴消息中间件起源于2001年的五彩石项目，Notify在这期间应运而生，用于交易核心消息的流转。
2010年，B2B开始大规模使用ActiveMQ作为消息内核，随着阿里业务的快速发展，急需一款支持顺序消息、拥有海量消息堆积能力的消息中间件，MetaQ1.0在2011年诞生。
2012年，MetaQ已经发展到了3.0版本，并抽象出了通用的消息引擎RocketMQ。随后，对RocketMQ进行了开源，阿里的消息中间件正式走人了公众视野。
2015年，RocketMQ已经经历了多年双十一的洗礼，在可用性、可靠性以及稳定性等方面都有出色的表现。与此同时，云计算大行其道，阿里消息中间件基于RocketMQ推出了AliwareMQ1.0，开始为阿里云上成千上万家企业提供消息服务。
2016年，MetaQ在双十一期间承载了万亿级消息的流转，跨越了一个新的里程碑，同时RocketMQ进入Apache孵化。

1.2 RocketMQ基本术语

1.2.1 message

代表一条消息，使用messageId唯一识别，用户在发送时可以设置messageKey，便于之后查询和跟踪。
一个message必须指定topic，相当于寄信的地址。message还有一个可选的tag设置，以便消费端可以基于tag进行过滤消息。也可以添加额外的键值对，例如你需要一个业务key来查找broker上的消息，方便在开发过程中诊断问题。

1.2.2 topic

topic用于将消息按主题做划分，producer将消息发往指定的topic，consumer订阅该topic就可以收到这条消息。topic跟发送方和消费方都没有强关联关系，发送方可以同时往多个topic投放消息，消费方也可以订阅多个topic的消息。在RocketMQ中，topic是一个上逻辑概念。消息存储不会按topic分开。
topic 表示消息的第一级类型，比如一个电商系统的消息可以分为：交易消息、物流消息等。一条消息必须有一个 topic。topic 是最细粒度的订阅单位，一个 Group 可以订阅多个 topic 的消息。

1.2.3 queue

消息的物理管理单位。一个topic下可以有多个queue，queue的引入使得消息的存储可以分布式集群化，具有了水平扩展能力。
topic和queue是一对多的关系，一个topic下可以包含多个queue，主要用于负载均衡。发送消息时，用户只指定topic，producer会根据topic的路由信息选择具体发到哪个queue上。consumer订阅消息时，会根据负载均衡策略决定订阅哪些queue的消息。
RocketMQ是磁盘消息队列的模式，对于同一个消费组，一个分区只支持一个消费线程来消费消息。过少的分区，会导致消费速度大大落后于消息的生产速度。所以在实际生产环境中，一个topic会设置成多分区的模式，来支持多个消费者，参照下图：

001.PNG

在RocketMQ中，所有消息队列都是持久化，长度无限的数据结构。所谓长度无限是指队列中的每个存储单元都是定长，访问其中的存储单元使用offset来访问，offset为java的long类型，理论上在100年内不会溢出，所以认为是长度无限。 另外队列中只保存最近几天的数据，之前的数据会按照过期时间来删除。
可以认为messagequeue是一个长度无限的数组，offset就是下标。

1.2.4 offset

RocketMQ在存储消息时会为每个topic下的每个queue生成一个消息的索引文件，每个queue都对应一个offset记录当前queue中消息条数

1.2.5 tag

标签可以被认为是对topic进一步细化。一般在相同业务模块中通过引入标签来标记不同用途的消息。tag表示消息的第二级类型，比如交易消息又可以分为：交易创建消息，交易完成消息等。
RocketMQ提供2级消息分类，方便灵活控制。

1.2.6 nameserver

1.2.6.1 基本概念

nameserver是整个消息队列中的状态服务器，集群的各个组件通过它来了解全局的信息 。同时，各个角色的机器都要定期向 nameserver上报自己的状态，超时不上报的话， nameserver会认为某个机器出故障不可用了，其他的组件会把这个机器从可用列表里移除 。
nameserver维护这些配置信息 、 状态信 息，其他角色都通过 nameserver 来协同执行
nameserver可以部署多个，相互之间独立，其他角色同时向多个 nameserver机器上报状态信息，从而达到热备份的目的。 nameserver本身是无状态的，也就是说 nameserver中的 Broker、 Topic 等状态信息不会持久存储，都是由各个角色定时上报并存储到内存中的。

相对来说，nameserver的稳定性非常高，原因有二：
（1）nameserver互相独立，彼此没有通信关系，单台nameserver挂掉，不影响其他nameserver，即使全部挂掉，也不影响业务系统使用。
（2）nameserver不会有频繁的读写，所以性能开销非常小，稳定性很高。

1.2.6.2 nameserver存在意义

002.PNG

服务发现机制：当发出请求服务时，客户端通过注册中心服务知道所有的服务实例。客户端接着使用负载均衡算法选择可用的服务实例中的一个并进行发送。

003.PNG

总结：nameserver是一个几乎无状态的节点，可集群部署，节点之间无任何信息同步。

1.2.7 broker

broker是RocketMQ的核心模块，负责接收并存储消息，同时提供Push/Pull接口来将消息发送给consumer。consumer可选择从master或者slave读取数据。
broker通常都是以集群的方式存在，多个主/从组成broker集群，集群内的master节点之间不做数据交互。
broker同时提供消息查询的功能，可以通过messageID和messageKey来查询消息。
broker会将自己的topic配置信息实时同步到nameserver。
broker部署相对复杂，broker分为master与slave。一个master可以对应多个slave，但是一个slave只能对应一个master。master与slave的对应关系通过指定相同的brokerName不同的brokerId来定义，brokerId为0表示master，非0表示slave。master也可以部署多个。每个broker与nameserver集群中的所有节点建立长连接，定时注册topic信息到所有nameserver。

1.2.8 producer

消息生产者，位于用户的进程内，producer 通过 nameserver 获取所有 broker 的路由信息，根据负载均衡策略选择将消息发到哪个 broker，然后调用 broker 接口提交消息。
producer与nameserver集群中的其中一个节点（随机选择，但不同于上一次）建立长连接，定期从nameserver取topic路由信息，并向提供topic服务的master建立长连接，且定时向master发送心跳。

1.2.9 producerGroup

生产者组，简单来说就是多个发送同一类消息的生产者称之为一个生产者组。

1.2.10 consumer

消息消费者，位于用户进程内。consumer通过nameserver获取所有broker的路由信息后，向broker发送Pull请求来获取消息数据。consumer可以以两种模式启动，广播（Broadcast）和集群（Cluster），广播模式下，一条消息会发送给所有consumer，集群模式下消息只会发送给一个consumer。
consumer与nameserver集群中的其中一个节点（随机选择，但不同于上一次）建立长连接，定期从nameserver取topic路由信息，并向提供topic服务的master、slave建立长连接，且定时向master、slave发送心跳

1.2.16 consumerGroup

消费者组，和生产者类似，消费同一类消息的多个consumer实例组成一个消费者组。

1.3 消息发送

1.3.1 简化流程

一个消息从发送到接收最简单的步骤：producer、topic、consumer。

004.PNG

消息先由生产者发送到topic，然后消费者去topic拿消息。topic在这里只是个概念。

1.3.2 细化流程

详细的消息发送及接收流程：

005.PNG

消息被发送到queue中进行标记：

006.PNG

1.3.3 发送消息的三种方式

从功能上来说，rocketmq支持三种发送消息的方式，分别是同步发送（sync），异步发送（async）和直接发送（oneway）。顺序消息只支持同步发送. 下面来简单说明一下这三种发送消息的方式，以便了解它们之间的差异。

同步发送 sync
发送消息采用同步模式，这种方式只有在消息完全发送完成之后才返回结果，此方式存在需要同步等待发送结果的时间代价。
这种方式具有内部重试机制，即在主动声明本次消息发送失败之前，内部实现将重试一定次数，默认为2次（DefaultMQProducer＃getRetryTimesWhenSendFailed）。 发送的结果存在同一个消息可能被多次发送给给broker，这里需要应用的开发者自己在消费端处理幂等性问题。

public class  {
    public static void main(String[] args) Exception {
        DefaultMQProducer producer = new DefaultMQProducer("producer_demo");
        //指定NameServer地址
        //修改为自己的
        //多个可以用";"隔开
        //producer.setNamesrvAddr("192.168.116.115:9876;192.168.116.116:9876");
        producer.setNamesrvAddr("112.74.43.136:9876");
        /*
         * Producer对象在使用之前必须要调用start初始化，初始化一次即可
         * 注意：切记不可以在每次发送消息时，都调用start方法
         */
        producer.start();
        for (int i = 0; i <= 100; i++) {
                /*
                构建消息
                参数  topic: Message 所属的 Topic
                      tags: 可理解为对消息进行再归类，方便 Consumer 指定过滤条件在 MQ 服务器过滤
                      keys: 设置代表消息的业务关键属性，请尽可能全局唯一,以方便您在无法正常收到消息情况下，可通过阿里云服务器管理控制台查询消息并补发。注意：不设置也不会影响消息正常收发
                      body: body可以是任何二进制形式的数据，MQ不做任何干预，需要 Producer 与 Consumer 协商好一致的序列化和反序列化方式
                 */
            Message msg = new Message("TopicTest", "TagA", "keys", ("测试RocketMQ" + i).getBytes("UTF-8")
            );
            try {
            //发送同步消息
                SendResult sendResult = producer.send(msg);
                System.out.printf("%s%n", sendResult);
            } catch (Exception e) {
                e.printStackTrace();
                // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理
                System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
                Thread.sleep(101);
            }
        }
        producer.shutdown();
    }
}

异步发送 async
发送消息采用异步发送模式，消息发送后立刻返回，当消息完全完成发送后，会调用回调函数sendCallback来告知发送者本次发送是成功或者失败。异步模式通常用于响应时间敏感业务场景，即承受不了同步发送消息时等待返回的耗时代价。
同同步发送一样，异步模式也在内部实现了重试机制，默认次数为2次（DefaultMQProducer#getRetryTimesWhenSendAsyncFailed}）。发送的结果同样存在同一个消息可能被多次发送给给broker，需要应用的开发者自己在消费端处理幂等性问题。

public class AsynProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("producer_demo");
        producer.setNamesrvAddr("112.74.43.136:9876");
        producer.start();
        for (int i = 0; i <= 100; i++) {
            Message msg = new Message("TopicTest", "TagA", "keys", ("测试RocketMQ" + i).getBytes("UTF-8")
            );
            try {
                // 异步发送消息, 发送结果通过 callback 返回给客户端。
                producer.send(msg, new SendCallback() { 
                    public void onSuccess(SendResult sendResult) {
                        // 消费发送成功
                        System.out.println("SUCCESS信息:" + sendResult.toString());
                        System.out.println("send message success. topic=" + sendResult.getRegionId() + ", msgId=" + sendResult.getMsgId());
                    }
                    public void onException(Throwable throwable) {
                        // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理
                        System.out.println("FAIL信息:" + throwable.getMessage());
                    }
                });
            } catch (Exception e) {
                e.printStackTrace();
                // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理
                System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
                Thread.sleep(101);
            }
        }
        producer.shutdown();
    }
}

直接发送 one-way
采用one-way发送模式发送消息的时候，发送端发送完消息后会立即返回，不会等待来自broker的ack来告知本次消息发送是否完全完成发送。这种方式吞吐量很大，但是存在消息丢失的风险，所以其适用于不重要的消息发送，比如日志收集。

public class OneWayProducer {

    public static void main(String[] args) throws MQClientException, InterruptedException, UnsupportedEncodingException {
        DefaultMQProducer producer = new DefaultMQProducer("producer_demo");
        producer.setNamesrvAddr("112.74.43.136:9876");
        producer.start();

        for (int i = 0; i <= 10; i++) {

            Message msg = new Message("TopicTest", "TagA", "keys", ("测试RocketMQ" + i).getBytes("UTF-8")
            );
            try {
                // 由于在 oneway 方式发送消息时没有请求应答处理，一旦出现消息发送失败，则会因为没有重试而导致数据丢失。若数据不可丢，建议选用可靠同步或可靠异步发送方式。
                producer.sendOneway(msg);
            } catch (Exception e) {
                e.printStackTrace();
                // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理
                System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
                Thread.sleep(101);
            }
        }
        producer.shutdown();
    }
}

1.4 消息存储

topic是一个逻辑上的概念，实际上message是在每个broker上以queue的形式记录。

007.PNG

从上面的图片可以总结下几条结论：
（1）消费者发送的message会在broker中的queue队列中记录
（2）一个topic的数据可能会存在多个broker中
（3）一个broker存在多个queue
也就是说每个topic在broker上会划分成几个逻辑队列，每个逻辑队列保存一部分消息数据，但是保存的消息数据实际上不是真正的消息数据，而是指向commitlog的消息索引

1.5 消息消费

1.5.1 广播消费

一条消息被多个Consumer消费，即使这些Consumer属于同一个ConsumerGroup，消息也会被ConsumerGroup中的每个Consumer消费一次，广播消费中的ConsumerGroup概念可以认为在消息划分层面没有意义，适用于一些分发消息的场景，比如我订单下单成功了，需要通知财务系统，客服系统等等这种分发的场景，可以通过修改Consumer中的MessageModel来设置消费方式为广播消费 。consumer.setMessageModel(MessageModel.BROADCASTING)

1.5.2 集群消费

一个ConsumerGroup中的Consumer实例平均分摊消费生产者发送的消息。例如某个Topic有九条消息，其中一个Consumer Group有三个实例（可能是3个进程，或者3台机器），那么每个实例只消费其中的3条消息，Consumer不指定消费方式的话默认是集群消费的，适用于大部分消息的业务 。

1.6 网络架构

008.PNG

009.png

对于上图中几个角色的说明：
（1）nameserver：RocketMQ集群的命名服务器（也可以说是注册中心），它本身是无状态的（实际情况下可能存在每个nameserver实例上的数据有短暂的不一致现象，但是通过定时更新，在大部分情况下都是一致的），用于管理集群的元数据（ 例如，KV配置、topic、broker的注册信息）。
（2）broker（master）：RocketMQ消息代理服务器主节点，起到串联producer的消息发送和consumer的消息消费，和将消息的落盘存储的作用；
（3）broker（slave）：RocketMQ消息代理服务器备份节点，主要是通过同步/异步的方式将主节点的消息同步过来进行备份，为RocketMQ集群的高可用性提供保障；
（4）producer（消息生产者）：在这里为普通消息的生产者，主要基于RocketMQ-Client模块将消息发送至RocketMQ的主节点。
对于上面图中几条通信链路的关系：
（1）producer与NamerServer：每一个producer会与nameserver集群中的一个实例建立TCP连接，从这个nameserver实例上拉取topic路由信息；
（2）producer和broker:producer会和它要发送的topic相关联的master的broker代理服务器建立TCP连接，用于发送消息以及定时的心跳信息；
（3）broker和NamerServer：broker（master or slave）均会和每一个nameserver实例来建立TCP连接。broker在启动的时候会注册自己配置的topic信息到nameserver集群的每一台机器中。
即每一个nameserver均有该broker的topic路由配置信息。其中，master与master之间无连接，master与slave之间有连接；

2 深入RocketMQ

2.1 发送消息负载均衡

010.png

发送消息通过轮询队列的方式发送，每个队列接收平均的消息量。通过增加机器，可以水平扩展队列容量。另外也可以自定义方式选择发往哪个队列。注：另外多个队列可以部署在一台机器上，也可以分别部署在多台不同的机器上。

2.1.1 消息发送时选择queue的算法

分为两种，一种是直接发消息，client内部有选择queue的算法，不允许外界改变。还有一种是可以自定义queue的选择算法（内置了三种算法，不喜欢的话可以自定义算法实现）

public class org.apache.rocketmq.client.producer.DefaultMQProducer {
    // 只发送消息，queue的选择由默认的算法来实现
    @Override
    public SendResult send(Collection<Message> msgs) {}
    // 自定义选择queue的算法进行发消息
    @Override
    public SendResult send(Collection<Message> msgs, MessageQueue messageQueue) {}
}

2.1.1.1 send(msg, mq)

2.1.1.1.1 使用场景

有时候我们不希望默认的queue选择算法，而是需要自定义，一般最常用的场景在顺序消息，顺序消息的发送一般都会指定某组特征的消息都发当同一个queue里，这样才能保证顺序，因为单queue是有序的。

2.1.1.1.2 原理剖析

内置了SelectMessageQueueByRandom、SelectMessageQueueByHash、SelectMessageQueueByMachineRoom三种算法，都实现了一个共同的接口：org.apache.rocketmq.client.producer.MessageQueueSelector.要想自定义逻辑的话，直接实现接口重写select方法即可。
很典型的策略模式，不同算法不同实现类，有个顶层接口。

SelectMessageQueueByRandom

public class SelectMessageQueueByRandom implements MessageQueueSelector {
    private Random random = new Random(System.currentTimeMillis());
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        // mqs.size()：队列的个数。假设队列个数是4，那么这个value就是0-3之间随机。
        int value = random.nextInt(mqs.size());
        return mqs.get(value);
    }
}

SelectMessageQueueByHash

public class SelectMessageQueueByHash implements MessageQueueSelector {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        int value = arg.hashCode();
        // 防止出现负数，取个绝对值，这也是我们平时开发中需要注意到的点
        if (value < 0) {
            value = Math.abs(value);
        }
        // 直接取余队列个数。
        value = value % mqs.size();
        return mqs.get(value);
    }
}

SelectMessageQueueByMachineRoom

public class SelectMessageQueueByMachineRoom implements MessageQueueSelector {
    private Set<String> consumeridcs;
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        return null;
    }
    public Set<String> getConsumeridcs() {
        return consumeridcs;
    }
    public void setConsumeridcs(Set<String> consumeridcs) {
        this.consumeridcs = consumeridcs;
    }
}

自定义算法

public class MySelectMessageQueue implements MessageQueueSelector {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        return mqs.get(0);
    }
}

2.1.1.2 send(msg)

2.1.1.2.1 使用场景

一般没特殊需求的场景都用这个。因为他默认的queue选择算法很不错，各种优化场景都替我们想到了。我们称之为随机递增取模算法。

2.1.1.2.2 原理剖析

// {@link org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#sendDefaultImpl}
// 这是发送消息核心原理

// 选择消息要发送的队列
MessageQueue mq = null;
for (int times = 0; times < 3; times++) {
    // 首次肯定是null
    String lastBrokerName = null == mq ? null : mq.getBrokerName();
    // 调用下面的方法进行选择queue
    MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
    if (mqSelected != null) {
        // 给mq赋值，如果首次失败了，那么下次重试的时候（也就是下次for的时候），mq就有值了。
        mq = mqSelected;
        ......
        // 很关键，能解答下面会提到的两个问题：
        // 1.faultItemTable是什么时候放进去的？
        // 2.isAvailable() 为什么只是判断一个时间就可以知道Broker是否可用？   
        this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);    
    }
}

选择queue的主入口

public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
    // 默认为false，代表不启用broker故障延迟
    if (this.sendLatencyFaultEnable) {
        try {
            // 随机数且+1
            int index = tpInfo.getSendWhichQueue().getAndIncrement();
            // 遍历
            for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
                // 先（随机数 +1） % queue.size()
                int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
                if (pos < 0) {
                    pos = 0;
                }
                MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
                // 看找到的这个queue所属的broker是不是可用的
                if (latencyFaultTolerance.isAvailable(mq.getBrokerName())) {
                    // 非失败重试，直接返回到的队列
                    // 失败重试的情况，如果和选择的队列是上次重试是一样的，则返回
                    
                    // 也就是说如果你这个queue所在的broker可用，
                    // 且不是重试进来的或失败重试的情况，如果和选择的队列是上次重试是一样的，那你就是天选之子了。
                    if (null == lastBrokerName || mq.getBrokerName().equals(lastBrokerName)) {
                        return mq;
                    }
                }
            }
            
   // 如果所有队列都不可用，那么选择一个相对好的broker，不考虑可用性的消息队列
            final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
            int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
            if (writeQueueNums > 0) {
                final MessageQueue mq = tpInfo.selectOneMessageQueue();
                if (notBestBroker != null) {
                    mq.setBrokerName(notBestBroker);
                    mq.setQueueId(tpInfo.getSendWhichQueue().getAndIncrement() % writeQueueNums);
                }
                return mq;
            } else {
                latencyFaultTolerance.remove(notBestBroker);
            }
        } catch (Exception e) {
            log.error("Error occurred when selecting message queue", e);
        }
  // 随机选择一个queue
        return tpInfo.selectOneMessageQueue();
    }
 // 当sendLatencyFaultEnable=false的时候选择queue的方法，默认就是false。
    return tpInfo.selectOneMessageQueue(lastBrokerName);
}

2.1.1.2.2.1 不启用broker故障延迟

既然sendLatencyFaultEnable默认是false，那就先看当sendLatencyFaultEnable=false时候的逻辑

public MessageQueue selectOneMessageQueue(final String lastBrokerName) {
    // 第一次就是null，第二次（也就是重试的时候）就不是null了。
    if (lastBrokerName == null) {
        // 第一次选择队列的逻辑
        return selectOneMessageQueue();
    } else {
        // 第一次选择队列发送消息失败了，第二次重试的时候选择队列的逻辑
        int index = this.sendWhichQueue.getAndIncrement();
        for (int i = 0; i < this.messageQueueList.size(); i++) {
            int pos = Math.abs(index++) % this.messageQueueList.size();
            if (pos < 0)
                pos = 0;
            MessageQueue mq = this.messageQueueList.get(pos);
   // 过滤掉上次发送消息失败的队列
            if (!mq.getBrokerName().equals(lastBrokerName)) {
                return mq;
            }
        }
        return selectOneMessageQueue();
    }
}

那就继续看第一次选择队列的逻辑：

public MessageQueue selectOneMessageQueue() {
    // 当前线程有个ThreadLocal变量，存放了一个随机数 
    // {@link org.apache.rocketmq.client.common.ThreadLocalIndex#getAndIncrement}
    // 然后取出随机数根据队列长度取模且将随机数+1
    int index = this.sendWhichQueue.getAndIncrement();
    int pos = Math.abs(index) % this.messageQueueList.size();
    if (pos < 0) {
        pos = 0;
    }
    return this.messageQueueList.get(pos);
}

其实也有点随机一个的意思。但是亮点在于取出随机数根据队列长度取模且将随机数+1，（getAndIncrement cas +1）。

当消息第一次发送失败时，lastBrokerName会存放当前选择失败的broker（mq = mqSelected），通过重试，此时lastBrokerName有值，代表上次选择的boker发送失败，则重新对sendWhichQueue本地线程变量+1，遍历选择消息队列，直到不是上次的broker，也就是为了规避上次发送失败的broker的逻辑所在。

举个例子：你这次随机数是1，队列长度是4，1%4=1，这时候失败了，进入重试，那么重试之前，也就是在上一步1%4之后，他把1进行了++操作，变成了2，那么你这次重试的时候就是2%4=2，直接过滤掉了刚才失败的broker。

那就继续看第二次重试选择队列的逻辑：

// +1
int index = this.sendWhichQueue.getAndIncrement();
for (int i = 0; i < this.messageQueueList.size(); i++) {
    // 取模
    int pos = Math.abs(index++) % this.messageQueueList.size();
    if (pos < 0)
        pos = 0;
    MessageQueue mq = this.messageQueueList.get(pos);
    // 过滤掉上次发送消息失败的队列
    if (!mq.getBrokerName().equals(lastBrokerName)) {
        return mq;
    }
}
// 没找到能用的queue的话继续走默认的那个
return selectOneMessageQueue();

2.1.1.2.2.2 启用broker故障延迟

也就是下面if里的逻辑

if (this.sendLatencyFaultEnable) {
    ....
}

我先（随机数 +1） % queue.size()，然后看你这个queue所属的broker是否可用，可用的话且不是重试进来的或失败重试的情况，如果和选择的队列是上次重试是一样的，那直接return你就完事了。那么怎么看broker是否可用的呢？

// {@link org.apache.rocketmq.client.latency.LatencyFaultToleranceImpl#isAvailable(String)}
public boolean isAvailable(final String name) {
    final FaultItem faultItem = this.faultItemTable.get(name);
    if (faultItem != null) {
        return faultItem.isAvailable();
    }
    return true;
}

// {@link org.apache.rocketmq.client.latency.LatencyFaultToleranceImpl.FaultItem#isAvailable()}
public boolean isAvailable() {
    return (System.currentTimeMillis() - startTimestamp) >= 0;
}

faultItemTable是什么时候放进去的？isAvailable() 为什么只是判断一个时间就可以知道Broker是否可用？这就需要上面发送消息完成后所调用的这个方法了：

// {@link org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#updateFaultItem}
// 发送开始时间
beginTimestampPrev = System.currentTimeMillis();
// 进行发送
sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout);
// 发送结束时间
endTimestamp = System.currentTimeMillis();
// 更新broker的延迟情况
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);

细节逻辑如下：

// {@link org.apache.rocketmq.client.latency.MQFaultStrategy#updateFaultItem}
public void updateFaultItem(final String brokerName, final long currentLatency, boolean isolation) {
    if (this.sendLatencyFaultEnable) {
        // 首次isolation传入的是false，currentLatency是发送消息所耗费的时间，如下
        // this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
        long duration = computeNotAvailableDuration(isolation ? 3010 : currentLatency);
        this.latencyFaultTolerance.updateFaultItem(brokerName, currentLatency, duration);
    }
}

private long[] latencyMax = {50L, 100L, 550L, 101L, 201L, 301L, 1501L};
private long[] notAvailableDuration = {0L, 0L, 3010L, 6010L, 12010L, 18010L, 60100L};

// 根据延迟时间对比MQFaultStrategy中的延迟级别数组latencyMax 不可用时长数组notAvailableDuration 来将该broker加进faultItemTable中。
private long computeNotAvailableDuration(final long currentLatency) {
    for (int i = latencyMax.length - 1; i >= 0; i--) {
        // 假设currentLatency花费了10ms，那么latencyMax里的数据显然不符合下面的所有判断，所以直接return 0;
        if (currentLatency >= latencyMax[i])
            return this.notAvailableDuration[i];
    }
    return 0;
}

// {@link org.apache.rocketmq.client.latency.LatencyFaultToleranceImpl#updateFaultItem()}
@Override
// 其实主要就是给startTimestamp赋值为当前时间+computeNotAvailableDuration(isolation ? 3010 : currentLatency);的结果，给isAvailable()所用
// 也就是说只有notAvailableDuration == 0的时候，isAvailable()才会返回true。
public void updateFaultItem(final String name, final long currentLatency, final long notAvailableDuration) {
    FaultItem old = this.faultItemTable.get(name);
    if (null == old) {
        final FaultItem faultItem = new FaultItem(name);
        faultItem.setCurrentLatency(currentLatency);
        // 给startTimestamp赋值为当前时间+computeNotAvailableDuration(isolation ? 3010 : currentLatency);的结果，给isAvailable()所用
        faultItem.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);

        old = this.faultItemTable.putIfAbsent(name, faultItem);
        if (old != null) {
            old.setCurrentLatency(currentLatency);
            // 给startTimestamp赋值为当前时间+computeNotAvailableDuration(isolation ? 3010 : currentLatency);的结果，给isAvailable()所用
            old.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);
        }
    } else {
        old.setCurrentLatency(currentLatency);
        // 给startTimestamp赋值为当前时间+computeNotAvailableDuration(isolation ? 3010 : currentLatency);的结果，给isAvailable()所用
        old.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);
    }
}

RocketMQ为每个Broker预测了个可用时间(当前时间+notAvailableDuration)，当当前时间大于该时间，才代表Broker可用，而notAvailableDuration有6个级别和latencyMax的区间一一对应，根据传入的currentLatency去预测该Broker在什么时候可用。所以再来看这个

public boolean isAvailable() {
    return (System.currentTimeMillis() - startTimestamp) >= 0;
}

根据执行时间来看落入哪个区间，在0~100的时间内notAvailableDuration都是0，都是可用的，大于该值后，可用的时间就会开始变大了，就认为不是最优解，直接舍弃。

2.1.1.3 总结

在不开启容错的情况下，轮询队列进行发送，如果失败了，重试的时候过滤失败的Broker
如果开启了容错策略，会通过RocketMQ的预测机制来预测一个Broker是否可用
如果上次失败的Broker可用那么还是会选择该Broker的队列
如果上述情况失败，则随机选择一个进行发送
在发送消息的时候会记录一下调用的时间与是否报错，根据该时间去预测broker的可用时间

2.2 消息存储

2.2.1 存储模型

011.png

RocketMQ文件存储模型层次结构如上图所示，根据类别和作用从概念模型上大致可以划分为5层，下面将从各个层次分别进行分析和阐述：
（1）RocketMQ业务处理器层：Broker端对消息进行读取和写入的业务逻辑入口，这一层主要包含了业务逻辑相关处理操作（根据解析RemotingCommand中的RequestCode来区分具体的业务操作类型，进而执行不同的业务处理流程），比如前置的检查和校验步骤、构造MessageExtBrokerInner对象、decode反序列化、构造Response返回对象等；
（2）RocketMQ数据存储组件层；该层主要是RocketMQ的存储核心类—DefaultMessageStore，其为RocketMQ消息数据文件的访问入口，通过该类的“putMessage()”和“getMessage()”方法完成对CommitLog消息存储的日志数据文件进行读写操作（具体的读写访问操作还是依赖下一层中CommitLog对象模型提供的方法）；另外，在该组件初始化时候，还会启动很多存储相关的后台服务线程，包括AllocateMappedFileService（MappedFile预分配服务线程）、ReputMessageService（回放存储消息服务线程）、HAService（Broker主从同步高可用服务线程）、StoreStatsService（消息存储统计服务线程）、IndexService（索引文件服务线程）等；
（3）RocketMQ存储逻辑对象层：该层主要包含了RocketMQ数据文件存储直接相关的三个模型类IndexFile、ConsumerQueue和CommitLog。IndexFile为索引数据文件提供访问服务，ConsumerQueue为逻辑消息队列提供访问服务，CommitLog则为消息存储的日志数据文件提供访问服务。这三个模型类也是构成了RocketMQ存储层的整体结构（对于这三个模型类的深入分析将放在后续篇幅中）；
（4）封装的文件内存映射层：RocketMQ主要采用JDK NIO中的MappedByteBuffer和FileChannel两种方式完成数据文件的读写。其中，采用MappedByteBuffer这种内存映射磁盘文件的方式完成对大文件的读写，在RocketMQ中将该类封装成MappedFile类。这里限制的问题在上面已经讲过；对于每类大文件（IndexFile/ConsumerQueue/CommitLog），在存储时分隔成多个固定大小的文件（单个IndexFile文件大小约为400M、单个ConsumerQueue文件大小约5.72M、单个CommitLog文件大小为1G），其中每个分隔文件的文件名为前面所有文件的字节大小数+1，即为文件的起始偏移量，从而实现了整个大文件的串联。这里，每一种类的单个文件均由MappedFile类提供读写操作服务（其中，MappedFile类提供了顺序写/随机读、内存数据刷盘、内存清理等和文件相关的服务）；
（5）磁盘存储层：主要指的是部署RocketMQ服务器所用的磁盘。这里，需要考虑不同磁盘类型（如SSD或者普通的HDD）特性以及磁盘的性能参数（如IOPS、吞吐量和访问时延等指标）对顺序写/随机读操作带来的影响

2.2.2 存储流程

012.png

（1）RocketMQ消息存储结构类型及缺点
上图即为RocketMQ的消息存储整体架构，RocketMQ采用的是混合型的存储结构，即为Broker单个实例下所有的队列共用一个日志数据文件（即为CommitLog）来存储。而Kafka采用的是独立型的存储结构，每个队列一个文件。这里小编认为，RocketMQ采用混合型存储结构的缺点在于，会存在较多的随机读操作，因此读的效率偏低。同时消费消息需要依赖ConsumeQueue，构建该逻辑消费队列需要一定开销。

（2）RocketMQ消息存储架构深入分析
从上面的整体架构图中可见，RocketMQ的混合型存储结构针对Producer和Consumer分别采用了数据和索引部分相分离的存储结构，Producer发送消息至Broker端，然后Broker端使用同步或者异步的方式对消息刷盘持久化，保存至CommitLog中。只要消息被刷盘持久化至磁盘文件CommitLog中，那么Producer发送的消息就不会丢失。正因为如此，Consumer也就肯定有机会去消费这条消息，至于消费的时间可以稍微滞后一些也没有太大的关系。退一步地讲，即使Consumer端第一次没法拉取到待消费的消息，Broker服务端也能够通过长轮询机制等待一定时间延迟后再次发起拉取消息的请求。
这里，RocketMQ的具体做法是，使用Broker端的后台服务线程—ReputMessageService不停地分发请求并异步构建ConsumeQueue（逻辑消费队列）和IndexFile（索引文件）数据。然后，Consumer即可根据ConsumerQueue来查找待消费的消息了。其中，ConsumeQueue（逻辑消费队列）作为消费消息的索引，保存了指定Topic下的队列消息在CommitLog中的起始物理偏移量offset，消息大小size和消息Tag的HashCode值。而IndexFile（索引文件）则只是为了消息查询提供了一种通过key或时间区间来查询消息的方法（ps：这种通过IndexFile来查找消息的方法不影响发送与消费消息的主流程）。

（3）PageCache与Mmap内存映射
这里有必要先稍微简单地介绍下page cache的概念。系统的所有文件I/O请求，操作系统都是通过page cache机制实现的。对于操作系统来说，磁盘文件都是由一系列的数据块顺序组成，数据块的大小由操作系统本身而决定，x86的linux中一个标准页面大小是4KB。
操作系统内核在处理文件I/O请求时，首先到page cache中查找（page cache中的每一个数据块都设置了文件以及偏移量地址信息），如果未命中，则启动磁盘I/O，将磁盘文件中的数据块加载到page cache中的一个空闲块，然后再copy到用户缓冲区中。
page cache本身也会对数据文件进行预读取，对于每个文件的第一个读请求操作，系统在读入所请求页面的同时会读入紧随其后的少数几个页面。因此，想要提高page cache的命中率（尽量让访问的页在物理内存中），从硬件的角度来说肯定是物理内存越大越好。从操作系统层面来说，访问page cache时，即使只访问1k的消息，系统也会提前预读取更多的数据，在下次读取消息时, 就很可能可以命中内存。
在RocketMQ中，ConsumeQueue逻辑消费队列存储的数据较少，并且是顺序读取，在page cache机制的预读取作用下，Consume Queue的读性能会比较高近乎内存，即使在有消息堆积情况下也不会影响性能。而对于CommitLog消息存储的日志数据文件来说，读取消息内容时候会产生较多的随机访问读取，严重影响性能。如果选择合适的系统IO调度算法，比如设置调度算法为“Noop”（此时块存储采用SSD的话），随机读的性能也会有所提升。
另外，RocketMQ主要通过MappedByteBuffer对文件进行读写操作。其中，利用了NIO中的FileChannel模型直接将磁盘上的物理文件直接映射到用户态的内存地址中（这种Mmap的方式减少了传统IO将磁盘文件数据在操作系统内核地址空间的缓冲区和用户应用程序地址空间的缓冲区之间来回进行拷贝的性能开销），将对文件的操作转化为直接对内存地址进行操作，从而极大地提高了文件的读写效率（这里需要注意的是，采用MappedByteBuffer这种内存映射的方式有几个限制，其中之一是一次只能映射1.5~2G 的文件至用户态的虚拟内存，这也是为何RocketMQ默认设置单个CommitLog日志数据文件为1G的原因了）。

2.3 RocketMQ消费者端消息列队六种负载均衡算法

在RocketMQ启动的时候会启动负载均衡线程，过程如下：

//DefaultMQPullConsumerImpl.start()
mQClientFactory.start();
//上面点进去 ->MQClientInstance.start()，rebalanceService继承了ServiceThread，
//ServiceThread实现了Runnable接口
this.rebalanceService.start();
//继续下一层，MQClientInstance.doRebalance()找到下面
impl.doRebalance();
//..在一层层点进去，最后找到RebalanceImpl.rebalanceByTopic方法,找到
AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;

AllocateMessageQueueStrategy就是实现消费者消息队列负载均衡算法的接口。
该接口在rocketMq-4.3.0版本中有六种实现方法：
AllocateMessageQueueAveragely：平均算法
AllocateMessageQueueAveragelyByCircle：环形平均算法
AllocateMessageQueueByConfig：根据配置负载均衡算法
AllocateMessageQueueByMachineRoom：根据机房负载均衡算法
AllocateMessageQueueConsistentHash：一致性哈希负载均衡算法
AllocateMachineRoomNearby：靠近机房策略
在客户端没有指定的情况下，RocketMQ默认使用AllocateMessageQueueAveragely平均算法。

2.3.1 AllocateMessageQueueAveragely 平均负载均衡算法

平均算法顾名思义就是取平均值，该方法四个参数，consumerGroup（消费者组名称）、
currentCID（当前消费者的id）、mqAll（当前topic下面所有的消息队列）、cidAll（当前消费者组下面所有的消费者id）。算法思想就是把算出平均值然后将连续的队列分配给每个消费者。假设队列大小是8（编号0-7），消费者数量3（编号0-2），分配结果就是：消费者0：队列0，1，2；消费者1：队列3，4，5；消费者2：队列6，7。

2.3.2 AllocateMessageQueueAveragelyByCircle 环形平均分配算法

环形分配就可以看成：所有消费者围成一个环，然后循环这个环分配队列。AllocateMessageQueueAveragely方法平均分配的是连续的队列，环形分配的就是间隔的队列。核心代码就一个for循环，也很好理解。假设mq8个，消费者3个，分配后的结果就是{0，3，6}，{1，4，7}，{2，5}。

013.PNG

2.3.3 AllocateMessageQueueByMachineRoom 机房分配算法

机房分配现根据MQ中的brokerName找出有效的机房信息（也就是消息队列），然后再平分。这个算法的逻辑是先算出平均值和余数，它和AllocateMessageQueueAveragely平均算法的不同在于，它是先给每个消费者分配mod（平均值个数）个消息队列，然后余数在从头开始一个个分配。假设mq有8个，消费者3个，那么平均值mod = 2，余数2，分配方式就是每个消费者先分配两个mq，{0，1}，{2，3}，{4，5}，然后余数2个在从头开始分配，最后就是{0，1，6}，{2，3，7}，{4，5}。

2.3.4 AllocateMessageQueueConsistentHash 一致性哈希负载均衡算法

一致性哈希负载均衡的目的是要保证相同的请求尽可能落在同一个服务器上。为什么是说尽可能？因为服务器会发生上下线，在少数服务器变化的时候不应该影响大多数的请求。

普通hash算法存在的问题
普通hash算法我们可以简单理解为对key值进行hash之后对服务器取模，也就是hash(key) % n。这个时候如果我们的一台服务器宕机或者需要新增一台服务器，那么我们的n值就会变更，这样就会导致我们所有的请求都会变更。举个简单的例子，我们有个redis集群，部署了4台服务器，如果我们将key1使用随机存储，那么我们找key1的时候可能就需要遍历4服务器，效率差。换种方式，对key1哈希操作后取模，将它定位到一台服务器上，这样在查找key1的时候我们就可以很快的定位到一台服务器上。可是这样还有种问题就是之前所说的，如果我们redis集群增加了一台服务器或者有一台服务器宕机，这样再通过哈希算法算出的值就发生了变化，短时间发生缓存雪崩。

一致性hash算法
哈希环：刚才的hash算法是对服务器取模，一致性哈希算法使用的是对2^{32取模，就是一致性哈希将整个hash空间组织成了一个圆环，0-2}32-1。
物理节点：将服务器（ip+端口）进行hash，映射成环上的一个节点。当请求到来时，根据请求的key，hash映射到环上，顺时针选取最近的一个服务器进行请求。
虚拟节点：当环上的服务器较少的时候，会出现分配不均匀的情况，即大量的请求落在同一台服务器上，为了避免这种情况，就引入了虚拟节点。比如通过添加后缀的方式给物理节点克隆出三个虚拟节点，如果两台物理节点，都克隆三个虚拟节点，那么环上就一共有8个节点。只是被克隆的虚拟节点最后还是会定位到实际物理节点上，但是可以有效的分摊请求。

一致性哈希相对于普通hash，优点在于映射到环上的其请求，是发送到环上离他最近的一个服务器，如果我们一台服务器宕机或者新增一台服务器，那么影响的请求只有这台服务器和前一个服务器节点之间的请求，其他的并不会影响。

2.3.5 AllocateMessageQueueByConfig 通过配置负载均衡

这个没什么好说的，自定义配置。

2.3.6 AllocateMachineRoomNearby 靠近机房策略

014.PNG

2.3 顺序消息

2.3.1 什么是顺序消息

顺序消息（FIFO 消息）是 MQ 提供的一种严格按照顺序进行发布和消费的消息类型。顺序消息由两个部分组成：顺序发布和顺序消费。
顺序消息包含两种类型：
分区顺序：一个Partition内所有的消息按照先进先出的顺序进行发布和消费
全局顺序：一个topic内所有的消息按照先进先出的顺序进行发布和消费

这是阿里云上对顺序消息的定义，把顺序消息拆分成了顺序发布和顺序消费。

那么多线程中发送消息算不算顺序发布？
多线程中若没有因果关系则没有顺序。那么用户在多线程中去发消息就意味着用户不关心那些在不同线程中被发送的消息的顺序。即多线程发送的消息，不同线程间的消息不是顺序发布的，同一线程的消息是顺序发布的。这是需要用户自己去保障的
而对于顺序消费，则需要保证哪些来自同一个发送线程的消息在消费时是按照相同的顺序被处理的（为什么不说他们应该在一个线程中被消费呢？）
全局顺序其实是分区顺序的一个特例，即使topic只有一个分区（以下不在讨论全局顺序，因为全局顺 序将面临性能的问题，而且绝大多数场景都不需要全局顺序）

2.3.2 如何保证顺序

在MQ的模型中，顺序需要由3个阶段去保障：
1.消息被发送时保持顺序
2.消息被存储时保持和发送的顺序一致
3.消息被消费时保持和存储的顺序一致
发送时保持顺序意味着对于有顺序要求的消息，用户应该在同一个线程中采用同步的方式发送。存储保持和发送的顺序一致则要求在同一线程中被发送出来的消息A和B，存储时在空间上A一定在B之前。而消费保持和存储一致则要求消息A、B到达消费者之后必须按照先A后B的顺序被处理。

015.png

对于两个订单的消息的原始数据：a1、b1、b2、a2、a3、b3（绝对时间下发生的顺序）：
在发送时，a订单的消息需要保持a1、a2、a3的顺序，b订单的消息也相同，但是a、b订单之间的消息没有顺序关系，这意味着a、b订单的消息可以在不同的线程中被发送出去
在存储时，需要分别保证a、b订单的消息的顺序，但是a、b订单之间的消息的顺序可以不保证
a1、b1、b2、a2、a3、b3是可以接受的
a1、a2、b1、b2、a3、b3是可以接受的
a1、a3、b1、b2、a2、b3是不可以接受的
消费时保证顺序的简单方式就是“什么都不做”，不对收到的消息的顺序进行调整，即只要一个分区的消息只由一个线程处理即可；当然，如果a、b在一个分区中，在收到消息后也可以将他们拆分到不同线程中处理，不过要权衡一下收益

2.3.3 RocketMQ顺序消息实现

016.png

上图是RocketMQ顺序消息原理的介绍，将不同订单的消息路由到不同的分区中。文档只是给出了Producer顺序的处理，Consumer消费时通过一个分区只能有一个线程消费的方式来保证消息顺序，具体实现如下。

Producer端
Producer端确保消息顺序唯一要做的事情就是将消息路由到特定的分区，在RocketMQ中，通过MessageQueueSelector来实现分区的选择。

017.png

• List<MessageQueue> mqs：消息要发送的Topic下所有的分区

• Message msg：消息对象

• 额外的参数：用户可以传递自己的参数

  比如如下实现就可以保证相同的订单的消息被路由到相同的分区：

long orderId = ((Order) object).getOrderId;
return mqs.get(orderId % mqs.size());

Consumer端
RocketMQ消费端有两种类型：MQPullConsumer和MQPushConsumer。
MQPullConsumer由用户控制线程，主动从服务端获取消息，每次获取到的是一个MessageQueue中的消息。PullResult中的List msgFoundList自然和存储顺序一致，用户需要再拿到这批消息后自己保证消费的顺序。
对于MQPushConsumer，由用户注册MessageListener来消费消息，在客户端中需要保证调用MessageListener时消息的顺序性。RocketMQ中的实现如下：

018.png

(1) PullMessageService单线程的从Broker获取消息
(2) PullMessageService将消息添加到ProcessQueue中（ProcessMessage是一个消息的缓存），之后提交一个消费任务到ConsumeMessageOrderlyService
(3) ConsumeMessageOrderlyService多线程执行，每个线程在消费消息时需要拿到MessageQueue的锁
(4) 拿到锁之后从ProcessQueue中获取消息

保证消费顺序的核心思想是：
(1) 获取到消息后添加到ProcessQueue中，单线程执行，所以ProcessQueue中的消息是顺序的
(2) 提交的消费任务时提交的是“对某个MQ进行一次消费”，这次消费请求是从ProcessQueue中获取消息消费，所以也是顺序的（无论哪个线程获取到锁，都是按照ProcessQueue中消息的顺序进行消费）

顺序和异常的关系

顺序消息需要Producer和Consumer都保证顺序。Producer需要保证消息被路由到正确的分区，消息需要保证每个分区的数据只有一个线程消息，那么就会有一些缺陷：
(1) 发送顺序消息无法利用集群的Failover特性，因为不能更换MessageQueue进行重试
(2) 因为发送的路由策略导致的热点问题，可能某一些MessageQueue的数据量特别大
(3) 消费的并行读依赖于分区数量
(4) 消费失败时无法跳过

不能更换MessageQueue重试就需要MessageQueue有自己的副本，通过Raft、Paxos之类的算法保证有可用的副本，或者通过其他高可用的存储设备来存储MessageQueue。
热点问题好像没有什么好的解决办法，只能通过拆分MessageQueue和优化路由方法来尽量均衡的将消息分配到不同的MessageQueue。
消费并行读理论上不会有太大问题，因为MessageQueue的数量可以调整。
消费失败的无法跳过是不可避免的，因为跳过可能导致后续的数据处理都是错误的。不过可以提供一些策略，由用户根据错误类型来决定是否跳过，并且提供重试队列之类的功能，在跳过之后用户可以在“其他”地方重新消费到这条消息。

2.4 消息去重

造成消息重复的根本原因是网络不可达。只要通过网络交换数据，就无法避免这个问题。所以解决这个问题的办法就是绕过这个问题。那么问题就变成了：如果消费端收到两条一样的消息，应该怎样处理？
1.消费端处理消息的业务逻辑保持幂等性
2.保证每条消息都有唯一编号且保证消息处理成功与去重表的日志同时出现
第1条很好理解，只要保持幂等性，不管来多少条重复消息，最后处理的结果都一样。第2条原理就是利用一张日志表来记录已经处理成功的消息的ID，如果新到的消息ID已经在日志表中，那么就不再处理这条消息。
第1条解决方案，很明显应该在消费端实现，不属于消息系统要实现的功能。第2条可以消息系统实现，也可以业务端实现。正常情况下出现重复消息的概率其实很小，如果由消息系统来实现的话，肯定会对消息系统的吞吐量和高可用有影响，所以最好还是由业务端自己处理消息重复的问题，这也是RocketMQ不解决消息重复的问题的原因。

RocketMQ不保证消息不重复，如果你的业务需要保证严格的不重复消息，需要你自己在业务端去重。
那么msgID记录在哪里呢？当然是缓存。具体做法如下：
• 消费端接收到消息的时候，调用redis提供的incr方法，以msgID作为key（具有唯一性），value则默认从1开始递增。
• 当incr返回值为1时，设置其失效时间为两分钟以后，并且该消息需要被消费。
• 当incr返回值大于1时，则忽略该消息。

public long incr(String key,Date expireTime){
    long count = redisNumber.incre(key);
    if(count==1){
        redisCache.setExpireTime(key,expireTime);
    }
    return count;
}

for(MsgExt msg: msgs){
    long currentTime = System.currentTimeMillis();
    currentTime += Constants.MSG_EXPIRES_TIME_MILLS;
    Date expireTime = new Date(currentTime);
    long msgIDCount = redisCacheHelper.incr(msg.getKeys(),expireTime);
    if(msgIDCount>1){
        continue
     }
     ...
}

2.5 消息堆积

消息中间件的主要功能是异步解耦，还有个重要功能是挡住前端的数据洪峰，保证后端系统的稳定性，这就要求消息中间件具有一定的消息堆积能力。消息堆积分以下两种情况：
消息堆积在内存Buffer，一旦超过内存Buffer，可以根据一定的丢弃策略来丢弃消息，如CORBA Notification规范中描述。适合能容忍丢弃消息的业务，这种情况消息的堆积能力主要在于内存Buffer大小，而且消息堆积后，性能下降不会太大，因为内存中数据多少对于对外提供的访问能力影响有限。
消息堆积到持久化存储系统中，例如DB，KV存储，文件记录形式。当消息不能在内存Cache命中时，要不可避免的访问磁盘，会产生大量读IO，读IO的吞吐量直接决定了消息堆积后的访问能力。

评估消息堆积能力主要有以下四点:
• 消息能堆积多少条，多少字节?即消息的堆积容量。
• 消息堆积后，发消息的吞吐量大小，是否会受堆积影响。
• 消息堆积后，正常消费的Consumer是否会受影响。
• 消息堆积后，访问堆积在磁盘的消息时，吞吐量有多大。

2.5.1 消息积压的生产故障的处理

首先要找到是什么原因导致的消息堆积，是Producer太多了，Consumer太少了导致的还是说其他情况，总之先定位问题。然后看下消息消费速度是否正常，正常的话，可以通过上线更多consumer临时解决消息堆积问题。

2.5.1.1 提高消费并行度

绝大部分消息消费行为都属于 IO 密集型，即可能是操作数据库，或者调用 RPC，这类消费行为的消费速度在于后端数据库或者外系统的吞吐量，通过增加消费并行度，可以提高总的消费吞吐量，但是并行度增加到一定程度，反而会下降。所以，应用必须要设置合理的并行度。 如下有几种修改消费并行度的方法：

同一个 ConsumerGroup 下，通过增加 Consumer 实例数量来提高并行度（需要注意的是超过订阅队列数的 Consumer 实例无效）。可以通过加机器，或者在已有机器启动多个进程的方式。 提高单个 Consumer 的消费并行线程，通过修改参数 consumeThreadMin、consumeThreadMax 实现。

2.5.1.2 批量方式消费

某些业务流程如果支持批量方式消费，则可以很大程度上提高消费吞吐量，例如订单扣款类应用，一次处理一个订单耗时 1 s，一次处理 10 个订单可能也只耗时 2 s，这样即可大幅度提高消费的吞吐量，通过设置 consumer 的 consumeMessageBatchMaxSize 返个参数，默认是 1，即一次只消费一条消息，例如设置为 N，那么每次消费的消息数小于等于 N。

2.5.1.3 跳过非重要消息

发生消息堆积时，如果消费速度一直追不上发送速度，可以选择丢弃不重要的消息

public ConsumeConcurrentlyStatus consumemessage
    (List<messageExt> msgs,ConsumeConcurrentlyContext context){ 
    long offset = msgs.get(0).getqueueOffset(); 
    String maxOffset = msgs.get(0).getProperty(message.PROPERTY_MAX_OFFSET); 
    long diff= Long.parseLong(max0ffset)- offset; 
    if (diff > 10100) {//消息堆积情况的特殊处理
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
    //正常消费过程
    return ConsumeConcurrentlyStatus.COMSUME_SUCCESS;
}

如以上代码所示，当某个队列的消息数堆积到 10100 条以上，则尝试丢弃部分戒全部消息，返样就可以快速追上収送消息的速度。

如果Consumer和Queue不对等，上线了多台也在短时间内无法消费完堆积的消息怎么办？
• 准备一个临时的topic，newTopic
• queue的数量是原先queue的几倍
• queue分布到多个Broker中
• 上线一台Consumer做消息的搬运工，把oldTopic中的消息挪到newTopic里，不做业务逻辑处理，只是挪过去
• 上线N台Consumer同时消费newTopic中的数据
• 改bug
• 恢复原来的Consumer，继续消费之前的topic，oldTopic

2.6 延时消息

2.6.1 什么是延时消息

延时消息是指消息发送到 broker 后，不能立刻被 consumer 消费，要到特定的时间点或者等待特定的时间后才能被消费。 RocketMQ 支持延时消息，但是不支持任意时间精度，支持特定的 level，例如定时 5s，10s，1m 等。

2.6.2 延时消息使用方法

（1）broker.conf 配置文件配置

#broker.conf配置文件 
brokerClusterName = DefaultCluster 
brokerName = broker-a 
brokerId = 0 
deleteWhen = 04 
fileReservedTime = 48 
brokerRole = ASYNC_master 
flushDiskType = ASYNC_FLUSH 
#可以设置消息延时级别 
messageDelayLevel = 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

延迟配置说明:
配置项配置了从1级开始，各级延时的时间，可以修改这个指定级别的延时时间； 时间单位支持：s、m、h、d，分别表示秒、分、时、天；
默认值就是上面声明的，可手工调整； 默认值已够用，不建议修改这个值。
（2）设置消息延时级别

//创建消息对象 
message message = new message("topic-A", "tagB", ("hello" +i).getBytes(RemotingHelper.DEFAULT_CHARSET)); 
//设置消息延时级别 
message.setDelayTimeLevel(6);

2.6.3 延时消息使用场景

我们经常购物，知道从下订单到支付中间需要一段时间。这个过程中涉及到订单服务和支付服务，当支付完成时由支付服务修改数据库订单状态，代表订单完成。购物中经常有这样的处理情况：超过30分钟未支付的订单叫做超时订单，超时订单必须关闭。传统的做法是起一个定时任务服务，每隔一段时间扫描订单表，查询是否有超时订单，然后修改订单状态关闭订单。这种做法的弊端是扫描表数据量大，对数据库造成很大压力。我们可以使用延时消息来提高此需求的执行效率。首先在客户下订单后将订单发送的rocketMQ，消息包含订单编号并设置延时时间30分钟，然后添加一个订单超时服务，订单超时服务订阅延时消息然后处理订单表。

019.png

2.7 事务消息

RocketMQ除了支持普通消息，顺序消息，另外还支持事务消息。首先讨论一下什么是事务消息以及支持事务消息的必要性。

2.7.1 相关概念

RocketMQ在其消息定义的基础上，对事务消息扩展了两个相关的概念：
1.Half(Prepare) message——半消息(预处理消息)
半消息是一种特殊的消息类型，该状态的消息暂时不能被consumer消费。当一条事务消息被成功投递到broker上，但是broker并没有接收到producer发出的二次确认时，该事务消息就处于"暂时不可被消费"状态，该状态的事务消息被称为半消息。
2.message Status Check——消息状态回查
由于网络抖动、producer重启等原因，可能导致producer向broker发送的二次确认消息没有成功送达。如果broker检测到某条事务消息长时间处于半消息状态，则会主动向producer端发起回查操作，查询该事务消息在producer端的事务状态(Commit 或 Rollback)。可以看出，message Status Check主要用来解决分布式事务中的超时问题。

2.7.2 执行流程

020.png

上面是官网提供的事务消息执行流程图，下面对具体流程进行分析：
1.Step1：producer向broker端发送Half message；
2.Step2：broker ACK，Half message发送成功；
3.Step3：producer执行本地事务；
4.Step4：本地事务完毕，根据事务的状态，producer向broker发送二次确认消息，确认该Half message的Commit或者Rollback状态。broker收到二次确认消息后，对于Commit状态，则直接发送到consumer端执行消费逻辑，而对于Rollback则直接标记为失败，一段时间后清除，并不会发给consumer。正常情况下，到此分布式事务已经完成，剩下要处理的就是超时问题，即一段时间后broker仍没有收到producer的二次确认消息；
5.Step5：针对超时状态，broker主动向producer发起消息回查；
6.Step6：producer处理回查消息，返回对应的本地事务的执行结果；
7.Step7：broker针对回查消息的结果，执行Commit或Rollback操作，同Step4

2.7.3 实际案例

我们以一个转帐的场景为例来说明这个问题：Bob向Smith转账100块。
图中执行本地事务（Bob账户扣款）和发送异步消息应该保持同时成功或者失败中，也就是扣款成功了，发送消息一定要成功，如果扣款失败了，就不能再发送消息。那问题是我们是先扣款还是先发送消息呢？

首先我们看下，先发送消息，大致的示意图如下：

021.png

存在的问题是：如果消息发送成功，但是扣款失败，消费端就会消费此消息，进而向Smith账户加钱。

先发消息不行，那我们就先扣款呗，大致的示意图如下：

022.png

存在的问题跟上面类似：如果扣款成功，发送消息失败，就会出现Bob扣钱了，但是Smith账户未加钱。

可能大家会有很多的方法来解决这个问题，比如：直接将发消息放到Bob扣款的事务中去，如果发送失败，抛出异常，事务回滚。这样的处理方式也符合“恰好”不需要解决的原则。

RocketMQ支持事务消息，下面我们来看看RocketMQ是怎样来实现的。

023.png

RocketMQ第一阶段发送Prepared消息时会拿到消息的地址，第二阶段执行本地事物，第三阶段通过第一阶段拿到的地址去访问消息并修改状态。细心的你可能又发现问题了，如果确认消息发送失败了怎么办？RocketMQ会定期扫描消息集群中的事物消息，这时候发现了Prepared消息 ，它会向消息发送者确认，Bob的钱到底是减了还是没减呢，如果减了是回滚还是继续发送确认消息呢，RocketMQ 会根据发送端设置的策略来决定是回滚还是继续发送确认消息。这样就保证了消息发送与本地事务同时成功或同时失败。

3 高可靠性

3.1 RocketMQ可用性

多master部署，防止单点故障

024.png

3.2 RocketMQ可靠性

3.2.1 消息发送

producer 的 send 方法本身支持内部重试，重试逻辑如下∶
1.至多重试3次。
2.如果发送失败，则轮转到下一个 broker。
3.这个方法的总耗时时间不超过 sendMsgTimeout 设置的值，默认10s。
所以，如果本身向 broker发送消息产生超时异常，就不会再做重试。
以上策略仍然不能保证消息一定发送成功，为保证消息一定成功，建议应用这样做：如果调用send同步方法发送失败，则尝试将消息存储到db，由后台线程定时重试，保证消息一定到达 broker。

3.2.2 broker服务

所有发往broker的消息，有同步刷盘和异步刷盘机制，总的来说，可靠性非常高
同步刷盘时，消息写入物理文件才会返回成功，因此非常可靠
异步刷盘时，只有机器宕机，才会产生消息丢失，broker挂掉可能会发生，但是机器宕机崩溃是很少发生的，除非突然断电

3.2.3 消息消费

RocketMQ的消费与存储结构

025.png