这三者是RocketMq中最最基本的概念。Producer是消息的生产者。Consumer是消息的消费者。消息通过Topic进行传递。Topic存放的是消息的逻辑地址。

　　具体来说是Producer将消息发往具体的Topic。Consumer订阅Topic，主动拉取或被动接受消息。

　　实际上，Topic还需要拆封出更多概念

　　这张图里有两个生产者，ProducerA和ProducerB。定义了两个Topic-TopicA和TopicB。ProducerA会发送两种消息。

　　所以这里的知识点是一个Producer可以发中Topic。

　　TopicA有3个MessageQueue，MessageQueue记录的是消息的物理存储地址（在consumelog里的位置），分布在两个broker上。Broker是一个集群部署架构上的概念，可以理解为对应的物理机器。最右边是ConsumerGroup，每一组下又有多个Consumer，实际上也就是启动的用来消费的JVM。一个Consumer可以订阅多个不同的Topic。这里我有话要说，虽然从代码层面上支持这种订阅。但是强烈不建议一个Consumer订阅多个不同的Topic。推荐用法是一组ConsumerGroup只订阅一种Topic。

　　另外多组ConsumerGroup之间，对于同一个Topic是广播订阅的。（翻译一下就是说：Topic的一条消息会广播给所有订阅的ConsumerGroup，就是每个ConsumerGroup都会收到），但是在一个ConsumerGroup内部给个Consumer是负载消费消息的，（翻译一下就是：一条消息在一个group内只会被一个Consumer消费）

存储模型

下面看看Rocketmq的消息实际是怎么存储的？

　　左边的是CommitLog。这个是真正存储消息的地方。可以看出RocketMQ所有生产者的消息都是往这一个地方存的。

　　右边是ConsumeQueue。这是一个逻辑队列。和上文中Topic下的messageQueue是一一对应的。消费者是直接和ConsumeQueue打交道。ConsumeQueue记录了消费位点，这个消费位点关联了commitlog的位置。所以即使ConsumeQueue出问题，只要commitlog还在，消息就没丢，可以恢复出来。还可以通过修改消费位点来重放或跳过一些消息。

部署模型 

　　在部署RocketMQ时，会部署两种角色。NameServer和Broker。NameServer主要做路由服务。生产者发送消息时，首先向NameServer拿到Topic的路由信息，即这个Topic在哪些Broker上有。Consumer也是一样，需要知道消费队列的路由情况。当然不是每次收发消息都去NameServer查询一遍，简单的说只有第一次初始化，和以后发送或这首出现问题时需要查询一下。

　　Broker一般我们会部署主备两个节点。

　　RocketMq没有选举，broker的角色是在部署时就人工确定好的。如果主挂了，备不会自动切换为主。

　　对于一个2主2备的集群来说，如果挂了一个主，是没有问题的。只要另一个主上你之前也创建了Topic，那么发送的消息流量会导流到存活的主节点上，业务代码端是无影响的。

示例代码

public class Producer {
    public static void main(String[] args) throws MQClientException, InterruptedException {

        //声明并初始化一个producer
        //需要一个producer group名字作为构造方法的参数，这里为concurrent_producer
        DefaultMQProducer producer = new DefaultMQProducer("concurrent_producer");

        //设置NameServer地址,此处应改为实际NameServer地址，多个地址之间用；分隔
        //NameServer的地址必须有，但是也可以通过环境变量的方式设置，不一定非得写死在代码里
        producer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");

        //调用start()方法启动一个producer实例
        producer.start();

        //发送10条消息到Topic为TopicTest，tag为TagA，消息内容为“Hello RocketMQ”拼接上i的值
        for (int i = 0; i < 10; i++) {
            try {
                Message msg = new Message("TopicTestConcurrent",// topic
                        "TagA",// tag
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)// body
                );

                //调用producer的send()方法发送消息
                //这里调用的是同步的方式，所以会有返回结果，同时默认发送的也是普通消息
                SendResult sendResult = producer.send(msg);

                //打印返回结果，可以看到消息发送的状态以及一些相关信息
                System.out.println(sendResult);
            } catch (Exception e) {
                e.printStackTrace();
                Thread.sleep(1000);
            }
        }

