devbox
我家小花儿
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Springboot2.X 集成 RocketMQ_mqclientexception: code: 13 desc: the message body

作者：我家小花儿 | 2024-03-17 15:58:00

赞

踩

mqclientexception: code: 13 desc: the message body size over max value, max:

废话少说，上才艺

1.首先导入maven依赖（请自行选择对应的版本）

  1. <!--注意： 这里的版本,要和部署在服务器上的版本号一致--> 
  2. <dependency>
  3.     <groupId>org.apache.rocketmq</groupId>
  4.     <artifactId>rocketmq-client</artifactId>
  5.     <version>4.7.1</version>
  6. </dependency>

2.RocketMQ基本使用流程

消息发送

  • 创建生产者producer，并指定生产者组名
  • 指定NameServer（rockerMq 链接地址）
  • 启动producer
  • 创建消息对象，指定主题topic，tag和消息体
  • 发送消息
  • 关闭producer

消息消费

  • 创建消费者consumer，指定消费组名
  • 指定NameServer
  • 订阅主题topic和tag
  • 注册消息监听器，设置回调函数，处理消息
  • 启动消费者consumer

3.1 基础消息发送

  1. package com.zqh.com.rocketmq.baseMsg;
  2.  
  3. import org.apache.rocketmq.client.exception.MQBrokerException;
  4. import org.apache.rocketmq.client.exception.MQClientException;
  5. import org.apache.rocketmq.client.producer.DefaultMQProducer;
  6. import org.apache.rocketmq.client.producer.SendCallback;
  7. import org.apache.rocketmq.client.producer.SendResult;
  8. import org.apache.rocketmq.client.producer.SendStatus;
  9. import org.apache.rocketmq.common.message.Message;
  10. import org.apache.rocketmq.remoting.exception.RemotingException;
  11.  
  12. import java.util.concurrent.CountDownLatch;
  13. import java.util.concurrent.TimeUnit;
  14.  
  15. /**
  16.  * SEND_OK：消息发送成功
  17.  * FLUSH_DISK_TIMEOUT：消息发送成功，但是服务器刷盘超时，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失
  18.  * FLUSH_SLAVE_TIMEOUT：消息发送成功，但是服务器同步到 Slave 时超时，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失
  19.  * SLAVE_NOT_AVAILABLE：消息发送成功，但是此时 Slave 不可用，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失
  20.  */
  21. public class BaseMsgProducer {
  22.  
  23.     final static String NAMESRV_ADDR = "192.168.0.30:9876";
  24.     final static String PRODUCER_GROUP = "base_group";
  25.     final static String TOPIC = "base_topic";
  26.     final static String TAGS = "base_tags";
  27.  
  28.     public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
  29.         sendSyncMsg();
  30. //        sendASyncMsg();
  31. //        sendOnewayMsg();
  32.     }
  33.  
  34.     /**
  35.      * 发送同步消息需要接到消息结果之后再发送下一个消息，这种可靠性同步地发送方式使用的比较广泛，比如：重要的消息通知，短信通知。
  36.      */
  37.     public static void sendSyncMsg() throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
  38.         // 创建生产者producer，并指定生产者组名
  39.         DefaultMQProducer producer = new DefaultMQProducer(PRODUCER_GROUP + "_syncMsg");
  40.         //指定NameServer（生产者需要把提供的消息注册到nameServer，方便消费端查找和订阅）
  41.         producer.setNamesrvAddr(NAMESRV_ADDR);
  42.         producer.start();
  43.         for (int i = 0; i < 1000; i++) {
  44.             //创建消息对象，指定主题topic，tag和消息体
  45.             Message message = new Message(TOPIC, TAGS, System.currentTimeMillis() + "" + i, ("hello rocketMq" + i).getBytes());
  46.             //发送消息
  47.             SendResult result = producer.send(message);
  48.             //发送状态
  49.             SendStatus sendStatus = result.getSendStatus();
  50.             //消息ID
  51.             String msgId = result.getMsgId();
  52.             //消息队列ID
  53.             int queueId = result.getMessageQueue().getQueueId();
  54.             System.out.println("发送结果：" + result);
  55.         }
  56.         //6.- 关闭producer
  57.         producer.shutdown();
  58.     }
  59.  
  60.  
  61.     /**
  62.      * 异步消息不必等待返回结果，立即发送下一个消息，可以通过send(Message msg, SendCallback sendCallback)中的回调函数，对返回结果进行处理。异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应。
  63.      */
  64.     public static void sendASyncMsg() throws InterruptedException, RemotingException, MQClientException {
  65.         DefaultMQProducer producer = new DefaultMQProducer(PRODUCER_GROUP + "_async");
  66.         producer.setNamesrvAddr(NAMESRV_ADDR);
  67.         producer.start();
  68.         final CountDownLatch countDownLatch = new CountDownLatch(10);
  69.         for (int i = 0; i < 10; i++) {
  70.             Message message = new Message(TOPIC, TAGS + "_async", System.currentTimeMillis() + "" + i, ("hello rocketMq" + i).getBytes());
  71.             producer.send(message, new SendCallback() {
  72.                 @Override
  73.                 public void onSuccess(SendResult sendResult) {
  74.                     countDownLatch.countDown();
  75.                     System.out.println("发送结果：" + sendResult);
  76.                 }
  77.  
  78.                 @Override
  79.                 public void onException(Throwable throwable) {
  80.                     countDownLatch.countDown();
  81.                     System.out.println("异常结果：" + throwable.getMessage());
  82.                 }
  83.             });
  84.         }
  85.  
  86.         countDownLatch.await(5, TimeUnit.SECONDS);
  87.         // 避免提前关闭了，所以等发送完再关闭
  88.         producer.shutdown();
  89.     }
  90.  
  91.     /**
  92.      * 单向消息通俗来说，就是发送消息不必等待返回结果，也无需执行回调函数。这种方式主要用在不特别关心发送结果的场景，例如日志发送。
  93.      */
  94.     public static void sendOnewayMsg() throws InterruptedException, RemotingException, MQClientException {
  95.         DefaultMQProducer producer = new DefaultMQProducer(PRODUCER_GROUP + "_oneway");
  96.         producer.setNamesrvAddr(NAMESRV_ADDR);
  97.         producer.start();
  98.         for (int i = 0; i < 10; i++) {
  99.             Message message = new Message(TOPIC, TAGS + "_oneway", System.currentTimeMillis() + "" + i, ("hello rocketMq" + i).getBytes());
  100.             producer.sendOneway(message);
  101.             System.out.println("发送完成");
  102.         }
  103.         producer.shutdown();
  104.     }
  105. }

