SpringBoot整合rockerMQ消息队列详解

Springboot整合RockerMQ

1、maven依赖

<dependencies>
   <!-- springboot-web组件 -->
   <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-web</artifactId>
   </dependency>
   <dependency>
       <groupId>org.apache.rocketmq</groupId>
       <artifactId>rocketmq-spring-boot-starter</artifactId>
       <version>2.0.3</version>
   </dependency>
   <dependency>
       <groupId>org.projectlombok</groupId>
       <artifactId>lombok</artifactId>
   </dependency>
</dependencies>

2、yml配置文件

rocketmq:
  ###连接地址nameServer
  name-server: www.kaicostudy.com:9876;
  producer:
    group: kaico_producer
server:
  port: 8088

3、生产者

@RequestMapping("/sendMsg")
   public String sendMsg() {
       OrderEntity orderEntity = new OrderEntity("123456","腾讯视频会员");
       SendResult kaicoTopic = rocketMQTemplate.syncSend("kaicoTopic"+":"+"tag1", orderEntity);
       System.out.println("返回发送消息状态：" + kaicoTopic);
       return "success";
   }

4、消费者

@Service
@RocketMQMessageListener(topic = "kaicoTopic", selectorExpression ="tag1", consumerGroup = "kaico_consumer", messageModel = MessageModel.CLUSTERING)
public class OrdeConsumer2 implements RocketMQListener<OrderEntity> {
   @Override
   public void onMessage(OrderEntity o) {
       System.out.println("kaico_consumer2消费者接收对象:" + o.toString());
   }
}

使用总结

消费模式

集群消费
当 consumer 使用集群消费时，每条消息只会被 consumer 集群内的任意一个 consumer 实例消费一次。
同时记住一点，使用集群消费的时候，consumer 的消费进度是存储在 broker 上，consumer 自身是不存储消费进度的。消息进度存储在 broker 上的好处在于，当你 consumer 集群是扩大或者缩小时，由于消费进度统一在broker上，消息重复的概率会被大大降低了。
注意： 在集群消费模式下，并不能保证每一次消息失败重投都投递到同一个 consumer 实例。

注解配置：messageModel = MessageModel.CLUSTERING

广播消费
当 consumer 使用广播消费时，每条消息都会被 consumer 集群内所有的 consumer 实例消费一次，也就是说每条消息至少被每一个 consumer 实例消费一次。
与集群消费不同的是，consumer 的消费进度是存储在各个 consumer 实例上，这就容易造成消息重复。还有很重要的一点，对于广播消费来说，是不会进行消费失败重投的，所以在 consumer 端消费逻辑处理时，需要额外关注消费失败的情况。
虽然广播消费能保证集群内每个 consumer 实例都能消费消息，但是消费进度的维护、不具备消息重投的机制大大影响了实际的使用。因此，在实际使用中，更推荐使用集群消费，因为集群消费不仅拥有消费进度存储的可靠性，还具有消息重投的机制。而且，我们通过集群消费也可以达到广播消费的效果。

注解配置：messageModel = MessageModel.BROADCASTING

生产者组和消费者组

生产者组
一个生产者组，代表着一群topic相同的Producer。即一个生产者组是同一类Producer的组合。

如果Producer是TransactionMQProducer，则发送的是事务消息。如果节点1发送完消息后，消息存储到broker的Half Message Queue中，还未存储到目标topic的queue中时，此时节点1崩溃，则可以通过同一Group下的节点2进行二阶段提交，或回溯。

使用时，一个节点下，一个topic会对应一个producer

消费者组
一个消费者组，代表着一群topic相同，tag相同（即逻辑相同）的Consumer。通过一个消费者组，则可容易的进行负载均衡以及容错

使用时，一个节点下，一个topic加一个tag可以对应一个consumer。一个消费者组就是横向上多个节点的相同consumer为一个消费组。

首先分析一下producer。习惯上我们不会创建多个订阅了相同topic的Producer实例，因为一个Producer实例发送消息时是通过ExecutorService线程池去异步执行的，不会阻塞完全够用，如果创建了多个相同topic的Producer则会影响性能。而Consumer则不同。消息会在一topic下会细分多个tag，需要针对tag需要针对不同的tag创建多个消费者实例。

注意：多个不同的消费者组订阅同一个topic、tag，如果设定的是集群消费模式，每一个消费者组中都会有一个消费者来消费。也就是说不同的消费者组订阅同一个topic相互之间是没有影响的。

生产者投递消息的三种方式

同步： 发送消息后需等待结果，消息的可靠性高发送速度慢；

SendResult kaicoTopic = rocketMQTemplate.syncSend("kaicoTopic"+":"+"tag1", orderEntity);

异步： 消息发送后，回调通知结果，消息发送速度快，消息可靠性低；

