RocketMQ整合Java与springboot

1 导入依赖

<!-- https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-client -->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.7.0</version>
</dependency>

2 消息生产者步骤

  1. 创建消息生产者producer，并指定生产者组名
  2. 指定Nameserver地址
  3. 启动producer
  4. 创建消息对象，指定主题Topic、Tag和消息体
  5. 发送消息
  6. 关闭生产者producer

3 消息消费者步骤

  1. 创建消费者Consumer，制定消费者组名
  2. 指定Nameserver地址
  3. 订阅主题Topic和Tag
  4. 设置回调函数，处理消息
  5. 启动消费者consumer

4 消息发送样例

4.1 生产消息

发送同步消息：
这种可靠性同步地发送方式使用的比较广泛，比如：重要的消息通知，短信通知。

//发送同步消息
public class SyncProducer {
    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        //1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.85.128:9876;192.168.85.129:9876");
        //3.启动producer
        producer.start();
        //4.创建消息对象，指定主题Topic、Tag和消息体
        for (int i = 0; i < 10; i++) {
            /**
             * Message(String topic, String tags, byte[] body)
             * 消息主题，消息tag，消息内容
             */
            Message msg = new Message("base","tag1",("hello world同"+i).getBytes());
            //5.发送同步消息
            SendResult result = producer.send(msg);
            String msgId = result.getMsgId();//消息id
            SendStatus sendStatus = result.getSendStatus();//发送状态
            int queueId = result.getMessageQueue().getQueueId();//消息队列id
            System.out.println(result);
            TimeUnit.SECONDS.sleep(1);
        }
        //6.关闭生产者producer
        producer.shutdown();
    }
}

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

//发送异步消息
public class AsyncProducer {
    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        //1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.85.128:9876;192.168.85.129:9876");
        //3.启动producer
        producer.start();
        //4.创建消息对象，指定主题Topic、Tag和消息体
        for (int i = 0; i < 10; i++) {
            /**
             * Message(String topic, String tags, byte[] body)
             * 消息主题，消息tag，消息内容
             */
            Message msg = new Message("base","tag2",("hello world异"+i).getBytes());
            //5.发送异步消息
            producer.send(msg, new SendCallback() {
                //发送成功回调
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println(sendResult);
                }
                //发送异常回调
                @Override
                public void onException(Throwable throwable) {
                    System.out.println(throwable);
                }
            });
            TimeUnit.SECONDS.sleep(1);
        }
        //6.关闭生产者producer
        producer.shutdown();
    }
}

单向发送消息
这种方式主要用在不特别关心发送结果的场景，例如日志发送。

//发送单向消息
public class OneWayProducer {
    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        //1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.85.128:9876;192.168.85.129:9876");
        //3.启动producer
        producer.start();
        //4.创建消息对象，指定主题Topic、Tag和消息体
        for (int i = 0; i < 10; i++) {
            /**
             * Message(String topic, String tags, byte[] body)
             * 消息主题，消息tag，消息内容
             */
            Message msg = new Message("base","tag3",("hello world单"+i).getBytes());
            //5.发送单向消息
            producer.sendOneway(msg);
            TimeUnit.SECONDS.sleep(1);
        }
        //6.关闭生产者producer
        producer.shutdown();
    }
}

4.2 消费消息

负载均衡模式（默认消费模式）
消费者采用负载均衡方式消费消息，多个消费者共同消费队列消息，每个消费者处理的消息不同，比如说一共发了10条数据，3个消费者消费，则可能分别接收到的数据条数为2，5，3，一共10条