3.2 消费者

  1. package com.zqh.com.rocketmq.baseMsg;
  2.  
  3. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  4. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
  5. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
  6. import org.apache.rocketmq.client.exception.MQClientException;
  7. import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
  8. import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
  9. import org.springframework.stereotype.Component;
  10.  
  11. @Component
  12. public class BaseMsgConsumer {
  13.     private DefaultMQPushConsumer consumer;
  14.     private static final String CONSUMER_GROUP = "base_group";
  15.     private final static String NAMESRV_ADDR = "192.168.0.30:9876";
  16.     private final static String TOPIC = "base_topic";
  17.     private final static String TAGS = "base_tags_async || base_tags_oneway || base_tags";
  18.  
  19.     public BaseMsgConsumer() throws MQClientException {
  20.         System.out.println("初始化成功");
  21.         consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
  22.         consumer.setNamesrvAddr(NAMESRV_ADDR);
  23.         consumer.subscribe(TOPIC, TAGS);
  24.         // 分散消费 MessageModel.CLUSTERING 同一个 Consumer ID 所标识的所有 Consumer 分散消费消息。
  25.         // 广播消费 MessageModel.BROADCASTING 同一个 Consumer ID 所标识的所有 Consumer 都会各自消费某条消息一次。
  26.         consumer.setMessageModel(MessageModel.BROADCASTING);
  27.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
  28.  
  29.         // 并发消费模式 （MessageListenerConcurrently）
  30.         consumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
  31.             System.out.println(Thread.currentThread().getName()
  32.                     + " Receive New Messages: " + list.size());
  33.             list.forEach(msg -> {
  34.                 // 业务实现
  35.                 System.out.println(Thread.currentThread().getName() + "\t" + "组1消费" + msg.getKeys() + new String(msg.getBody()));
  36.             });
  37.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  38.         });
  39.         consumer.start();
  40.     }
  41. }