        //发送完消息之后，调用shutdown()方法关闭producer
        producer.shutdown();
    }
}

• 消费者

public class Consumer {

    public static void main(String[] args) throws InterruptedException, MQClientException {

        //声明并初始化一个consumer
        //需要一个consumer group名字作为构造方法的参数，这里为concurrent_consumer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("concurrent_consumer");

        //同样也要设置NameServer地址
        consumer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");

        //这里设置的是一个consumer的消费策略
        //CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，即跳过历史消息
        //CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
        //CONSUME_FROM_TIMESTAMP 从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        //设置consumer所订阅的Topic和Tag，*代表全部的Tag
        consumer.subscribe("TopicTestConcurrent", "*");

        //设置一个Listener，主要进行消息的逻辑处理
        //注意这里使用的是MessageListenerConcurrently这个接口
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {

                System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);

                //返回消费状态
                //CONSUME_SUCCESS 消费成功
                //RECONSUME_LATER 消费失败，需要稍后重新消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        //调用start()方法启动consumer
        consumer.start();

        System.out.println("Consumer Started.");
    }
}

有序消息

　　有序消息就是按照一定的先后顺序的消息类型。

　　举个例子来说，producer 依次发送 order id 为 1、2、3 的消息到 broker，consumer 接到的消息顺序也就是 1、2、3 ，而不会出现普通消息那样的 2、1、3 等情况。

　　那么有序消息是如何保证的呢？我们都知道消息首先由 producer 到 broker，再从 broker 到 consumer，分这两步走。那么要保证消息的有序，势必这两步都是要保证有序的，即要保证消息是按有序发送到 broker，broker 也是有序将消息投递给 consumer，两个条件必须同时满足，缺一不可。

进一步还可以将有序消息分成

• 全局有序消息
• 局部有序消息

之前我们讲过，topic 只是消息的逻辑分类，内部实现其实是由 queue 组成。当 producer 把消息发送到某个 topic 时，默认是会消息发送到具体的 queue 上。

　　举个例子，producer 发送 order id 为 1、2、3、4 的四条消息到 topicA 上，假设 topicA 的 queue 数为 3 个（queue0、queue1、queue2），那么消息的分布可能就是这种情况，id 为 1 的在 queue0，id 为 2 的在 queue1，id 为 3 的在 queue2，id 为 4 的在 queue0。同样的，consumer 消费时也是按 queue 去消费，这时候就可能出现先消费 1、4，再消费 2、3，和我们的预期不符。那么我们如何实现 1、2、3、4 的消费顺序呢？道理其实很简单，只需要把订单 topic 的 queue 数改为 1，如此一来，只要 producer 按照 1、2、3、4 的顺序去发送消息，那么 consumer 自然也就按照 1、2、3、4 的顺序去消费，这就是全局有序消息。

　　由于一个 topic 只有一个 queue ，即使我们有多个 producer 实例和 consumer 实例也很难提高消息吞吐量。就好比过独木桥，大家只能一个挨着一个过去，效率低下。

　　那么有没有吞吐量和有序之间折中的方案呢？其实是有的，就是局部有序消息。

　　我们知道订单消息可以再细分为订单创建、订单付款、订单完成等消息，这些消息都有相同的 order id。同时，也只有按照订单创建、订单付款、订单完成的顺序去消费才符合业务逻辑。但是不同 order id 的消息是可以并行的，不会影响到业务。这时候就常见做法就是将 order id 进行处理，将 order id 相同的消息发送到 topicB 的同一个 queue，假设我们 topicB 有 2 个 queue，那么我们可以简单的对 id 取余，奇数的发往 queue0，偶数的发往 queue1，消费者按照 queue 去消费时，就能保证 queue0 里面的消息有序消费，queue1 里面的消息有序消费。