//异步发送
       rocketMQTemplate.asyncSend("kaicoTopic" + ":" + "tag1", orderEntity, new SendCallback() {
           @Override
           public void onSuccess(SendResult sendResult) {
               System.out.println("异步发送消息成功");
           }
           @Override
           public void onException(Throwable throwable) {
               System.out.println("异步发送消息失败");
           }
       });

单向(oneway)：消息发送后，不关心结果，发送速度最快，消息可靠性最差，适用于在大量日志数据和用户行为数据等场景发送数据。

//单向(oneway)发送
       rocketMQTemplate.sendOneWay("kaicoTopic"+":"+"tag1", orderEntity);

如何保证消息不丢失

主要三个步骤
1、生产者保证消息发送成功
采用同步发送消息的方式，发送消息后有返回结果，保证消息发送成功。（代码见上面）
返回四个状态

• SEND_OK：消息发送成功。需要注意的是，消息发送到 broker 后，还有两个操作：消息刷盘和消息同步到 slave 节点，默认这两个操作都是异步的，只有把这两个操作都改为同步，SEND_OK 这个状态才能真正表示发送成功。

• FLUSH_DISK_TIMEOUT：消息发送成功但是消息刷盘超时。

• FLUSH_SLAVE_TIMEOUT：消息发送成功但是消息同步到 slave 节点时超时。

• SLAVE_NOT_AVAILABLE：消息发送成功但是 broker 的 slave 节点不可用。

2、rocketMQ将消息持久化，保证宕机后消息不会丢失。持久化策略（刷盘策略）

• 异步刷盘：默认。消息写入 CommitLog 时，并不会直接写入磁盘，而是先写入 PageCache 缓存后返回成功，然后用后台线程异步把消息刷入磁盘。异步刷盘提高了消息吞吐量，但是可能会有消息丢失的情况，比如断点导致机器停机，PageCache 中没来得及刷盘的消息就会丢失。

• 同步刷盘：消息写入内存后，立刻请求刷盘线程进行刷盘，如果消息未在约定的时间内(默认 5 s)刷盘成功，就返回 FLUSH_DISK_TIMEOUT，Producer 收到这个响应后，可以进行重试。同步刷盘策略保证了消息的可靠性，同时降低了吞吐量，增加了延迟。要开启同步刷盘，需要增加下面配置：

flushDiskType=SYNC_FLUSH

3、Broker 多副本和高可用
Broker 为了保证高可用，采用一主多从的方式部署。
消息发送到 master 节点后，slave 节点会从 master 拉取消息保持跟 master 的一致。这个过程默认是异步的，即 master 收到消息后，不等 slave 节点复制消息就直接给 Producer 返回成功。

这样会有一个问题，如果 slave 节点还没有完成消息复制，这时 master 宕机了，进行主备切换后就会有消息丢失。为了避免这个问题，可以采用 slave 节点同步复制消息，即等 slave 节点复制消息成功后再给 Producer 返回发送成功。只需要增加下面的配置：
brokerRole=SYNC_MASTER

改为同步复制后，消息复制流程如下：

• slave 初始化后，跟 master 建立连接并向 master 发送自己的 offset；

• master 收到 slave 发送的 offset 后，将 offset 后面的消息批量发送给 slave；

• slave 把收到的消息写入 commitLog 文件，并给 master 发送新的 offset；

• master 收到新的 offset 后，如果 offset >= producer 发送消息后的 offset，给 Producer 返回 SEND_OK。

4、消费者保证消息消费成功
消费者消费消息后，如果 Consumer 消费成功，返回 CONSUME_SUCCESS，提交 offset 并从 Broker 拉取下一批消息。