4.1 批量信息

  1. package com.zqh.com.rocketmq.batchMsg;
  2.  
  3. import org.apache.rocketmq.client.exception.MQBrokerException;
  4. import org.apache.rocketmq.client.exception.MQClientException;
  5. import org.apache.rocketmq.client.producer.DefaultMQProducer;
  6. import org.apache.rocketmq.client.producer.SendResult;
  7. import org.apache.rocketmq.common.message.Message;
  8. import org.apache.rocketmq.remoting.exception.RemotingException;
  9. import org.slf4j.Logger;
  10. import org.slf4j.LoggerFactory;
  11.  
  12. import java.util.ArrayList;
  13. import java.util.List;
  14.  
  15. /**
  16.  * 批量信息
  17.  * 前面发送的消息都是一条一条发送，批量消息则是一次发送多条消息，这一批消息总大小（无论是单条消息还是消息总大小）不应该超过4MB（默认大小）
  18.  * 批量消息一般有两种使用情况：
  19.  * 达到一定消息数量之后发送
  20.  * 一定时间发送（比如1分钟发送一次）
  21.  */
  22. public class BatchMsgProducer {
  23.  
  24.     private final static Logger logger = LoggerFactory.getLogger(BatchMsgProducer.class);
  25.     private static final String PRODUCER_GROUP = "batch_group";
  26.     private final static String NAMESRV_ADDR = "192.168.0.30:9876";
  27.     private final static String TOPIC = "batch_topic";
  28.     private final static String TAGS = "batch_tags";
  29.  
  30.     public static void sendBatchMsg() throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
  31.         DefaultMQProducer producer = new DefaultMQProducer(PRODUCER_GROUP);
  32.         producer.setNamesrvAddr(NAMESRV_ADDR);
  33.         producer.start();
  34.         for (int i = 0; i < 10; i++) {
  35.             // 创建消息集合
  36.             List<Message> messageList = new ArrayList<>();
  37.             // CODE: 13  DESC: the message body size over max value, MAX: 4194304
  38.             byte[] bytes = new byte[1024 * 1024 * 3];
  39.             messageList.add(new Message(TOPIC, TAGS, "bytes".getBytes()));
  40.             messageList.add(new Message(TOPIC, TAGS, "bytes".getBytes()));
  41.             messageList.add(new Message(TOPIC, TAGS, "bytes".getBytes()));
  42.             SendResult sendResult = producer.send(messageList);
  43.             System.out.println("发送结果：" + sendResult);
  44.         }
  45.         producer.shutdown();
  46.     }
  47.  
  48.     public static void main(String[] args) throws Exception {
  49.         sendBatchMsg();
  50.     }
  51. }

4.2 消费者

  1. package com.zqh.com.rocketmq.batchMsg;
  2.  
  3. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  4. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
  5. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
  6. import org.apache.rocketmq.client.exception.MQClientException;
  7. import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
  8. import org.slf4j.Logger;
  9. import org.slf4j.LoggerFactory;
  10. import org.springframework.stereotype.Component;
  11.  
  12. @Component
  13. public class BatchMsgConsumer {
  14.     private final static Logger logger = LoggerFactory.getLogger(BatchMsgConsumer.class);
  15.     private DefaultMQPushConsumer consumer;
  16.     private static final String CONSUMER_GROUP = "batch_group";
  17.     private final static String NAMESRV_ADDR = "192.168.0.30:9876";
  18.     private final static String TOPIC = "batch_topic";
  19.     private final static String TAGS = "batch_tags";
  20.  
  21.     public BatchMsgConsumer() throws MQClientException {
  22.         System.out.println("初始化成功");
  23.         consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
  24.         consumer.setNamesrvAddr(NAMESRV_ADDR);
  25.         consumer.subscribe(TOPIC, TAGS);
  26.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
  27.  
  28.         // 并发消费模式 （MessageListenerConcurrently）
  29.         consumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
  30.             System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + list.size());
  31.             list.forEach(msg -> {
  32.                 // 业务实现
  33.                 System.out.println(Thread.currentThread().getName() + "\t" + "组1消费" + msg.getKeys() + new String(msg.getBody()));
  34.             });
  35.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  36.         });
  37.         consumer.start();
  38.     }
  39. }