　　由于一个 topic 可以有多个 queue，所以在性能比全局有序高得多。假设 queue 数是 n，理论上性能就是全局有序的 n 倍，当然 consumer 也要跟着增加才行。在实际情况中，这种局部有序消息是会比全局有序消息用的更多。

示例代码：

• 生产者

public class Producer {
    public static void main(String[] args) throws UnsupportedEncodingException {
        try {
            // 声明并初始化一个producer
            // 需要一个producer group名字作为构造方法的参数，这里为ordered_producer
            DefaultMQProducer orderedProducer = new DefaultMQProducer("ordered_producer");

            // 设置NameServer地址,此处应改为实际NameServer地址，多个地址之间用；分隔
            //NameServer的地址必须有，但是也可以通过环境变量的方式设置，不一定非得写死在代码里
            orderedProducer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");

            // 调用start()方法启动一个producer实例
            orderedProducer.start();

            // 自定义一个tag数组
            String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};

            // 发送10条消息到Topic为TopicTestOrdered，tag为tags数组按顺序取值，
            // key值为“KEY”拼接上i的值，消息内容为“Hello RocketMQ”拼接上i的值
            for (int i = 0; i < 10; i++) {

                int orderId = i % 10;
                Message msg =
                        new Message("TopicTestOrdered", tags[i % tags.length], "KEY" + i,
                                ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));

                SendResult sendResult = orderedProducer.send(msg, new MessageQueueSelector() {

                    // 选择发送消息的队列
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {

                        // arg的值其实就是orderId
                        Integer id = (Integer) arg;

                        // mqs是队列集合，也就是topic所对应的所有队列
                        int index = id % mqs.size();

                        // 这里根据前面的id对队列集合大小求余来返回所对应的队列
                        return mqs.get(index);
                    }
                }, orderId);

                System.out.println(sendResult);
            }

            orderedProducer.shutdown();
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

至于是要实现全局有序，还是局部有序，在此示例代码中，就取决于 TopicTestOrdered 这个 Topic 的队列数了。

• 消费者

public class Consumer {

