RocketMQ发送消息

目录

一.消费模式​编辑

二.发送消息

1.普通消息

同步消息(***) 

异步消息(***)

单向消息(*)

日志服务的编写思路

2.延迟消息(***)

延迟等级 

3.批量消息

4.顺序消息(*)

三.Tag过滤

订阅关系的一致性

①订阅一个Topic且订阅一个Tag

②订阅一个Topic且订阅多个Tag

③订阅多个Topic且订阅多个Tag

什么时候该用 Topic，什么时候该用 Tag

四.消息中的key

五.消费重复问题 

问题产生

概念引入

幂等性

解决方案步骤

六.消息重试和死信消息

消息重试

死信消息

处理方案①：直接监听死信主题的消息

处理方案②（用的较多）

七.两种消费模式

八.消息积压问题

九.消息丢失问题

生产者角度 

MQ角度

消费者角度

一.消费模式

MQ的消费模式可以大致分为两种，一种是 推Push，一种是 拉Pull。 

• Push 是 服务端 (MQ) 主动推送消息给客户端，优点是及时性较好，但如果客户端没有做好流控，一旦服务端推送大量消息到客户端时，就会导致客户端消息堆积甚至崩溃。
• Pull 是 客户端 需要主动到 服务端 (MQ) 取数据，优点是客户端可以依据自己的消费能力进行消费，但拉取的频率也需要用户自己控制，拉取频繁容易造成服务端和客户端的压力，拉取间隔长又容易造成消费不及时。

Push模式也是基于Pull模式的，所以不管是Push模式还是Pull模式，都是Pull模式。一般情况下，优先选择Pull模式

二.发送消息

1.普通消息

同步消息(***) 

同步消息 发送过后会有一个返回值，也就是mq服务器接收到消息后返回的一个确认，这种方式非常安全，但是性能上并没有这么高，而且在mq集群中，也是要等到所有的从机都复制了消息以后才会返回，所以针对重要的消息可以选择这种方式。

可靠的同步传输被广泛应用于各种场景，如重要的通知消息、短消息通知等。

原生依赖引入：

        <!--  原生api,不是starter      -->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.9.0</version>
        </dependency>

同步消息生产者：

public class Producer {
    public static void main(String[] args) throws Exception {
            /*
             1. 谁来发？
             2. 发给谁？
             3. 怎么发？
             4. 发什么？
             5. 发的结果是什么？
             6. 关闭连接
             **/
            //1.创建一个发送消息的对象Producer，并指定生产者组名
            DefaultMQProducer producer = new DefaultMQProducer("sync-producer-group");
            //2.设定发送的命名服务器地址
            producer.setNamesrvAddr("ip:9876");
            producer.setSendMsgTimeout(1000000);

            //3.1启动发送的服务
            producer.start();
            //4.创建要发送的消息对象,指定topic，指定内容body
            Message msg = new Message("sync-topic", "hello-rocketmq".getBytes(StandardCharsets.UTF_8));
            //3.2发送消息
            SendResult result = producer.send(msg);
            System.out.println("返回结果：" + result);
            //5.关闭连接
            producer.shutdown();
    }
}

同步消息消费者：

public class Consumer {
    public static void main(String[] args) throws Exception {
        //1.创建一个接收消息的对象Consumer，并指定消费者组名
        //两种模式：①消费者定时拉取模式  ②建立长连接让Broker推送消息（选择第二种）
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("sync-producer-group");
        //2.设定接收的命名服务器地址
        consumer.setNamesrvAddr("ip:9876");
        //3.订阅一个主题，* 表示订阅这个主题的所有消息，后期会有消息过滤
        consumer.subscribe("sync-topic","*");
        //设置当前消费者的消费模式（默认模式：负载均衡）
        consumer.setMessageModel(MessageModel.CLUSTERING);
        //3.设置监听器，用于接收消息（一直监听，异步回调，异步线程）
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            //消费消息
            //消费上下文:consumeConcurrentlyContext
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                // 这个就是消费的方法 （业务处理）
                System.out.println("我是消费者");
                System.out.println(msgs.get(0).toString());
                System.out.println("消息内容:" + new String(msgs.get(0).getBody()));
                System.out.println("消费上下文:" + context);

                //签收消息，消息会从mq出队
                //如果返回 RECONSUME_LATER 或 null 或 产生异常 那么消息会重新 回到队列 过一会重新投递出来 ，给当前消费者或者其他消费者消费的
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //4.启动接收消息的服务
        consumer.start();
        System.out.println("接受消息服务已经开启！");
        //5 不要关闭消费者！因为需要监听！
        //挂起
        System.in.read();
    }
}