5.1 延时消费

  1. package com.zqh.com.rocketmq.delayMsg;
  2.  
  3. import org.apache.rocketmq.client.exception.MQBrokerException;
  4. import org.apache.rocketmq.client.exception.MQClientException;
  5. import org.apache.rocketmq.client.producer.DefaultMQProducer;
  6. import org.apache.rocketmq.client.producer.SendResult;
  7. import org.apache.rocketmq.common.message.Message;
  8. import org.apache.rocketmq.remoting.exception.RemotingException;
  9. import org.slf4j.Logger;
  10. import org.slf4j.LoggerFactory;
  11.  
  12. /**
  13.  * 延时信息
  14.  * 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
  15.  */
  16. public class DelayMsgProducer {
  17.  
  18.     private final static Logger logger = LoggerFactory.getLogger(DelayMsgProducer.class);
  19.     private static final String PRODUCER_GROUP = "delay_group";
  20.     private final static String NAMESRV_ADDR = "192.168.0.30:9876";
  21.     private final static String TOPIC = "delay_topic";
  22.     private final static String TAGS = "delay_tags";
  23.  
  24.  
  25.     public static void sendDelayMsg() throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
  26.         DefaultMQProducer producer = new DefaultMQProducer(PRODUCER_GROUP);
  27.         producer.setNamesrvAddr(NAMESRV_ADDR);
  28.         producer.start();
  29.         for (int i = 0; i < 10; i++) {
  30.             Message message = new Message(TOPIC, TAGS, System.currentTimeMillis() + "" + i, "rocketMq".getBytes());
  31.             // 3对应的10s
  32.             message.setDelayTimeLevel(3);
  33.             // 发送消息
  34.             SendResult sendResult = producer.send(message);
  35.             System.out.println("发送结果：" + sendResult);
  36.         }
  37.         producer.shutdown();
  38.     }
  39.  
  40.     public static void main(String[] args) throws Exception {
  41.         sendDelayMsg();
  42.     }
  43. }

5.2 消费端

  1. package com.zqh.com.rocketmq.delayMsg;
  2.  
  3. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  4. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
  5. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
  6. import org.apache.rocketmq.client.exception.MQClientException;
  7. import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
  8. import org.slf4j.Logger;
  9. import org.slf4j.LoggerFactory;
  10. import org.springframework.stereotype.Component;
  11.  
  12. @Component
  13. public class DelayMsgConsumer {
  14.     private final static Logger logger = LoggerFactory.getLogger(DelayMsgConsumer.class);
  15.     private DefaultMQPushConsumer consumer;
  16.     private static final String CONSUMER_GROUP = "delay_group";
  17.     private final static String NAMESRV_ADDR = "192.168.0.30:9876";
  18.     private final static String TOPIC = "delay_topic";
  19.     private final static String TAGS = "delay_tags";
  20.  
  21.     public DelayMsgConsumer() throws MQClientException {
  22.         System.out.println("初始化成功");
  23.         consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
  24.         consumer.setNamesrvAddr(NAMESRV_ADDR);
  25.         consumer.subscribe(TOPIC, TAGS);
  26.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
  27.  
  28.         // 并发消费模式 （MessageListenerConcurrently）
  29.         consumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
  30.             System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + list.size());
  31.             list.forEach(msg -> {
  32.                 // 业务实现
  33.                 System.out.println(Thread.currentThread().getName() + "\t" + "组1消费" + msg.getKeys() + new String(msg.getBody()));
  34.             });
  35.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  36.         });
  37.         consumer.start();
  38.     }
  39. }