@Service
public class NoSpringBootOrderConsumer {
   private DefaultMQPushConsumer defaultMQPushConsumer;
   @Value("${rocketmq.name-server}")
   private String namesrvAddr;
   protected String consumerGroup;
   protected String topic;
   protected String topicTag;
   public void setNamesrvAddr(String namesrvAddr) {
       this.namesrvAddr = namesrvAddr;
   }
   public void setConsumerGroup(String consumerGroup) {
       this.consumerGroup = consumerGroup;
   }
   public void setTopic(String topic) {
       this.topic = topic;
   }
   public void setTopicTag(String topicTag) {
       this.topicTag = topicTag;
   }
   public static String encoding = System.getProperty("file.encoding");
   /*
    * @Author ex_fengkai
    * @Description //TODO 初始化数据（消费者组名称、topic、topic的tag、nameServer的信息）
    * @Date 2020/11/9 14:36
    * @Param []
    * @return void
    **/
   private void initParam() {
       this.consumerGroup = "kaico_consumer3";
       this.topic = "kaicoTopic";
       this.topicTag = "tag1";
       this.setNamesrvAddr(namesrvAddr);
   }
   @PostConstruct
   private void init() throws InterruptedException, MQClientException {
       initParam();
       // ConsumerGroupName需要由应用来保证唯一，用于把多个Consumer组织到一起，提高并发处理能力
       defaultMQPushConsumer = new DefaultMQPushConsumer(consumerGroup);
       defaultMQPushConsumer.setNamesrvAddr(namesrvAddr); //设置nameServer服务器
       defaultMQPushConsumer.setInstanceName(String.valueOf(System.currentTimeMillis()));
       defaultMQPushConsumer.setVipChannelEnabled(false);
       // 设置Consumer第一次启动是从队列头部开始消费
       defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
       // 订阅指定Topic下的topicTag
       System.out.println("consumerGroup:" + consumerGroup + " topic:" + topic + " ,topicTag:" + topicTag);
       defaultMQPushConsumer.subscribe(topic, topicTag);
       // 设置为集群消费
       defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING);
       // 通过匿名消息监听处理消息消费
       defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
           // 默认msgs里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息
           @Override
           public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
               MessageExt msg = msgs.get(0);
               if (msg.getTopic().equals(topic) && msg.getTags() != null && msg.getTags().equals(topicTag)) {
                   // 执行topic下对应tag的消费逻辑
                   try {
                       onMessage(new String(msg.getBody(),"utf-8"));

} catch (UnsupportedEncodingException e) {
                       System.out.println("系统不支持消息编码格式:" + encoding);
                       return ConsumeConcurrentlyStatus.RECONSUME_LATER;

} catch (Exception e) {
                       System.out.println("消息处理异常");
                       return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                   }
                   System.out.println("consumerGroup:" + consumerGroup + " MsgId:" + msg.getMsgId() + " was done!");
               }
               // 如果没有return success ，consumer会重新消费该消息，直到return success
               return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
           }
       });
       // Consumer对象在使用之前必须要调用start初始化，初始化一次即可
       defaultMQPushConsumer.start();
       System.out.println("consumerGroup:" + consumerGroup + " namesrvAddr:" + namesrvAddr + "  start success!");
   }
   @PreDestroy
   public void destroy() {
       defaultMQPushConsumer.shutdown();
   }
   private void onMessage(String s) {
       System.out.println(consumerGroup + "用spring的方式的消费者消费：" + s);
   }
}

Consumer 重试
Consumer 消费失败，这里有 3 种情况：

• 返回 RECONSUME_LATER

• 返回 null

• 抛出异常

Broker 收到这个响应后，会把这条消息放入重试队列，重新发送给 Consumer。

注意：Broker 默认最多重试 16 次，如果重试 16 次都失败，就把这条消息放入死信队列，Consumer 可以订阅死信队列进行消费。重试只有在集群模式（MessageModel.CLUSTERING）下生效，在广播模式（MessageModel.BROADCASTING）下是不生效的。Consumer 端一定要做好幂等处理。

顺序消息

• 生产者投递消息根据key投递到同一个队列中存放

• 消费者应该订阅到同一个队列实现消费

• 最终应该使用同一个线程去消费消息(不能够实现多线程消费。)

生产者代码

//发送顺序消息
   @RequestMapping("/sendMsg1")
   public String sendMsg1() throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
       Long orderId = System.currentTimeMillis();
       String insertSql = getSqlMsg("insert", orderId);
       String updateSql = getSqlMsg("update", orderId);
       String deleteSql = getSqlMsg("delete", orderId);
       Message insertMsg = new Message("kaicoTopic", "tag6", insertSql.getBytes());
       Message updateMsg = new Message("kaicoTopic", "tag6", updateSql.getBytes());
       Message deleteMsg = new Message("kaicoTopic", "tag6", deleteSql.getBytes());
       DefaultMQProducer producer = rocketMQTemplate.getProducer();
       rocketMQTemplate.getProducer().send(insertMsg
               , new MessageQueueSelector() {
                   @Override
                   public MessageQueue select(List<MessageQueue> mqs, Message msg,
                                              Object arg) {
                       // 该消息存放到队列0中
                       return  mqs.get(0);
                   }
               }, orderId);
       rocketMQTemplate.getProducer().send(updateMsg
               , new MessageQueueSelector() {
                   @Override
                   public MessageQueue select(List<MessageQueue> mqs, Message msg,
                                              Object arg) {
                       // 该消息存放到队列0中
                       return mqs.get(0);
                   }
               }, orderId);
       rocketMQTemplate.getProducer().send(deleteMsg
               , new MessageQueueSelector() {
                   @Override
                   public MessageQueue select(List<MessageQueue> mqs, Message msg,
                                              Object arg) {
                       // 该消息存放到队列0中
                       return  mqs.get(0);
                   }
               }, orderId);
       return orderId + "";
   }

消费者代码