    public static void main(String[] args) throws MQClientException {

        //声明并初始化一个consumer
        //需要一个consumer group名字作为构造方法的参数，这里为concurrent_consumer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ordered_consumer");

        //同样也要设置NameServer地址
        consumer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");

        //这里设置的是一个consumer的消费策略
        //CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，即跳过历史消息
        //CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
        //CONSUME_FROM_TIMESTAMP 从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        //设置consumer所订阅的Topic和Tag
        consumer.subscribe("TopicTestOrdered", "TagA || TagC || TagD");

        //设置一个Listener，主要进行消息的逻辑处理
        //注意这里使用的是MessageListenerOrderly这个接口
        consumer.registerMessageListener(new MessageListenerOrderly() {

            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {

                System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);

                //返回消费状态
                //SUCCESS 消费成功
                //SUSPEND_CURRENT_QUEUE_A_MOMENT 消费失败，暂停当前队列的消费
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        //调用start()方法启动consumer
        consumer.start();

        System.out.println("Consumer Started.");
    }
}

 

延时消息

　　延时消息，简单来说就是当 producer 将消息发送到 broker 后，会延时一定时间后才投递给 consumer 进行消费。

　　RcoketMQ的延时等级为：1s，5s，10s，30s，1m，2m，3m，4m，5m，6m，7m，8m，9m，10m，20m，30m，1h，2h。level=0，表示不延时。level=1，表示 1 级延时，对应延时 1s。level=2 表示 2 级延时，对应5s，以此类推。

　　这种消息一般适用于消息生产和消费之间有时间窗口要求的场景。比如说我们网购时，下单之后是有一个支付时间，超过这个时间未支付，系统就应该自动关闭该笔订单。那么在订单创建的时候就会就需要发送一条延时消息（延时15分钟）后投递给 consumer，consumer 接收消息后再对订单的支付状态进行判断是否关闭订单。

　　设置延时非常简单，只需要在Message设置对应的延时级别即可：

Message msg = new Message("TopicTest",// topic
                        "TagA",// tag
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)// body
                );
                // 这里设置需要延时的等级即可
                msg.setDelayTimeLevel(3);
                SendResult sendResult = producer.send(msg);

 

RocketMQ的消息发送方式

1. 同步发送
2. 异步发送
3. 单向发送

同步发送 

　　简单来说，同步发送就是指 producer 发送消息后，会在接收到 broker 响应后才继续发下一条消息的通信方式。

　　由于这种同步发送的方式确保了消息的可靠性，同时也能及时得到消息发送的结果，故而适合一些发送比较重要的消息场景，比如说重要的通知邮件、营销短信等等。在实际应用中，这种同步发送的方式还是用得比较多的。

 

异步发送

　　接着就是异步发送，异步发送是指 producer 发出一条消息后，不需要等待 broker 响应，就接着发送下一条消息的通信方式。需要注意的是，不等待 broker 响应，并不意味着 broker 不响应，而是通过回调接口来接收 broker 的响应。所以要记住一点，异步发送同样可以对消息的响应结果进行处理。

　　由于异步发送不需要等待 broker 的响应，故在一些比较注重 RT（响应时间）的场景就会比较适用。比如，在一些视频上传的场景，我们知道视频上传之后需要进行转码，如果使用同步发送的方式来通知启动转码服务，那么就需要等待转码完成才能发回转码结果的响应，由于转码时间往往较长，很容易造成响应超时。此时，如果使用的是异步发送通知转码服务，那么就可以等转码完成后，再通过回调接口来接收转码结果的响应了。

 

单向发送 

　　单向发送，见名知意，就是一种单方向通信方式，也就是说 producer 只负责发送消息，不等待 broker 发回响应结果，而且也没有回调函数触发，这也就意味着 producer 只发送请求不等待响应结果。

　　由于单向发送只是简单地发送消息，不需要等待响应，也没有回调接口触发，故发送消息所耗费的时间非常短，同时也意味着消息不可靠。所以这种单向发送比较适用于那些耗时要求非常短，但对可靠性要求并不高的场景，比如说日志收集。

 

下面通过一张表格，简单总结一下同步发送、异步发送和单向发送的特点。 

      

　　可以看到，从发送 TPS 来看，由于单向发送不需要等待响应也没有回调接口触发，发送速度非常快，一般都是微秒级的，在消息体大小一样的情况下，其发送 TPS 最大。而同步发送，需要等待响应结果的返回，受网络状况的影响较大，故发送 TPS 就比较小。异步发送不等待响应结果，发送消息时几乎不受网络的影响，故相比同步发送来说，其发送 TPS 要大得多。

　　关于可靠性，大家需要牢记前面提过的，异步发送并不意味着消息不可靠，异步发送也是会接收到响应结果，也能对响应结果进行处理。即使发送失败，也可以通过一些补偿手段进行消息重发。和同步发送比起来，异步发送的发送 TPS 更大，更适合那些调用链路较长的一些场景。在实际使用中，同步发送和异步发送都是较为常用的两种方式，大家要视具体业务场景进行合理地选择。

 

pull和push消费模式

1. pull是主动型消费，即能从服务器拉取到数据就开始消费
2. 首先通过打算消费的topic拿到MessageQueue中的集合消息，然后遍历拿取，并记录下次取消息时的offset位

 

1. push是被动型消费，多了一个注册消费监听器，本质还是从服务器拉取数据，但是要等到消费监听器被触发以后，才会进行消费
2. push方式中，注册MessageListener监听器，取到消息后，唤醒MessageListener中的consumerMessage（）来消费

 

RocketMQ的消费模式

1. 集群消费
2. 广播消费
3. ( 使用集群消费模拟广播消费 )

　　当 consumer 使用集群消费时，每条消息只会被 consumer 集群内的任意一个 consumer 实例消费一次。举个例子，当一个 consumer 集群内有 3 个consumer 实例（假设为consumer 1、consumer 2、consumer 3）时，一条消息投递过来，只会被consumer 1、consumer 2、consumer 3中的一个消费。

　　同时记住一点，使用集群消费的时候，consumer 的消费进度是存储在 broker 上，consumer 自身是不存储消费进度的。消息进度存储在 broker 上的好处在于，当你 consumer 集群是扩大或者缩小时，由于消费进度统一在broker上，消息重复的概率会被大大降低了。

　　注意：在集群消费模式下，并不能保证每一次消息失败重投都投递到同一个 consumer 实例。

　　当 consumer 使用广播消费时，每条消息都会被 consumer 集群内所有的 consumer 实例消费一次，也就是说每条消息至少被每一个 consumer 实例消费一次。举个例子，当一个 consumer 集群内有 3 个 consumer 实例（假设为 consumer 1、consumer 2、consumer 3）时，一条消息投递过来，会被 consumer 1、consumer 2、consumer 3都消费一次。

　　与集群消费不同的是，consumer 的消费进度是存储在各个 consumer 实例上，这就容易造成消息重复。还有很重要的一点，对于广播消费来说，是不会进行消费失败重投的，所以在 consumer 端消费逻辑处理时，需要额外关注消费失败的情况。

　　虽然广播消费能保证集群内每个 consumer 实例都能消费消息，但是消费进度的维护、不具备消息重投的机制大大影响了实际的使用。因此，在实际使用中，更推荐使用集群消费，因为集群消费不仅拥有消费进度存储的可靠性，还具有消息重投的机制。而且，我们通过集群消费也可以达到广播消费的效果。

　　如果业务上确实需要使用广播消费，那么我们可以通过创建多个 consumer 实例，每个 consumer 实例属于不同的 consumer group，但是它们都订阅同一个 topic。举个例子，我们创建 3 个 consumer 实例，consumer 1（属于consumer group 1）、consumer 2（属于 consumer group 2）、consumer 3（属于consumer group 3），它们都订阅了 topic A ，那么当 producer 发送一条消息到 topic A 上时，由于 3 个consumer 属于不同的 consumer group，所以 3 个consumer都能收到消息，也就达到了广播消费的效果了。 除此之外，每个 consumer 实例的消费逻辑可以一样也可以不一样，每个consumer group还可以根据需要增加 consumer 实例，比起广播消费来说更加灵活。 

　　说到消息过滤，就不得不说到 tag。没错，就是我们之前在专业术语中提到过的 tag。也称为消息标签，用来标记 Topic 下的不同用途的消息。

　　在 RocketMQ 中消费者是可以按照 Tag 对消息进行过滤。举个电商交易场景的例子，用户下完订单之后，在后台会产生一系列的消息，比如说订单消息、支付消息和物流消息。假设这些消息都发送到 Topic 为 Trade 中，同时用 tag 为 order 来标记订单消息，用 tag 为 pay 来标记支付消息，用 tag 为 logistics 来标记物流消息。需要支付消息的支付系统（相当于一个 consumer）订阅 Trade 中 tag 为 pay 的消息，此时，broker 则只会把 tag 为 pay 的消息投递给支付系统。而如果是一个实时计算系统，它可能需要接收所有和交易相关的消息，那么只要它订阅 Trade 中 tag 为 order、pay、logistics 的消息，broker 就会把带有这些 tag 的消息投递给实时计算系统。

　　对于消息分类，我们可以选择创建多个 Topic 来区分，也可以选择在同一个 Topic 下创建多个 tag 来区分。这两种方式都是可行的，但是一般情况下，不同的 Topic 之间的消息是没有什么必然联系的，使用 tag 来区分同一个 Topic 下相互关联的消息则更加合适一些。

　　讲完了消息过滤，我们接着讲讲什么是订阅关系一致性呢？其实在讲 RocketMQ 消费模式的时候提到过，除了使用同一个 group name，订阅的 tag 也必须是一样的，只有符合这两个条件的 consumer 实例才能组成 consumer 集群。这里所说的其实就是订阅关系一致性。在 RocketMQ 中，订阅关系由 Topic和 Tag 组成，因此要保证订阅关系一致性，就必须同时保证这两点：

1. 订阅的 Topic 必须一致

2. 订阅的 Topic 中的 tag 必须一致

　　保证订阅关系一致性是非常重要的，一旦订阅关系不一致，消息消费的逻辑就会混乱，甚至导致消息丢失，这对于大部分业务场景来说都是不允许的，甚至是致命的。在实际使用中，切记同一个消费者集群内的所有消费者实例务必要保证订阅关系的一致性。

　　　　　　　　　　　　　　　　　　　　　图 1