6.1 过滤消息

  1. package com.zqh.com.rocketmq.filterMsg;
  2.  
  3. import org.apache.rocketmq.client.exception.MQBrokerException;
  4. import org.apache.rocketmq.client.exception.MQClientException;
  5. import org.apache.rocketmq.client.producer.DefaultMQProducer;
  6. import org.apache.rocketmq.client.producer.SendResult;
  7. import org.apache.rocketmq.common.message.Message;
  8. import org.apache.rocketmq.remoting.exception.RemotingException;
  9. import org.slf4j.Logger;
  10. import org.slf4j.LoggerFactory;
  11.  
  12. /**
  13.  * 过滤信息
  14.  */
  15. public class FilterMsgProducer {
  16.  
  17.     private final static Logger logger = LoggerFactory.getLogger(FilterMsgProducer.class);
  18.     private static final String PRODUCER_GROUP = "filter_group";
  19.     private final static String NAMESRV_ADDR = "192.168.0.30:9876";
  20.     private final static String TOPIC = "filter_topic";
  21.     private final static String TAGS = "filter_tags";
  22.  
  23.     public static void sendFilterMsg() throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
  24.         DefaultMQProducer producer = new DefaultMQProducer(PRODUCER_GROUP);
  25.         producer.setNamesrvAddr(NAMESRV_ADDR);
  26.         producer.start();
  27.         for (int i = 0; i < 10; i++) {
  28.             Message message = new Message(TOPIC, TAGS, System.currentTimeMillis() + "" + i, "bytes".getBytes());
  29.             // 绑定自定义属性
  30.             message.putUserProperty("i", i + "");
  31.  
  32.             SendResult sendResult = producer.send(message);
  33.             System.out.println("发送结果：" + sendResult);
  34.         }
  35.         producer.shutdown();
  36.     }
  37.  
  38.     public static void main(String[] args) throws Exception {
  39.         sendFilterMsg();
  40.     }
  41. }

6.2 消费端

  1. package com.zqh.com.rocketmq.filterMsg;
  2.  
  3. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  4. import org.apache.rocketmq.client.consumer.MessageSelector;
  5. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
  6. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
  7. import org.apache.rocketmq.client.exception.MQClientException;
  8. import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
  9. import org.slf4j.Logger;
  10. import org.slf4j.LoggerFactory;
  11. import org.springframework.stereotype.Component;
  12.  
  13. @Component
  14. public class FileterMsgConsumer {
  15.     private final static Logger logger = LoggerFactory.getLogger(FileterMsgConsumer.class);
  16.     private DefaultMQPushConsumer consumer;
  17.     private static final String CONSUMER_GROUP = "filter_group";
  18.     private final static String NAMESRV_ADDR = "192.168.0.30:9876";
  19.     private final static String TOPIC = "filter_topic";
  20.     // * 的话代表全部接收
  21.     private final static String TAGS = "filter_tags";
  22.  
  23.     public FileterMsgConsumer() throws MQClientException {
  24.         System.out.println("初始化成功");
  25.         consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
  26.         consumer.setNamesrvAddr(NAMESRV_ADDR);
  27. //        通过tag来过滤需要的
  28. //        consumer.subscribe(TOPIC, TAGS);
  29.         //通过sql过滤消息，只要i>=3的消息
  30.         consumer.subscribe(TOPIC, MessageSelector.bySql("i>=3"));
  31.  
  32.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
  33.  
  34.         // 并发消费模式 （MessageListenerConcurrently）
  35.         consumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
  36.             System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + list.size());
  37.             list.forEach(msg -> {
  38.                 // 业务实现
  39.                 System.out.println(Thread.currentThread().getName() + "\t" + "组1消费" + msg.getKeys() + new String(msg.getBody()));
  40.             });
  41.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  42.         });
  43.         consumer.start();
  44.     }
  45. }