@Service
@RocketMQMessageListener(topic = "kaicoTopic", selectorExpression ="tag6", consumerGroup = "kaico_consumer1",
       messageModel = MessageModel.CLUSTERING, consumeMode = ConsumeMode.ORDERLY, consumeThreadMax = 1)
public class OrdeConsumer implements RocketMQListener<MessageExt> {
   @Override
   public void onMessage(MessageExt msg) {
       System.out.println(Thread.currentThread().getName() + "-kaico_consumer1消费者接收对象:队列" + msg.getQueueId()
               + "=消息：" +  new String(msg.getBody()));
   }
}

分布式事务

实现思路

• 生产者（发送方）投递事务消息到Broker中，设置该消息为半消息 不可以被消费；

• 开始执行我们的本地事务，将本地事务执行的结果（回滚或者提交）发送给Broker

• Broker获取回滚或者提交，如果是回滚的情况则删除该消息、如果是提交的话，该消息就可以被消费者消费；

• Broker如果没有及时的获取发送方本地事务结果的话，会主动查询本地事务结果。

1、生产者发送事务消息sendMessageInTransaction

public String saveOrder() {
    // 提前生成我们的订单id
    String orderId = System.currentTimeMillis() + "";
    /**
     * 1.提前生成我们的半消息
     * 2.半消息发送成功之后，在执行我们的本地事务
     */
    OrderEntity orderEntity = createOrder(orderId);
    String msg = JSONObject.toJSONString(orderEntity);
    MessageBuilder<String> stringMessageBuilder = MessageBuilder.withPayload(msg);
    stringMessageBuilder.setHeader("msg", msg);
    Message message = stringMessageBuilder.build();
    // 该消息不允许被消费者消费，生产者的事务逻辑代码在生产者事务监听类中executeLocalTransaction方法中执行。
    rocketMQTemplate.sendMessageInTransaction("kaicoProducer",
            "orderTopic", message, null);
    return orderId;

}

2、事务监听类

@Slf4j
@Component
@RocketMQTransactionListener(txProducerGroup = "kaicoProducer") //这个mayiktProducer生产者的事务管理
public class SyncProducerListener implements RocketMQLocalTransactionListener {
   @Autowired
   private OrderMapper orderMapper;

@Autowired
   private TransationalUtils transationalUtils;
   /**
    * 执行我们订单的事务
    * @param msg
    * @param arg
    * @return
    */
   @Override
   public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
       MessageHeaders headers = msg.getHeaders();
       //拿到消息
       Object object = headers.get("msg");
       if (object == null) {
           return null;
       }
       String orderMsg = (String) object;
       OrderEntity orderEntity = JSONObject.parseObject(orderMsg, OrderEntity.class);
       TransactionStatus begin = null;
       try {
           begin = transationalUtils.begin();
           int result = orderMapper.addOrder(orderEntity);
           transationalUtils.commit(begin);
           if (result <= 0) {
               return RocketMQLocalTransactionState.ROLLBACK;
           }
           // 告诉我们的Broke可以消费者该消息
           return RocketMQLocalTransactionState.COMMIT;
       } catch (Exception e) {
           if (begin != null) {
               transationalUtils.rollback(begin);
               return RocketMQLocalTransactionState.ROLLBACK;
           }
       }
       //add.Order
       return null;
   }
   /**
    * 提供给我们的Broker定时检查
    * @param msg
    * @return
    */
   @Override
   public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
       MessageHeaders headers = msg.getHeaders();
       Object object = headers.get("msg");
       if (object == null) {
           return RocketMQLocalTransactionState.ROLLBACK;
       }
       String orderMsg = (String) object;
       OrderEntity orderEntity = JSONObject.parseObject(orderMsg, OrderEntity.class);
       String orderId = orderEntity.getOrderId();
       // 直接查询我们的数据库
       OrderEntity orderDbEntity = orderMapper.findOrderId(orderId);
       if (orderDbEntity == null) {
           //不确认，继续重试
           return RocketMQLocalTransactionState.UNKNOWN;
       }
       //提交事务
       return RocketMQLocalTransactionState.COMMIT;
   }
}

3、消费者消费消息

@Service
@RocketMQMessageListener(topic = "orderTopic", consumerGroup = "kaicoTopic")
public class OrdeConsumer implements RocketMQListener<String> {
   @Autowired
   private DispatchMapper dispatchMapper;
   @Override
   public void onMessage(String msg) {
       OrderEntity orderEntity = JSONObject.parseObject(msg, OrderEntity.class);
       String orderId = orderEntity.getOrderId();
       // 模拟userid为=123456
       DispatchEntity dispatchEntity = new DispatchEntity(orderId, 123456L);
       dispatchMapper.insertDistribute(dispatchEntity);
   }
}

来源：https://blog.csdn.net/weixin_44044929/article/details/125847015