消息MessageExt组成：消息头（消息的基本属性）和消息体

异步消息(***)

异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应。发送完以后会有一个异步消息通知。

例如，视频上传后通知启动转码服务，转码完成后通知推送转码结果等。

 异步消息生产者：

public class AsyncProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("async-producer-group");
        producer.setNamesrvAddr("ip:9876");
        producer.start();
        Message message = new Message("async-topic", "我是一个异步消息".getBytes());
        //没有返回值的
        producer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("发送成功");
            }

            @Override
            public void onException(Throwable e) {
                System.err.println("发送失败:" + e.getMessage());
            }
        });
        System.out.println("我先执行");
        //需要接收异步回调，这里需要挂起
        System.in.read();
    }
}

消费者无特殊变化：

public class SimpleConsumer {
    public static void main(String[] args) throws Exception{
        // 创建一个消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("async-producer-group");
        // 连接namesrv
        consumer.setNamesrvAddr("ip:9876");
        // 订阅一个主题  * 标识订阅这个主题中所有的消息  后期会有消息过滤
        consumer.subscribe("async-topic", "*");
        // 设置一个监听器 (一直监听的， 异步回调方式)
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                // 这个就是消费的方法 （业务处理）
                System.out.println("我是消费者");
                System.out.println(msgs.get(0).toString());
                System.out.println("消息内容:" + new String(msgs.get(0).getBody()));
                System.out.println("消费上下文:" + context);
                // 返回值  CONSUME_SUCCESS成功，消息会从mq出队
                // RECONSUME_LATER（报错/null） 失败 消息会重新回到队列 过一会重新投递出来 给当前消费者或者其他消费者消费的
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动
        consumer.start();
        // 挂起当前的jvm
        System.in.read();
    }
}

单向消息(*)

这种方式主要用在不关心发送结果的场景，这种方式吞吐量很大，但是存在消息丢失的风险,一般用于结果不重要的场景，例如日志信息的发送

单向消息生产者：

public class SingleWayProducer {
    public static void main(String[] args) throws Exception{
        // 创建默认的生产者
        DefaultMQProducer producer = new DefaultMQProducer("single-way-producer-group");
        // 设置nameServer地址
        producer.setNamesrvAddr("ip:9876");
        // 启动实例
        producer.start();
        Message msg = new Message("single-way-topic", ("单向消息").getBytes());
        // 发送单向消息
        producer.sendOneway(msg);
        // 关闭实例
        producer.shutdown();
    }

}

日志服务的编写思路

产生日志的服务利用MQ发送单向消息，不用等回复，大大减少了发送日志的时间，由log-service统一写入日志表中。并且由于日志过于庞大，可以对日志进行冷热分离，近一个月的为热数据，近一年的为冷数据(实际情况据业务而定)，存储的位置不同，时间过于久远的日志可以删掉

2.延迟消息(***)

消息放入MQ后，过一段时间，才会被监听到，然后消费

比如下订单业务，提交了一个订单就可以发送一个延时消息，15min后去检查这个订单的状态，如果还是未付款就取消订单释放库存（订单超时）。

在分布式定时调度触发、任务超时处理等场景，使用 RocketMQ 的延时消息可以简化定时调度任务的开发逻辑，实现高性能、可扩展、高可靠的定时触发能力。

延迟等级 

延迟消息生产者：

public class DelayProducer {
    public static void main(String[] args) throws Exception{
        // 创建默认的生产者
        DefaultMQProducer producer = new DefaultMQProducer("delay-producer-group");
        // 设置nameServer地址
        producer.setNamesrvAddr("ip:9876");
        // 启动实例
        producer.start();
        Message msg = new Message("delay-topic", ("延迟消息").getBytes());
        // 给这个消息设定一个延迟等级
        // messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
        msg.setDelayTimeLevel(3);
        // 发送单向消息
        producer.send(msg);
        // 打印时间
        System.out.println(new Date());
        // 关闭实例
        producer.shutdown();
    }
}

 延迟消息消费者(无特殊变化)：