7.1 顺序消息

  1. package com.zqh.com.rocketmq.orderMsg;
  2.  
  3. import org.apache.rocketmq.client.exception.MQBrokerException;
  4. import org.apache.rocketmq.client.exception.MQClientException;
  5. import org.apache.rocketmq.client.producer.DefaultMQProducer;
  6. import org.apache.rocketmq.client.producer.MessageQueueSelector;
  7. import org.apache.rocketmq.client.producer.SendResult;
  8. import org.apache.rocketmq.common.message.Message;
  9. import org.apache.rocketmq.common.message.MessageQueue;
  10. import org.apache.rocketmq.remoting.exception.RemotingException;
  11.  
  12. import java.util.ArrayList;
  13. import java.util.List;
  14.  
  15. /**
  16.  * SEND_OK：消息发送成功
  17.  * FLUSH_DISK_TIMEOUT：消息发送成功，但是服务器刷盘超时，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失
  18.  * FLUSH_SLAVE_TIMEOUT：消息发送成功，但是服务器同步到 Slave 时超时，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失
  19.  * SLAVE_NOT_AVAILABLE：消息发送成功，但是此时 Slave 不可用，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失
  20.  * <p>
  21.  * 保证消息的顺序性 全局顺序消息
  22.  * 一般通过建立单队列的方式，消息全在一个队列里，保证消息的顺序性（FIFO）
  23.  * 局部顺序消息
  24.  * 不需要限定单个队列，将需要按序消费的消息，放入到同一个队列中即可。
  25.  * 比如：顺序消息A1，B2，C3，D4，都放入到队列1中，就能保证消息的顺序性。
  26.  */
  27. public class OrderMsgProducer {
  28.  
  29.     final static String NAMESRV_ADDR = "192.168.0.30:9876";
  30.     final static String PRODUCER_GROUP = "order_group";
  31.     final static String TOPIC = "order_topic";
  32.     final static String TAGS = "order_tags";
  33.  
  34.     public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
  35.         sendOrderMsg();
  36.     }
  37.  
  38.     /**
  39.      * 发送同步消息需要接到消息结果之后再发送下一个消息，这种可靠性同步地发送方式使用的比较广泛，比如：重要的消息通知，短信通知。
  40.      */
  41.     public static void sendOrderMsg() throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
  42.         // 创建生产者producer，并指定生产者组名
  43.         DefaultMQProducer producer = new DefaultMQProducer(PRODUCER_GROUP);
  44.         // 指定NameServer（生产者需要把提供的消息注册到nameServer，方便消费端查找和订阅）
  45.         producer.setNamesrvAddr(NAMESRV_ADDR);
  46.         producer.start();
  47.  
  48.         List<OrderMsg> orderMsgList = OrderMsg.orderMsgList();
  49.  
  50.         int i = 0;
  51.         for (OrderMsg orderMsg : orderMsgList) {
  52.             Message message = new Message(TOPIC, TAGS, System.currentTimeMillis() + "" + i, ("hello rocketMq" + i++).getBytes());
  53.  
  54.             // 发送消息
  55.             // 参数一：消息对象
  56.             // 参数二：消息队列的选择器
  57.             // 参数三：选择队列的业务标识（此处为订单ID）
  58.             // 其实就是：使用业务标识，进行一定的规则，选出该标识对应存储的队列
  59.             SendResult sendResult = producer.send(message, new MessageQueueSelector() {
  60.                 /**
  61.                  * @param list    消息队列
  62.                  * @param message 消息对象（即上面传递过来的message）
  63.                  * @param o       业务标识的参数（即传递过来的orderId）
  64.                  * @return: org.apache.rocketmq.common.message.MessageQueue
  65.                  */
  66.                 @Override
  67.                 public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
  68.                     int index = Math.abs(o.hashCode()) % (list.size());
  69.                     return list.get(index);
  70.                 }
  71.             }, orderMsg.getId());
  72.             // 上面的orderMsg.getId()这个orderid就是用来确认使用的队列，一般是业务订单id，如果是0的话，就是全局顺序
  73.             System.out.println("发送结果：" + sendResult);
  74.         }
  75.         producer.shutdown();
  76.     }
  77. }
  78.  
  79. class OrderMsg {
  80.     private String id;
  81.     private String msg;
  82.  
  83.     public String getId() {
  84.         return id;
  85.     }
  86.  
  87.     public void setId(String id) {
  88.         this.id = id;
  89.     }
  90.  
  91.     public String getMsg() {
  92.         return msg;
  93.     }
  94.  
  95.     public void setMsg(String msg) {
  96.         this.msg = msg;
  97.     }
  98.  
  99.     public OrderMsg(String id, String msg) {
  100.         this.id = id;
  101.         this.msg = msg;
  102.     }
  103.  
  104.     public static List<OrderMsg> orderMsgList() {
  105.         List<OrderMsg> objects = new ArrayList<>();
  106.         for (int i = 0; i < 20; i++) {
  107.             objects.add(new OrderMsg("12", i + ""));
  108.         }
  109.         return objects;
  110.     }
  111. }