//负载均衡模式
public class Consumer {
    public static void main(String[] args) throws MQClientException {
        //1.创建消费者Consumer，指定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.85.128:9876;192.168.85.129:9876");
        //3.订阅主题Topic和Tag
        consumer.subscribe("base","tag2");
        //负载均衡模式消费
        consumer.setMessageModel(MessageModel.CLUSTERING);
        //4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            //接收消息内容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt messageExt : list) {
                    System.out.println(new String(messageExt.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.启动消费者consumer
        consumer.start();
    }
}

广播模式
消费者采用广播的方式消费消息，每个消费者消费的消息都是相同的。即每个消费者都收到10条

//广播模式
public class Consumer {
    public static void main(String[] args) throws MQClientException {
        //1.创建消费者Consumer，指定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.85.128:9876;192.168.85.129:9876");
        //3.订阅主题Topic和Tag
        consumer.subscribe("base","tag2");
        //负载广播模式消费
        consumer.setMessageModel(MessageModel.BROADCASTING);
        //4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            //接收消息内容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt messageExt : list) {
                    System.out.println(new String(messageExt.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.启动消费者consumer
        consumer.start();
    }
}

5 顺序消息

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

在默认的情况下消息发送会采取轮询方式把消息发送到 不同 的queue(分区队列)；而消费消息的时候从 多个 queue上拉取消息，这种情况发送和消费是不能保证顺序的。但是如果控制发送的顺序消息只依次发送到同一个queue中，消费的时候只从这个queue上依次拉取，则就保证了顺序。当发送和消费参与的queue只有一个，则是全局有序；如果多个queue参与，则为分区有序，即相对每个queue，消息都是有序的。

代码模拟：

/**
 * 订单的步骤
 */
class OrderStep {
    private long orderId;
    private String desc;

    public long getOrderId() {
        return orderId;
    }

    public void setOrderId(long orderId) {
        this.orderId = orderId;
    }

    public String getDesc() {
        return desc;
    }

    public void setDesc(String desc) {
        this.desc = desc;
    }

    @Override
    public String toString() {
        return "OrderStep{" +
                "orderId=" + orderId +
                ", desc='" + desc + '\'' +
                '}';
    }
    /**
     * 生成模拟订单数据
     */
    public static List<OrderStep> buildOrders() {
        List<OrderStep> orderList = new ArrayList<OrderStep>();

        OrderStep orderDemo = new OrderStep();
        orderDemo.setOrderId(15103111039L);
        orderDemo.setDesc("创建");
        orderList.add(orderDemo);

        orderDemo = new OrderStep();
        orderDemo.setOrderId(15103111065L);
        orderDemo.setDesc("创建");
        orderList.add(orderDemo);

        orderDemo = new OrderStep();
        orderDemo.setOrderId(15103111039L);
        orderDemo.setDesc("付款");
        orderList.add(orderDemo);

        orderDemo = new OrderStep();
        orderDemo.setOrderId(15103117235L);
        orderDemo.setDesc("创建");
        orderList.add(orderDemo);

        orderDemo = new OrderStep();
        orderDemo.setOrderId(15103111065L);
        orderDemo.setDesc("付款");
        orderList.add(orderDemo);

        orderDemo = new OrderStep();
        orderDemo.setOrderId(15103117235L);
        orderDemo.setDesc("付款");
        orderList.add(orderDemo);

        orderDemo = new OrderStep();
        orderDemo.setOrderId(15103111065L);
        orderDemo.setDesc("完成");
        orderList.add(orderDemo);

        orderDemo = new OrderStep();
        orderDemo.setOrderId(15103111039L);
        orderDemo.setDesc("推送");
        orderList.add(orderDemo);

        orderDemo = new OrderStep();
        orderDemo.setOrderId(15103117235L);
        orderDemo.setDesc("完成");
        orderList.add(orderDemo);

        orderDemo = new OrderStep();
        orderDemo.setOrderId(15103111039L);
        orderDemo.setDesc("完成");
        orderList.add(orderDemo);

        return orderList;
    }
}

public class Producer {
    public static void main(String[] args) throws Exception{
        //1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.85.128:9876;192.168.85.129:9876");
        //3.启动producer
        producer.start();
        //构建消息集合
        List<OrderStep> orderSteps = OrderStep.buildOrders();

        //发送消息
        for (int i=0;i<orderSteps.size();i++) {
            java.lang.String body = orderSteps.get(i)+"";
            Message message = new Message("OrderTopic","Order","i"+i,body.getBytes());
            //参数一：消息队列，参数二:消息队列的选择器，参数三，选择队列的业务标识（订单ID）
            SendResult sendResult = producer.send(message, new MessageQueueSelector() {
                /**
                 *
                 * @param mqs 队列集合
                 * @param message  消息对象
                 * @param o 业务标识的参数
                 * @return
                 */
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message message, Object o) {
                    long orderId = (long) o;
                    //为了让同一个人的信息存储在同一个顺序队列中
                    long index = orderId % mqs.size();
                    return mqs.get((int) index);
                }
            }, orderSteps.get(i).getOrderId());

            System.out.println("发送结果："+sendResult);
        }
        producer.shutdown();
    }
}

public class Consumer {
    public static void main(String[] args) throws Exception {
        //1.创建消费者Consumer，指定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.85.128:9876;192.168.85.129:9876");
        //3.订阅主题Topic和Tag
        consumer.subscribe("OrderTopic","*");

        //4.注册消息监听器
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                for (MessageExt messageExt : list) {
                    System.out.println("线程名称"+Thread.currentThread().getName()+"消费消息："+new String(messageExt.getBody()));
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        //启动消费者
        consumer.start();
        System.out.println("消费者启动");
    }
}

查看执行结果，可以发现同一个人的步骤都是有序的。

image.png

6 延时消息

比如京东提交了一个订单就可以发送一个延时消息，1h后去检查这个订单的是否付款，如果是未付款就取消订单释放库存。

延时级别:
现在RocketMq并不支持任意时间的延时，需要设置几个固定的延时等级，从1s到2h分别对应着等级1到18

// org/apache/rocketmq/store/config/MessageStoreConfig.java
private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

测试代码：

public class Producer {
    public static void main(String[] args) throws Exception{
        //1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.85.128:9876;192.168.85.129:9876");
        //3.启动producer
        producer.start();
        //4.创建消息对象，指定主题Topic、Tag和消息体
        for (int i = 0; i < 10; i++) {
            /**
             * Message(String topic, String tags, byte[] body)
             * 消息主题，消息tag，消息内容
             */
            Message msg = new Message("DelayTopic","tag1",("hello world同"+i).getBytes());
            msg.setDelayTimeLevel(2);
            //5.发送同步消息
            SendResult result = producer.send(msg);
            String msgId = result.getMsgId();//消息id
            SendStatus sendStatus = result.getSendStatus();//发送状态
            int queueId = result.getMessageQueue().getQueueId();//消息队列id
            System.out.println(result);
            TimeUnit.SECONDS.sleep(1);
        }
        //6.关闭生产者producer
        producer.shutdown();
    }
}

public class Consumer {
    public static void main(String[] args) throws MQClientException {
        //1.创建消费者Consumer，指定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.85.128:9876;192.168.85.129:9876");
        //3.订阅主题Topic和Tag
        consumer.subscribe("DelayTopic","*");
        //4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            //接收消息内容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt messageExt : list) {
                    System.out.println("消息id:"+messageExt.getMsgId()+"延迟时间："+(System.currentTimeMillis()-messageExt.getStoreTimestamp()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.启动消费者consumer
        consumer.start();
    }
}

7 批量消息

批量发送消息能显著提高传递小消息的性能。限制是这些批量消息应该有相同的topic，相同的waitStoreMsgOK，而且不能是延时消息。此外，这一批消息的总大小不应超过1MB。
代码测试：

public class Consumer {
    public static void main(String[] args) throws MQClientException {
        //1.创建消费者Consumer，指定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.85.128:9876;192.168.85.129:9876");
        //3.订阅主题Topic和Tag
        consumer.subscribe("BatchTopic","*");
        //4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            //接收消息内容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt messageExt : list) {
                    System.out.println(new String(messageExt.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.启动消费者consumer
        consumer.start();
    }
}

public class Producer {
    public static void main(String[] args) throws Exception {
        //1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.85.128:9876;192.168.85.129:9876");
        //3.启动producer
        producer.start();
        //4.创建消息对象，指定主题Topic、Tag和消息体

        List<Message> msgs = new ArrayList<>();

        /**
         * Message(String topic, String tags, byte[] body)
         * 消息主题，消息tag，消息内容
         */
        Message msg1 = new Message("BatchTopic", "tag1", ("hello world同" + 1).getBytes());
        Message msg2 = new Message("BatchTopic", "tag1", ("hello world同" + 2).getBytes());
        Message msg3 = new Message("BatchTopic", "tag1", ("hello world同" + 3).getBytes());

        msgs.add(msg1);
        msgs.add(msg2);
        msgs.add(msg3);
        //5.发送批量消息
        SendResult result = producer.send(msgs);
        String msgId = result.getMsgId();//消息id
        SendStatus sendStatus = result.getSendStatus();//发送状态
        int queueId = result.getMessageQueue().getQueueId();//消息队列id
        System.out.println(result);
        TimeUnit.SECONDS.sleep(1);

        //6.关闭生产者producer
        producer.shutdown();
    }
}

若长度超过1M，则需要将消息分割

public class ListSplitter implements Iterator<List<Message>> {
   private final int SIZE_LIMIT = 1024 * 1024;
   private final List<Message> messages;
   private int currIndex;
   public ListSplitter(List<Message> messages) {
           this.messages = messages;
   }
    @Override 
    public boolean hasNext() {
       return currIndex < messages.size();
   }
    @Override 
    public List<Message> next() {
       int nextIndex = currIndex;
       int totalSize = 0;
       for (; nextIndex < messages.size(); nextIndex++) {
           Message message = messages.get(nextIndex);
           int tmpSize = message.getTopic().length() + message.getBody().length;
           Map<String, String> properties = message.getProperties();
           for (Map.Entry<String, String> entry : properties.entrySet()) {
               tmpSize += entry.getKey().length() + entry.getValue().length();
           }
           tmpSize = tmpSize + 20; // 增加日志的开销20字节
           if (tmpSize > SIZE_LIMIT) {
               //单个消息超过了最大的限制
               //忽略,否则会阻塞分裂的进程
               if (nextIndex - currIndex == 0) {
                  //假如下一个子列表没有元素,则添加这个子列表然后退出循环,否则只是退出循环
                  nextIndex++;
               }
               break;
           }
           if (tmpSize + totalSize > SIZE_LIMIT) {
               break;
           } else {
               totalSize += tmpSize;
           }

       }
       List<Message> subList = messages.subList(currIndex, nextIndex);
       currIndex = nextIndex;
       return subList;
   }
}
//把大的消息分裂成若干个小的消息
ListSplitter splitter = new ListSplitter(messages);
while (splitter.hasNext()) {
  try {
      List<Message>  listItem = splitter.next();
      producer.send(listItem);
  } catch (Exception e) {
      e.printStackTrace();
      //处理error
  }
}

8 过滤消息

在大多数情况下，TAG是一个简单而有用的设计，其可以来选择您想要的消息。例如：

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_EXAMPLE");
consumer.subscribe("TOPIC", "TAGA || TAGB || TAGC");

消费者将接收包含TAGA或TAGB或TAGC的消息。但是限制是一个消息只能有一个标签，这对于复杂的场景可能不起作用。在这种情况下，可以使用SQL表达式筛选消息。SQL特性可以通过发送消息时的属性来进行计算。在RocketMQ定义的语法下，可以实现一些简单的逻辑。

------------
| message  |
|----------|  a > 5 AND b = 'abc'
| a = 10   |  --------------------> Gotten
| b = 'abc'|
| c = true |
------------
------------
| message  |
|----------|   a > 5 AND b = 'abc'
| a = 1    |  --------------------> Missed
| b = 'abc'|
| c = true |
------------

8.1 SQL基本语法

RocketMQ只定义了一些基本语法来支持这个特性。你也可以很容易地扩展它。

  • 数值比较，比如：>，>=，<，<=，BETWEEN，=；
  • 字符比较，比如：=，<>，IN；
  • IS NULL 或者 IS NOT NULL；
  • 逻辑符号 AND，OR，NOT；

常量支持类型为：

  • 数值，比如：123，3.1415；
  • 字符，比如：'abc'，必须用单引号包裹起来；
  • NULL，特殊的常量
  • 布尔值，TRUE 或 FALSE

测试代码如下：

public class Producer {
    public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        //1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.85.128:9876;192.168.85.129:9876");
        //3.启动producer
        producer.start();
        //4.创建消息对象，指定主题Topic、Tag和消息体
        for (int i = 0; i < 10; i++) {
            /**
             * Message(String topic, String tags, byte[] body)
             * 消息主题，消息tag，消息内容
             */
            Message msg = new Message("SqlTagTopic","Tag1",("hello world同"+i).getBytes());
            msg.putUserProperty("i",String.valueOf(i));
            //5.发送同步消息
            SendResult result = producer.send(msg);
            String msgId = result.getMsgId();//消息id
            SendStatus sendStatus = result.getSendStatus();//发送状态
            int queueId = result.getMessageQueue().getQueueId();//消息队列id
            System.out.println(result);
            TimeUnit.SECONDS.sleep(1);
        }
        //6.关闭生产者producer
        producer.shutdown();
    }
}

public class Consumer {
    public static void main(String[] args) throws MQClientException {
        //1.创建消费者Consumer，指定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.85.128:9876;192.168.85.129:9876");
        //3.订阅主题Topic和Tag
        consumer.subscribe("SqlTagTopic", MessageSelector.bySql("i>5"));
        //广播模式消费
//        consumer.setMessageModel(MessageModel.BROADCASTING);
        //负载均衡模式消费（默认）
//        consumer.setMessageModel(MessageModel.CLUSTERING);
        //4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            //接收消息内容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt messageExt : list) {
                    System.out.println(new String(messageExt.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.启动消费者consumer
        consumer.start();
        System.out.println("消费者启动");
    }
}

9 事务消息

9.1 流程

事务消息的大致方案，其中分为两个流程：正常事务消息的发送及提交、事务消息的补偿流程，如下图。

事务消息

1）事务消息发送及提交

(1) 发送消息（half消息）。

(2) 服务端响应消息写入结果。

(3) 根据发送结果执行本地事务（如果写入失败，此时half消息对业务不可见，本地逻辑不执行）。

(4) 根据本地事务状态执行Commit或者Rollback（Commit操作生成消息索引，消息对消费者可见）

2）事务补偿

(1) 对没有Commit/Rollback的事务消息（pending状态的消息），从服务端发起一次“回查”

(2) Producer收到回查消息，检查回查消息对应的本地事务的状态

(3) 根据本地事务状态，重新Commit或者Rollback

其中，补偿阶段用于解决消息Commit或者Rollback发生超时或者失败的情况。

3）事务消息状态

事务消息共有三种状态，提交状态、回滚状态、中间状态：

  • TransactionStatus.CommitTransaction: 提交事务，它允许消费者消费此消息。
  • TransactionStatus.RollbackTransaction: 回滚事务，它代表该消息将被删除，不允许被消费。
  • TransactionStatus.Unknown: 中间状态，它代表需要检查消息队列来确定状态。

9.2 代码测试

public class Producer {
    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        //1.创建消息生产者producer，并制定生产者组名
        TransactionMQProducer producer = new TransactionMQProducer("group1");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.85.128:9876;192.168.85.129:9876");
        //3.设置生产者事务监听器
        producer.setTransactionListener(new TransactionListener() {
            /**
             * 该方法中执行本地事务
             * @param message
             * @param o
             * @return
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                if(StringUtils.equals("TAGA",message.getTags())){
                    return LocalTransactionState.COMMIT_MESSAGE;
                }else if(StringUtils.equals("TAGB",message.getTags())){
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }else{
                    return LocalTransactionState.UNKNOW;
                }
            }

            /**
             * 该方法是MQ进行消息事务状态的回查
             * @param messageExt
             * @return
             */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println("当前消息的tag:"+messageExt.getTags());
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        //4.启动producer
        producer.start();
        String[] tag = {"TAGA","TAGB","TAGC"};
        //4.创建消息对象，指定主题Topic、Tag和消息体
        for (int i = 0; i < 3; i++) {
            /**
             * Message(String topic, String tags, byte[] body)
             * 消息主题，消息tag，消息内容
             */
            Message msg = new Message("TransactionTopic",tag[i],("hello world"+i).getBytes());
            //5.发送同步消息
            SendResult result = producer.sendMessageInTransaction(msg,null);
            System.out.println(result);
            TimeUnit.SECONDS.sleep(1);
        }
    }
}

public class Consumer {
    public static void main(String[] args) throws MQClientException {
        //1.创建消费者Consumer，指定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("192.168.85.128:9876;192.168.85.129:9876");
        //3.订阅主题Topic和Tag
        consumer.subscribe("TransactionTopic","*");
        //4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            //接收消息内容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt messageExt : list) {
                    System.out.println(new String(messageExt.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.启动消费者consumer
        consumer.start();
        System.out.println("消费者已启动");
    }
}

9.3 使用限制

  1. 事务消息 不支持 延时消息和批量消息。
  2. 为了避免单个消息被检查太多次而导致半队列消息累积，我们默认将单个消息的检查次数限制为 15 次，但是用户可以通过 Broker 配置文件的 transactionCheckMax参数来修改此限制。如果已经检查某条消息超过 N 次的话（ N = transactionCheckMax ） 则 Broker 将丢弃此消息，并在默认情况下同时打印错误日志。用户可以通过重写 AbstractTransactionCheckListener 类来修改这个行为。
  3. 事务消息将在 Broker 配置文件中的参数 transactionMsgTimeout 这样的特定时间长度之后被检查。当发送事务消息时，用户还可以通过设置用户属性 CHECK_IMMUNITY_TIME_IN_SECONDS 来改变这个限制，该参数优先于 transactionMsgTimeout 参数。
  4. 事务性消息可能不止一次被检查或消费。
  5. 提交给用户的目标主题消息可能会失败，目前这依日志的记录而定。它的高可用性通过 RocketMQ 本身的高可用性机制来保证，如果希望确保事务消息不丢失、并且事务完整性得到保证，建议使用同步的双重写入机制。
  6. 事务消息的生产者 ID 不能与其他类型消息的生产者 ID 共享。与其他类型的消息不同，事务消息允许反向查询、MQ服务器能通过它们的生产者 ID 查询到消费者。

10 整合springboot

10.1 导入依赖

        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.1.0</version>
        </dependency>

10.2 编写配置文件

rocketmq.name-server=192.168.25.135:9876;192.168.25.138:9876
rocketmq.producer.group=my-group

10.3 消息生产者

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Test
    void testProducer() {
        rocketMQTemplate.convertAndSend("springboot-rocketmq","hello worldaaa");
    }

10.4 消费消息

@RocketMQMessageListener(topic = "springboot-rocketmq",consumerGroup = "${rocketmq.producer.group}")
@Component
public class Consumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String s) {
        System.out.println("收到的消息为"+s);
    }
}

这里监听到了刚刚生产的消息


image.png
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容

  • RocketMQ（二）RocketMQ高级特性
    一、 关键特性 1 消息发送和消费 1)消息发送者步骤分析: 创建消息生产者producer，并制定生产者组名 指...
    TiaNa_na阅读 2,063评论 0赞 2
  • RocketMQ学习记录
    各位在面试过程中比较被常问到的中间件可能就有MQ吧，提起MQ，不能说都用过，但是也都听过吧，实在不行叫消息...
    少年丶要淡定阅读 634评论 1赞 5
  • java中间件之rocketmq
    一、RocketMq简介 1.1 RocketMq是什么 RcoketMQ是一款低延迟、高可靠、可伸缩、易于使用的...
    这一刻_776b阅读 3,000评论 0赞 0
  • 消息队列之 RocketMQ
    简介 RocketMQ 特点 RocketMQ 是阿里巴巴在2012年开源的分布式消息中间件，目前已经捐赠给 Ap...
    预流阅读 39,246评论 7赞 55
  • rocketmq 与 kafka 对比漫谈
      今年的一个周末，去参加了一场rocketMq的meet up分享，由此对rocketMq产生了极大的兴趣，ro...
    左小星阅读 16,075评论 7赞 28