public class MSConsumer {
    public static void main(String[] args) throws Exception{
        // 创建一个消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("delay-producer-group");
        // 连接namesrv
        consumer.setNamesrvAddr("ip:9876");
        // 订阅一个主题  * 标识订阅这个主题中所有的消息  后期会有消息过滤
        consumer.subscribe("delay-topic", "*");
        // 设置一个监听器 (一直监听的， 异步回调方式)
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                System.out.println(msgs.get(0).toString());
                System.out.println("消息内容:" + new String(msgs.get(0).getBody()));
                System.out.println("收到时间："+new Date());
                // 返回值  CONSUME_SUCCESS成功，消息会从mq出队
                // RECONSUME_LATER（报错/null） 失败 消息会重新回到队列 过一会重新投递出来 给当前消费者或者其他消费者消费的
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动
        consumer.start();
        // 挂起当前的jvm
        System.in.read();
    }
}

可以通过打印一下时间差来检测一下(第一次有误差很正常)

3.批量消息

Rocketmq可以一次性发送一组消息，那么这一组消息会被当做一个消息消费。

在对吞吐率有一定要求的情况下，可以将一些消息聚成一批以后进行发送，可以增加吞吐率，并减少API和网络调用次数。

将消息打包成 Collection<Message> msgs 传入方法中即可，需要注意的是批量消息的大小不能超过 1MB（否则需要自行分割），其次同一批 batch 中 topic 必须相同。

批量消息生产者：

public class BatchProducer {
    public static void main(String[] args) throws Exception{
        // 创建默认的生产者
        DefaultMQProducer producer = new DefaultMQProducer("batch-producer-group");
        // 设置nameServer地址
        producer.setNamesrvAddr("ip:9876");
        // 启动实例
        producer.start();
        List<Message> msgs = Arrays.asList(
                //需要是同一种主题
                new Message("batch-topic", "我是一组消息的A消息".getBytes()),
                new Message("batch-topic", "我是一组消息的B消息".getBytes()),
                new Message("batch-topic", "我是一组消息的C消息".getBytes())

        );
        SendResult send = producer.send(msgs);
        System.out.println(send);
        // 关闭实例
        producer.shutdown();
    }
}

通过查看面板我们可以发现，批量消息全部放到了一个队列

 批量消息消费者(无特殊变化)： 

public class BatchConsumer {
    public static void main(String[] args) throws Exception{
        // 创建默认消费者组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("batch-producer-group");
        // 设置nameServer地址
        consumer.setNamesrvAddr("ip:9876");
        // 订阅一个主题来消费   表达式，默认是*
        consumer.subscribe("batch-topic", "*");
        // 注册一个消费监听 MessageListenerConcurrently是并发消费
        // 默认是20个线程一起消费，可以参看 consumer.setConsumeThreadMax()
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                // 这里执行消费的代码 默认是多线程消费
                System.out.println("msgs.size():"+msgs.size());
                System.out.println(Thread.currentThread().getName() + "----" + new String(msgs.get(0).getBody()));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.in.read();
    }
}

另外 consumeMessage 方法里的 List<MessageExt> msgs 参数 每次只有 1条，因为MessageListenerConcurrently 默认是并发消费模式 

打印结果如下：

4.顺序消息(*)

在默认的情况下普通消息的发送会采取 Round Robin 轮询方式 把 消息 发送到不同的queue(分区队列)；而消费消息的时候从多个queue上拉取消息，这未满足 生产顺序性 和 消费顺序性。比方说一个订单的生成、付款和发货，这三个操作需要被顺序执行，但如果是普通消息，订单A的消息可能会被轮询发送到不同的队列中，不同队列的消息将无法保持顺序。

消息有序指的是可以按照消息的发送顺序来消费(FIFO)，消息有序可以分为：分区有序 或者 全局有序。

• 全局有序：发送和消费参与的queue只有一个。即控制发送的顺序消息只依次发送到同一个queue中，消费的时候只从这个queue上依次拉取。要保证全局顺序，需要把 Topic 的读写队列数设置为 1，然后生产者和消费者的并发设置也是 1，不能使用多线程。所以这样的话 高并发，高吞吐量的功能完全用不上。
• 分区有序：发送和消费参与的queue有多个。但相对每个queue，消息都是有序的。比方说 订单A的生成、付款 和 发货 的消息由于订单号相同（设立标准）所以分在同一个queue（如queue1）中，订单B的三个操作的消息也会因为这个标准在同一个queue中，但不一定是queue1，因为我们希望它们分散在不同的queue中以提高吞吐量。