7.2 消费端

  1. package com.zqh.com.rocketmq.orderMsg;
  2.  
  3. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  4. import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
  5. import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
  6. import org.apache.rocketmq.client.exception.MQClientException;
  7. import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
  8. import org.slf4j.Logger;
  9. import org.slf4j.LoggerFactory;
  10. import org.springframework.stereotype.Component;
  11.  
  12. import java.time.LocalDateTime;
  13.  
  14. @Component
  15. public class OrderMsgConsumer {
  16.     private final static Logger logger = LoggerFactory.getLogger(OrderMsgConsumer.class);
  17.     private DefaultMQPushConsumer consumer;
  18.     private static final String CONSUMER_GROUP = "order_group";
  19.     private final static String NAMESRV_ADDR = "192.168.0.30:9876";
  20.     private final static String TOPIC = "order_topic";
  21.     private final static String TAGS = "order_tags";
  22.  
  23.     public OrderMsgConsumer() throws MQClientException {
  24.         System.out.println("初始化成功");
  25.         consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
  26.         consumer.setNamesrvAddr(NAMESRV_ADDR);
  27.         consumer.subscribe(TOPIC, TAGS);
  28.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
  29.         // 注册消息监听器
  30.         // 顺序消息监听 - MessageListenerOrderly 使用该监听器时，只启用一个线程对一个队列进行消费
  31.         // 即：一个队列只会被一个线程取到，第二个线程无法访问这个队列（对消费队列上锁，在消费消息之前，先去获取对应队列对应的锁，保证同一个队列不会被并发消费）
  32.         consumer.registerMessageListener((MessageListenerOrderly) (list, consumeOrderlyContext) -> {
  33.             System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + list.size());
  34.             list.forEach(msg -> {
  35.                         System.out.println(Thread.currentThread().getName() + "\t" + LocalDateTime.now() + "\t\t消费消息：" + new String(msg.getBody()));
  36.                     }
  37.             );
  38.             return ConsumeOrderlyStatus.SUCCESS;
  39.         });
  40.         consumer.start();
  41.     }
  42. }

