RocketMQ顺序消息和事务消息

逃离我推掉我的手 2022-10-26 00:52 2阅读 0赞

# RocketMq有3种消息类型 #

*  **普通消息**
 *  **顺序消息**
 *  **事务消息**

--------------------

## 普通消息 ##

正常发送消费消息即可。

*  **生产者**

//发送同步消息
            rocketMQTemplate.syncSend("topicSend", "Hello World!");
    		
    		//发送同步消息
            rocketMQTemplate.syncSend("topic名称:tags名称", "消息内容");
            User user = new User();
            user.setName("张三");
            rocketMQTemplate.syncSend("topic名称", user);
            
            //发送直接消息
            rocketMQTemplate.sendOneWay("topic名称", "消息内容");
    
            //同步发送并返回string类型结果
            String res = rocketMQTemplate.sendAndReceive("topic名称", "消息内容", String.class);
    
            //发送异步消息
            rocketMQTemplate.asyncSend("topic名称", "消息内容", new SendCallback() { 
                @Override
                public void onSuccess(SendResult sendResult) { 
                    System.out.println("成功");
                }
    
                @Override
                public void onException(Throwable throwable) { 
                    System.out.println("失败");
                }
            });

*  **消费者**

/** * @Author: LailaiMonkey * @Description： * @Date：Created in 2021-01-25 16:19 * @Modified By： */
    @Component
    @RocketMQMessageListener(nameServer = "nameServer", topic = "topicSend", consumerGroup = "test_consumer")
    public class RocketMqConsumer implements RocketMQListener<String> { 
    
        @Override
        public void onMessage(String message) { 
            System.out.println("消息：" + message);
        }
    }

## 顺序消息 ##

转账等业务时需要保证消息顺序性，RocketMQ顺序性不是消费者保证，而是生产者保证。

生产者发送消息时需要指定路由Key，RocketMQ会自动分配到同一MessageQueue中。

*  **生产者**

//发送顺序消息，根据hashKey的hash值路由到同一个brock中，消费者需要设置ConsumeMode.ORDERLY顺序消费
            rocketMQTemplate.syncSendOrderly("topicSend1", "消息内容1", "ORDERLY");
            rocketMQTemplate.syncSendOrderly("topicSend1", "消息内容2", "ORDERLY");
            rocketMQTemplate.syncSendOrderly("topicSend1", "消息内容3", "ORDERLY");
            rocketMQTemplate.syncSendOrderly("topicSend1", "消息内容4", "ORDERLY");
            rocketMQTemplate.syncSendOrderly("topicSend1", "消息内容5", "ORDERLY");

*  **消费者**

/** * @Author: LailaiMonkey * @Description： * @Date：Created in 2021-01-25 16:19 * @Modified By： */
    @Component
    @RocketMQMessageListener(nameServer = "nameServer", topic = "topicSend", consumerGroup = "test_consumer_transaction", consumeMode = ConsumeMode.ORDERLY)
    public class RocketMqConsumerTransaction implements RocketMQListener<String> { 
    
        @Override
        public void onMessage(String message) { 
            System.out.println("这是消息内容：" + message);
        }
    }

## 事务消息 ##

事务消息可以保证执行本地事务和发消息操作是原子性。

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

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNzM5Nzk1ODY_size_16_color_FFFFFF_t_70]

*  **生产者**

Message<String> message = MessageBuilder.withPayload("这是事务消息")
                    .build();
            rocketMQTemplate.sendMessageInTransaction("topicSend", message, "topicSend");

*  **全局配置**

/** * @Author: LailaiMonkey * @Description： * @Date：Created in 2021-01-30 14:09 * @Modified By： */
    @RocketMQTransactionListener
    public class RocketMqTransaction implements RocketMQLocalTransactionListener { 
    
        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) { 
        	//如果状态为已提交则不会走回查方法
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    
        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message msg) { 
            System.out.println("这是消息内容：" + message);
            return RocketMQLocalTransactionState.COMMIT;
        }
    
    }

/** * @Author: LailaiMonkey * @Description： * @Date：Created in 2021-01-25 16:19 * @Modified By： */
    @Component
    @RocketMQMessageListener(nameServer = "nameServer", topic = "topicSend", consumerGroup = "test_consumer_transaction")
    public class RocketMqConsumer implements RocketMQListener<String> { 
    
        @Override
        public void onMessage(String message) { 
            System.out.println("消息：" + message);
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNzM5Nzk1ODY_size_16_color_FFFFFF_t_70]: /images/20221024/91b67f7e7cea4e19b64ba0332a1d8852.png