程序模拟，封装实体类MsgModel：

@Data
@AllArgsConstructor
@NoArgsConstructor
public class MsgModel {

    private String orderSn;
    private Integer userId;
    private String desc; // 下单 付款 发货

}

顺序消息生产者：

public class OrderProducer {
    private static final List<MsgModel> msgModels = Arrays.asList(
            new MsgModel("qwer", 1, "下单"),
            new MsgModel("qwer", 1, "付款"),
            new MsgModel("qwer", 1, "发货"),
            new MsgModel("zxcv", 2, "下单"),
            new MsgModel("zxcv", 2, "付款"),
            new MsgModel("zxcv", 2, "发货")
    );

    public static void main(String[] args) throws Exception{
        DefaultMQProducer producer = new DefaultMQProducer("orderly-producer-group");
        producer.setNamesrvAddr("ip:9876");
        producer.start();
        // 发送顺序消息  发送时要确保有序 并且要发到同一个队列下面去
        msgModels.forEach(msgModel -> {
            Message message = new Message("orderly-topic", msgModel.toString().getBytes());
            try {
                // 发 相同的订单号去相同的队列
                //MessageQueueSelector() 消息队列选择器
                producer.send(message, new MessageQueueSelector() {
                    @Override
                    //Object arg 即为 select的第三个参数 msgModel.getOrderSn()
                    //作为消息发送分区的分类标准
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        // 在这里 选择队列
                        // 保证 订单号 相同的消息在同一个 queue
                        int hashCode = arg.toString().hashCode();
                        // 周期性函数
                        int i = hashCode % mqs.size();
                        return mqs.get(i);
                    }
                }, msgModel.getOrderSn());

            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        producer.shutdown();
        System.out.println("发送完成");
    }
}

我们详细看看MessageQueueSelector的接口：

public interface MessageQueueSelector {
    MessageQueue select(final List<MessageQueue> mqs, final Message msg, final Object arg);
}

其中 mqs 是可以发送的队列，msg是消息，arg是上述send接口中传入的Object对象(第三个参数)，返回的是该消息需要发送到的队列。本例是以orderSn(订单编号)作为分区分类标准，对所有队列个数取余，来对将相同orderId的消息发送到同一个队列中。

生产环境中建议选择最细粒度的分区键进行拆分，例如，将订单ID、用户ID作为分区键关键字，可实现同一终端用户的消息按照顺序处理，不同用户的消息无需保证顺序。

顺序消息消费者：

public class OrderConsumer {
    public static void main(String[] args) throws Exception{
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("orderly-consumer-group");
        consumer.setNamesrvAddr("ip:9876");
        consumer.subscribe("orderly-topic", "*");
        // MessageListenerConcurrently 并发模式 多线程的  重试16次 后会将其放入 死信队列
        // MessageListenerOrderly 顺序模式 单线程的   无限重试Integer.Max_Value
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                System.out.println("线程id:" + Thread.currentThread().getId());
                System.out.println(new String(msgs.get(0).getBody()));
                //若返回 ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT 则会被挂起，等待一段时间再重试
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
        System.in.read();
    }
}

• MessageListenerConcurrently 并发模式 多线程的  重试16次 后会将其放入 死信队列
• MessageListenerOrderly 顺序模式 单线程的   无限重试（Integer.Max_Value 次）

若返回 ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT 则会被挂起，等待一段时间再重试

打印结果如下：

三.Tag过滤

Rocketmq提供消息过滤功能，通过 tag 或者 key 进行区分

tag 消息标签，用来进一步区分某个 Topic 下的消息分类，消息队列 RocketMQ 允许消费者按照 Tag 对消息进行过滤，确保消费者最终只消费到他关注的消息类型。

实操：producer会生产vip1、vip2两种 tag 的消息，而consumer1 只订阅 vip1的消息，consumer2 则订阅vip1 和 vip2 两种消息。consumer1 和 consumer2 订阅的虽是同一个topic，但 tag不同，为了保证订阅关系的一致性，它们应属于 不同的 消费者组。

TagProducer：

public class TagProducer {
    public static void main(String[] args) throws Exception{
        // 创建默认的生产者
        DefaultMQProducer producer = new DefaultMQProducer("tag-producer-group");
        // 设置nameServer地址
        producer.setNamesrvAddr("ip:9876");
        // 启动实例
        producer.start();
        Message msg1 = new Message("tag-topic","vip1", "vip1的消息".getBytes());
        Message msg2 = new Message("tag-topic","vip2", "vip2的消息".getBytes());
        producer.send(msg1);
        producer.send(msg2);
        System.out.println("发送成功");
        // 关闭实例
        producer.shutdown();
    }
}

TagConsumer1：

public class TagConsumer1 {
    public static void main(String[] args) throws Exception{
        // 订阅关系一致性
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tag-consumer-group-a");
        consumer.setNamesrvAddr("ip:9876");
        //只能收到 vip1 的消息
        consumer.subscribe("tag-topic", "vip1");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                System.out.println("我是vip1的消费者，我正在消费消息" + new String(msgs.get(0).getBody()));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.in.read();
    }
}

TagConsumer2：

public class TagConsumer2 {
    public static void main(String[] args) throws Exception{
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tag-consumer-group-b");
        consumer.setNamesrvAddr("ip:9876");
        consumer.subscribe("tag-topic", "vip1 || vip2");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                System.out.println("我是vip2的消费者，我正在消费消息" + new String(msgs.get(0).getBody()));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.in.read();
    }
}

订阅关系的一致性

• 订阅关系：一个消费者组订阅一个 Topic 的某一个 Tag，这种记录被称为订阅关系。
• 订阅关系一致：同一个消费者组下所有消费者实例所订阅的Topic、Tag必须完全一致。如果订阅关系（消费者组名-Topic-Tag）不一致，会导致消费消息紊乱，甚至消息丢失。

①订阅一个Topic且订阅一个Tag

如下图所示，同一Group ID下的三个Consumer实例C1、C2和C3分别都订阅了TopicA，且订阅TopicA的Tag也都是Tag1，符合订阅关系一致原则。

②订阅一个Topic且订阅多个Tag

如下图所示，同一Group ID下的三个Consumer实例C1、C2和C3分别都订阅了TopicB，订阅TopicB的Tag也都是Tag2和Tag3，表示订阅TopicB中所有Tag为Tag2或Tag3的消息，且顺序一致都是Tag2||Tag3，符合订阅关系一致性原则。

③订阅多个Topic且订阅多个Tag

如下图所示，同一Group ID下的三个Consumer实例C1、C2和C3分别都订阅了TopicA和TopicB，且订阅的TopicA都未指定Tag，即订阅TopicA中的所有消息，订阅的TopicB的Tag都是Tag2和Tag3，表示订阅TopicB中所有Tag为Tag2或Tag3的消息，且顺序一致都是Tag2||Tag3，符合订阅关系一致原则。

什么时候该用 Topic，什么时候该用 Tag

不同的业务应该使用不同的Topic，如果是相同的业务里面有不同表的表现形式，那么我们要使用tag进行区分

可以从以下几个方面进行判断：

1. 消息类型是否一致：如普通消息、事务消息、定时（延时）消息、顺序消息，不同的消息类型使用不同的 Topic，无法通过 Tag 进行区分。
2. 业务是否相关联：没有直接关联的消息，如淘宝交易消息，京东物流消息使用不同的 Topic 进行区分；而同样是天猫交易消息，电器类订单、女装类订单、化妆品类订单的消息可以用 Tag 进行区分。

3. 消息优先级是否一致：如同样是物流消息，盒马必须小时内送达，天猫超市 24 小时内送达，淘宝物流则相对会慢一些，不同优先级的消息用不同的 Topic 进行区分。

4. 消息量级是否相当：有些业务消息虽然量小但是实时性要求高，如果跟某些万亿量级的消息使用同一个 Topic，则有可能会因为过长的等待时间而“饿死”，此时需要将不同量级的消息进行拆分，使用不同的 Topic。

四.消息中的key

带key的消息生产者：

public class KeyProducer {
    public static void main(String[] args) throws Exception{
        DefaultMQProducer producer = new DefaultMQProducer("key-producer-group");
        producer.setNamesrvAddr("36.133.174.153:9876");
        producer.start();
        /*
         * 业务参数 我们自身要确保唯一
         * 为了查阅和去重
         * 我这里就用没有业务意义的UUID暂替一下
         */
        String key = UUID.randomUUID().toString();
        System.out.println("key:"+key);
        Message message = new Message("key-topic", "vip1", key, "我是vip1的文章".getBytes());
        producer.send(message);
        System.out.println("发送成功");
        producer.shutdown();
    }
}

带key的消息消费者：

public class KeyConsumer {
    public static void main(String[] args) throws Exception{
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("key-consumer-group");
        consumer.setNamesrvAddr("ip:9876");
        consumer.subscribe("key-topic", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                MessageExt messageExt = msgs.get(0);
                System.out.println("我是vip1的消费者，我正在消费消息" + new String(messageExt.getBody()));
                System.out.println("我们业务的标识：" + messageExt.getKeys());//取自己定义的业务标识key
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.in.read();
    }
}

五.消费重复问题 

问题产生

客户端重投机制导致的重复

比方说生产者发送消息的时候使用了重试机制，发送消息后由于网络原因没有收到MQ的响应信息，报了个超时异常，然后又去重新发送了一次消息。

但其实MQ已经接到了消息，并返回了响应，只是因为网络原因超时了。这种情况下，一条消息就会被发送两次。

消费者组重平衡时

当消费者组扩容时（反之一个消费者掉线后，也一样）会Reblance，重新分配 queue。一个队列所对应的新的消费者要获取之前消费的offset（偏移量，也就是消息消费的点位），此时之前的消费者可能已经消费了一条消息，但是并没有把offset提交给broker，那么新的消费者可能会重新消费一次。

概念引入

幂等性

多次操作所产生的影响均和第一次操作产生的影响相同

• 新增：可能是非幂等操作，可能是幂等操作。如果插入的字段中设置了唯一索引则为幂等操作（因为后面的插入不会生效）
• 修改：可能是非幂等操作，可能是幂等操作 

  update goods set stock = 10 where id = 1 //幂等性
  update goods set stock = stock -1 where id = 1 //非幂等性

• 查询：一般认为是幂等操作
• 删除：一般认为是幂等操作

解决方案步骤

第一步：发送方需要给消息带一个唯一的标记（自己业务控制）

第二步：消费者方在业务层面进行去重处理。

• 借助关系数据库(如MySQL)进行去重，设计一个去重表，对消息的唯一key添加唯一索引。每次消费消息的时候，先插入数据库。如果成功则执行业务逻辑(如果业务逻辑执行出现报错，则删除该去重记录)；如果插入失败，则说明消息来过了，直接签收。
• 使用布隆过滤器(BloomFilter)。布隆过滤器实际上是一个很长的二进制向量和一系列随机映射函数。布隆过滤器可以用于检索一个元素是否在一个集合中。它的优点是空间效率和查询时间都比一般的算法要好的多，缺点是有一定的误识别率和删除困难。
• 通过redis的setnx，根据设置是否成功判重

六.消息重试和死信消息

消息重试

分为生产者重试和消费者重试，主要还是消费者重试，所以这里重点讲讲消费者重试

如果消费者返回 RECONSUME_LATER 或 null 或 消费时产生异常，那么消息会重新 回到队列，过一会重新投递出来 ，给当前消费者或者其他消费者消费的

我们再实际生产过程中，一般重试3-5次，如果还没有消费成功，则可以把消息签收了，通知人工等处理。

重试时间间隔（根据那个延迟间隔来的）

10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

默认重试16次(并发模式) 顺序模式下(int最大值次)，可以自定义重试次数，如果重试后依然失败，则会放在一个死信主题中去 主题的名称：%DLQ%消费者组名

死信消息

当消费重试到达阈值以后，消息不会被投递给消费者了，而是进入了死信队列。这类消息称为死信消息，存储死信消息的特殊队列称为死信队列，死信队列是死信Topic下分区数唯一的单独队列(只有一个queue)。如果产生了死信消息，那对应的ConsumerGroup的死信Topic名称为%DLQ%ConsumerGroupName。注意：死信Topic能被订阅，死信队列也能被监听

处理方案①：直接监听死信主题的消息

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("retry-dead-consumer-group");
        consumer.setNamesrvAddr("ip:9876");
        consumer.subscribe("%DLQ%消费者组名", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                MessageExt messageExt = msgs.get(0);
                System.out.println(new Date());
                System.out.println(new String(messageExt.getBody()));
                System.out.println("记录到特别的位置 文件 mysql 通知人工处理");
                // 业务报错了 返回null 返回 RECONSUME_LATER 都会重试
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.in.read();

缺点：每一种死信队列都得写消费者组监听，很麻烦

处理方案②（用的较多）

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("retry-consumer-group");
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
        consumer.subscribe("ip:9876", "*");
        // 设定重试次数
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                MessageExt messageExt = msgs.get(0);
                System.out.println(new Date());
                // 业务处理
                try {
                   //注入的 service
                } catch (Exception e) {
                    // 重试
                    int reconsumeTimes = messageExt.getReconsumeTimes();
                    if (reconsumeTimes >= 3) {
                        // 不要重试了
                        System.out.println("记录到特别的位置 文件 mysql 通知人工处理");
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                // 业务报错了 返回null 返回 RECONSUME_LATER 都会重试
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.in.read();

七.两种消费模式

Rocketmq消息消费的模式分为两种：集群模式和广播模式

• 集群模式表示多个消费者交替消费同一个主题里面的消息

集群模式下，队列会被消费者分摊，队列数量最好>=消费者数量，并且MQ服务器会记录处理消息的消费位点

• 广播模式表示每个每个消费者都消费一遍订阅的主题的消息

广播模式下，消息会被每一个消费者都处理一次，MQ服务器不会记录消费点位，也不会重试

八.消息积压问题

单条队列queue最多存储30万w条消息，一般认为单条队列消息差值 >=10w 时 算堆积问题

出现问题原因：

①生产速度过快

解决方案：

1. 生产者方可以做业务限流
2. 适当增加消费者的数量，但是始终保持 消费者数量<=队列数量
3. 可以适当设置消费者的最大消费线程数量，如果消费者是I/O密集型(即操作数据库、文件或者调用 RPC等居多)，可设置 2n 个最大线程；如果消费者是CPU密集型(计算居多)，可设置 n+1 个最大线程。n 为服务器的逻辑处理器个数。
4. 动态扩容队列数量，从而增加消费者数量（与运维协商）
5. 如果业务对数据要求不高的话，可以选择丢弃不重要的消息。

②消费者出现问题/挂机

排查消费者程序的问题

九.消息丢失问题

我们可以分别从生产者角度、MQ角度、消费者角度来分析

文章来源地址https://www.toymoban.com/news/detail-501959.html

生产者角度 

• 生产者如果想尽量保证消息不丢失的情况下，可以选择同步发送
• ACK确认机制。RocketMQ提供了ACK机制，以保证消息能够被正常消费。发送者为了保证消息肯定消费成功，只有使用方明确表示消费成功，RocketMQ才会认为消息消费成功。中途断电，抛出异常等都不会认为成功——即引出了重试机制。
• 重试机制
• 生产者在发送前可以先记录发送日志，等消费者消费完后再去修改该日志的状态，采用定时任务查询如果存在 生产日志状态长时间未改变 的情况，可以进行消息补发，当然在消费层面需要做好幂等性处理。

MQ角度

• 提供同步刷盘的策略。当消息投递到broker之后，会先存到page cache，然后根据broker设置的刷盘策略是否立即刷盘，也就是如果刷盘策略为异步，broker并不会等待消息落盘就会返回producer成功，也就是说当broker所在的服务器突然宕机，则会丢失部分页的消息。
• 提供主从模式，同时主从支持同步双写。即使broker设置了同步刷盘，如果主broker磁盘损坏，也是会导致消息丢失。 因此可以给broker指定slave，同时设置master为SYNC_MASTER，然后将slave设置为同步刷盘策略。此模式下，producer每发送一条消息，都会等消息投递到master和slave都落盘成功了，broker才会当作消息投递成功，保证休息不丢失。但是这样性能较低，所以主从的同步策略也默认是异步同步。 

消费者角度

• 正确消费处理完才提交ACK
• 重试机制

到了这里，关于RocketMQ发送消息的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！