8.1 事务消息

  1. package com.zqh.com.rocketmq.transactionMsg;
  2.  
  3. import org.apache.rocketmq.client.exception.MQClientException;
  4. import org.apache.rocketmq.client.producer.LocalTransactionState;
  5. import org.apache.rocketmq.client.producer.SendResult;
  6. import org.apache.rocketmq.client.producer.TransactionListener;
  7. import org.apache.rocketmq.client.producer.TransactionMQProducer;
  8. import org.apache.rocketmq.common.message.Message;
  9. import org.apache.rocketmq.common.message.MessageExt;
  10. import org.slf4j.Logger;
  11. import org.slf4j.LoggerFactory;
  12.  
  13. /**
  14.  * 事务信息
  15.  */
  16. public class TransactionMsgProducer {
  17.  
  18.     private final static Logger logger = LoggerFactory.getLogger(TransactionMsgProducer.class);
  19.     private static final String PRODUCER_GROUP = "transaction_group";
  20.     private final static String NAMESRV_ADDR = "192.168.0.30:9876";
  21.     private final static String TOPIC = "transaction_topic";
  22.     private final static String TAGS = "transaction_tags";
  23.  
  24.     public static void sendTransactionMsg() throws MQClientException {
  25.         TransactionMQProducer producer = new TransactionMQProducer(PRODUCER_GROUP);
  26.         producer.setNamesrvAddr(NAMESRV_ADDR);
  27.         producer.setTransactionListener(new TransactionListener() {
  28.             /**
  29.              * @Description 在该方法中执行本地事务
  30.              * @param message   回传的消息，利用transactionId即可获取到该消息的唯一Id
  31.              * @param o         调用send方法时传递的参数，当send时候若有额外的参数可以传递到send方法中，这里能获取到
  32.              * @return: org.apache.rocketmq.client.producer.LocalTransactionState
  33.              *          返回事务状态，COMMIT：提交  ROLLBACK：回滚  UNKNOW：回调
  34.              */
  35.             @Override
  36.             public LocalTransactionState executeLocalTransaction(Message message, Object o) {
  37.                 //这边可以创建一个全局共享变量concurrentHashMap，用来存储transactionId以及对应的值（比如正常执行1，异常执行-1等），便于回查时进行判断，这里就不赘述了
  38.                 System.out.print("执行本地事务" + message.getTransactionId());
  39.                 int i = (int) o;
  40.                 if (i % 3 == 0) {
  41.                     //提交消息
  42.                     System.out.println("提交");
  43.                     return LocalTransactionState.COMMIT_MESSAGE;
  44.                 } else if (i % 3 == 1) {
  45.                     //回滚消息
  46.                     System.out.println("回滚");
  47.                     return LocalTransactionState.ROLLBACK_MESSAGE;
  48.                 }
  49.                 //消息回查
  50.                 System.out.println("回调");
  51.                 return LocalTransactionState.UNKNOW;
  52.             }
  53.  
  54.             /**
  55.              * @Description 事务消息状态回查
  56.              * @param messageExt    通过获取transactionId来判断这条消息的本地事务执行状态
  57.              * @return: org.apache.rocketmq.client.producer.LocalTransactionState
  58.              *          返回事务状态，COMMIT：提交  ROLLBACK：回滚  UNKNOW：回调
  59.              */
  60.             @Override
  61.             public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
  62.                 System.out.println("消息回查：" + messageExt.getTransactionId() + "\t" + new String(messageExt.getBody()));
  63.                 //可以根据消息体中的信息去数据库中查询该消息是否已经被执行
  64.                 //或者根据上方执行本地事务时concurrentHashMap中存储的transactionId对应的值进行判断，返回对应的操作值
  65.  
  66.                 //这里演示就直接提交了
  67.                 return LocalTransactionState.COMMIT_MESSAGE;
  68.             }
  69.         });
  70.         // 启动生产者
  71.         producer.start();
  72.  
  73.         for (int i = 0; i < 10; i++) {
  74.             Message message = new Message(TOPIC, TAGS, System.currentTimeMillis() + "" + i, ("rocketMq" + i).getBytes());
  75.             // 发送事务消息
  76.             // 参数一：消息对象
  77.             // 参数二：附加参数，可用于事务监听器中执行本地事务时的获取参数
  78.             SendResult sendResult = producer.sendMessageInTransaction(message, i);
  79.             System.out.println("发送消息：" + sendResult);
  80.         }
  81.         // 由于MQ要回查生产者，所以不需要关闭生产者
  82. //        producer.shutdown();
  83.     }
  84.  
  85.     public static void main(String[] args) throws Exception {
  86.         sendTransactionMsg();
  87.     }
  88. }

8.2 消费端

  1. package com.zqh.com.rocketmq.transactionMsg;
  2.  
  3. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  4. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
  5. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
  6. import org.apache.rocketmq.client.exception.MQClientException;
  7. import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
  8. import org.slf4j.Logger;
  9. import org.slf4j.LoggerFactory;
  10. import org.springframework.stereotype.Component;
  11.  
  12. @Component
  13. public class TransactionMsgConsumer {
  14.     private final static Logger logger = LoggerFactory.getLogger(TransactionMsgConsumer.class);
  15.     private DefaultMQPushConsumer consumer;
  16.     private static final String CONSUMER_GROUP = "transaction_group";
  17.     private final static String NAMESRV_ADDR = "192.168.0.30:9876";
  18.     private final static String TOPIC = "transaction_topic";
  19.     private final static String TAGS = "transaction_tags";
  20.  
  21.     public TransactionMsgConsumer() throws MQClientException {
  22.         System.out.println("初始化成功");
  23.         consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
  24.         consumer.setNamesrvAddr(NAMESRV_ADDR);
  25.         consumer.subscribe(TOPIC, TAGS);
  26.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
  27.  
  28.         // 并发消费模式 （MessageListenerConcurrently）
  29.         consumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
  30.             System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + list.size());
  31.             list.forEach(msg -> {
  32.                 // 业务实现
  33.                 System.out.println(Thread.currentThread().getName() + "\t" + "组1消费" + msg.getKeys() + new String(msg.getBody()));
  34.             });
  35.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  36.         });
  37.         consumer.start();
  38.     }
  39. }

题外话

1. 事务流程图

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家小花儿/article/detail/256912
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号