 备注：图中 “*” 代表订阅该Topic下所有的 tag。

　　我们用具体的例子来解释一下，如图 1 所示，消费者集群中有 3 个 consumer 实例，分别为 C1、C2、C3，各自订阅的 topic 和 tag 各不相同。首先 C1 和 C2 都订阅 TopicA，满足了订阅关系一致性的第一点，但是 C1 订阅的是 TopicA 的 Tag1，而 C2 订阅的是 TopicA 的 Tag2，不满足订阅关系一致性的第二点，所以 C1、C2 不满足订阅关系一致性。而 C3 订阅的 Topic 和 Tag 都与 C1 和 C2不一样，同样也不满足订阅关系一致性。

　　　　　　　　　　　　　　　　　　　　　　图 2

备注：图中 “||” 用来连接不用的 tag，表示与的意思。

　　在图 2 中，消费者集群中有 3 个 consumer 实例，分别为 C1、C2、C3，都是订阅 TopicA 下的 Tag1 和 Tag2，满足了订阅关系一致性的两点要求，所以满足了订阅关系一致性。

　　　　　　　　　　　　　　　　　　　　　　　　图 3

　　如图 3 所示，一个 consumer 也可以订阅多个 Topic，同时也必须保证该 consumer 集群里的多个消费者实例的订阅关系一致性，才不会造成不必要的麻烦。

　　在实际使用中，消息过滤可以帮助我们只消费我们所需要的消息，这是在broker端就帮我们处理好的，大大减少了在 consumer 端的消息过滤处理，一方面减少了代码量，另一方面更减少了不必要消息的网络传输消耗。

　　订阅消息一致性则保证了同一个消费者集群中 consumer 实例的正常运行，避免消息逻辑的混乱和消息的丢失。所以在实际使用中，在 producer 端要做好消息的分类，便于 consumer 可以使用 tag 进行消息的准确订阅，而在 consumer 端，则要保证订阅关系一致性。

　　首先明确之前说过的，消息重试只针对集群消费模式，广播消费没有消息重试的特性，消费失败之后，只会继续消费下一条消息。这也是为什么我们一再强调，推荐大家使用集群消费模式，其消息重试的特性能给开发者带来极大的方便。

　　那么什么是消息重试呢？简单来说，就是当消费者消费消息失败后，broker 会重新投递该消息，直到消费成功。在 RocketMQ 中，当消费者使用集群消费模式时，消费者接收到消息并进行相应的逻辑处理之后，最后都要返回一个状态值给 broker。这样 broker 才知道是否消费成功，需不需要重新投递消息。也就是说，我们可以通过设置返回的状态值来告诉 broker 是否重新投递消息。

　　到这里，可能大家会有一个疑问，那如果这条消息本身就是一条脏数据，就算你消费 100 次也不会消费成功，难道还是一直去重试嘛？其实 RocketMQ 并不会无限制地重试下去，默认每条消息最多重试 16 次，而每次重试的间隔时间如下表所示：