RocketMQ单机版详解，事务消息、顺序消息、转换消息、可回复消息等

朱雀 2022-10-31 00:45 13阅读 0赞

### 文章目录 ###

*   *  一、说明
     *   *  1-1、安装和快速测试
         *  1-2、下面的代码使用的版本
         *  1-3、相关文档
     *  二、RocketMQ组件
     *   *  2-1、封装RocketMQTemplate
         *  2-2、生产者（Producer）
         *  2-3、消费者（consumer）
         *  2-3-1、push消费
         *   *  2-3-1-1、获取不需要返回值的消息
             *  2-3-1-2、获取需要返回值的消息
         *  2-3-2、pull消费
         *   *  2-3-2-1、默认的消费者
             *  2-3-2-2、自定义消费者
             *  2-3-2-3、其它
         *  2-3、发送消息
         *   *  2-3-1、普通消息
             *  2-3-2、可接受回复的消息
             *  2-3-3、单向消息
             *  2-3-4、单向顺序消息
             *  2-3-5、同步消息
             *  2-3-6、异步消息
             *  2-3-7、转换消息
             *  2-3-8、事务消息
     *  三、源码

## 一、说明 ##

### 1-1、安装和快速测试 ###

*  [Linux安装RocketMQ【阿里云ESC，CentOS7+,踩坑无数】][Linux_RocketMQ_ESC_CentOS7]
 *  [SpringBoot整合RocketMQ，三种测试附带源码【rocketmq-spring-boot-starter】][SpringBoot_RocketMQ_rocketmq-spring-boot-starter]

### 1-2、下面的代码使用的版本 ###

*  SpringBoot   2.2.0.RELEASE
 *  RocketMQ   4.6.1
 *  Rocketmq-spring-boot-starter   2.2.0

### 1-3、相关文档 ###

*  [RocketMQ中文文档][RocketMQ]
 *  [Rocketmq-spring-boot-starter文档][Rocketmq-spring-boot-starter]

## 二、RocketMQ组件 ##

### 2-1、封装RocketMQTemplate ###

为了使用的方便，我们把这RocketMQTemplate进行静态化

import org.apache.rocketmq.spring.core.RocketMQTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    /** * rocketMQ生产者 */
    @Component
    public class ProducerUtils { 
        public static RocketMQTemplate rocketMQTemplate;
    
        @Autowired
        public ProducerUtils(RocketMQTemplate rocketMQTemplate) { 
            ProducerUtils.rocketMQTemplate = rocketMQTemplate;
        }
    }

### 2-2、生产者（Producer） ###

下面发消息的时候一起用

### 2-3、消费者（consumer） ###

消费者可以主动消费（pull），也可以被动消费（push）

如果我们的消费者压力不大的话，可以选择被动消费，这样消息没有延迟，而且处理起来简单。

### 2-3-1、push消费 ###

> 也就是当有消息来的时候，broker会把消息推送给我们消费者

#### 2-3-1-1、获取不需要返回值的消息 ####

import org.apache.rocketmq.common.message.MessageExt;
    import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
    import org.apache.rocketmq.spring.core.RocketMQListener;
    import org.springframework.stereotype.Service;
    
    
    @Service
    @RocketMQMessageListener(consumerGroup = "my-consumer_sync-topic_two", topic = "topic")
    public class Consumer implements RocketMQListener<MessageExt> { 
    
        public void onMessage(MessageExt message) { 
            // 1、这个message里面包含了很多消息，具体可以点进去查看
            System.out.println(message);
            // 2、这个body才是生产者发送的具体消息
            System.out.println(new String(message.getBody()));
        }
    }

#### 2-3-1-2、获取需要返回值的消息 ####

RocketMQReplyListener的第二个泛型就是返回值类型

import org.apache.rocketmq.common.message.MessageExt;
    import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
    import org.apache.rocketmq.spring.core.RocketMQReplyListener;
    import org.springframework.stereotype.Service;
    
    
    @Service
    @RocketMQMessageListener(consumerGroup = "my-consumer_sync-topic_two", topic = "topic")
    public class Consumer implements RocketMQReplyListener<MessageExt,String> { 
    
        public String onMessage(MessageExt message) { 
            // 1、这个message里面包含了很多消息，具体可以点进去查看
            System.out.println(message);
            // 2、这个body才是生产者发送的具体消息
            System.out.println(new String(message.getBody()));
    
            return "返回值";
        }
    }

### 2-3-2、pull消费 ###

#### 2-3-2-1、默认的消费者 ####

@GetMapping("/test2")
    public String fun2() { 
        List<MessageExt> receive = ProducerUtils.rocketMQTemplate.receive(MessageExt.class);
        for (MessageExt item : receive){ 
            System.out.println(item);
        }
        return "成功";
    }

*  泛型传递 MessageExt 就获取全部的消息信息
 *  泛型传递消息类型，比如String，那么就只返回我们生产者发送的消息了

eg：如果我们改成这样的 List receive = ProducerUtils.rocketMQTemplate.receive(String.class); 那么打印结果就是 message

#### 2-3-2-2、自定义消费者 ####

可以对自定义的消费者进行设置，比如线程池的个数、消费的topic等

@Component
    public class ProducerUtils { 
    
        public static RocketMQTemplate rocketMQTemplate;
    
        @Autowired
        public ProducerUtils(RocketMQTemplate rocketMQTemplate) throws MQClientException { 
            // 设置默认的消费者 主要要自定义消费者组名称，不然重复了
            DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("MY_CONSUMER");
    	// 自定义消费的topic
            consumer.subscribe("topicA","*");
            rocketMQTemplate.setConsumer(consumer);
            // 开启消费者
            rocketMQTemplate.getConsumer().start();
            ProducerUtils.rocketMQTemplate = rocketMQTemplate;
        }
    }

#### 2-3-2-3、其它 ####

主动消费的话有两种方式

*  一种是写定时任务，定时去执行我们的消费方法
 *  另一种是轮询，我们写一个 while(true) 反复去获取最新的消息

轮询的方式还是相对好一些，可以及时的去消费。我们可以开一个线程池去轮询，每次没有获取到最新的消息可以让线程休息一段时间。

### 2-3、发送消息 ###

#### 2-3-1、普通消息 ####

携带topic的普通消息

Message<String> message = MessageBuilder.withPayload("message").build();
    ProducerUtils.rocketMQTemplate.send("topic",message);

不携带topic的普通消息（需要在创建生产者的时候设置默认的topic不然会报错）

ProducerUtils.rocketMQTemplate.send(message);

#### 2-3-2、可接受回复的消息 ####

如标题所示此类消息，可以接受消费者返回的数据。发送的方法为xxxxReceive

<table> 
 <thead> 
  <tr> 
   <th>参数名</th> 
   <th>含义</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>destination</td> 
   <td>topic名称</td> 
  </tr> 
  <tr> 
   <td>payload</td> 
   <td>消息内容</td> 
  </tr> 
  <tr> 
   <td>type</td> 
   <td>返回的数据类型</td> 
  </tr> 
  <tr> 
   <td>message</td> 
   <td>消息内容</td> 
  </tr> 
  <tr> 
   <td>timeout</td> 
   <td>超时时间，大于这个时间没有返回数据就超时，默认3000ms</td> 
  </tr> 
  <tr> 
   <td>delayLevel</td> 
   <td>延迟消息 DelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h ，0=0s，1=1s，2=5s…</td> 
  </tr> 
  <tr> 
   <td>hashKey</td> 
   <td>通过源码我们可以查看这个是用来选择发送队列的，如果hashKey一样那么队列也一样</td> 
  </tr> 
  <tr> 
   <td>rocketMQLocalRequestCallback</td> 
   <td>自定义接受回调和异常处理</td> 
  </tr> 
 </tbody> 
</table>

知道了上面的参数含义，调用起来就很简单了，这里写一个自定义回调的demo

import org.apache.rocketmq.spring.core.RocketMQLocalRequestCallback;
    
    public class RequestCallback implements RocketMQLocalRequestCallback<String> { 
        public void onSuccess(String s) { 
            System.out.println("啊！我接收到了消息 : " + s);
        }
    
        public void onException(Throwable throwable) { 
            String message = throwable.getMessage();
            System.out.println("啊！我出现了异常 " + message);
        }
    }
    
    RequestCallback requestCallback = new RequestCallback();
    ProducerUtils.rocketMQTemplate.sendAndReceive("topic", "message",requestCallback);

#### 2-3-3、单向消息 ####

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

ProducerUtils.rocketMQTemplate.sendOneWay("topic","message");

#### 2-3-4、单向顺序消息 ####

从上面我们知道指定hashKey就指定了一个队列发送，而队列的消费是顺序的，所以这样就是顺序消息了

ProducerUtils.rocketMQTemplate.sendOneWayOrderly("topic","message","1");

#### 2-3-5、同步消息 ####

SendResult topic = ProducerUtils.rocketMQTemplate.syncSend("topic", "sync-message");
    System.out.println(topic);

使用 syncSendOrderly 来发送顺序的同步消息。

#### 2-3-6、异步消息 ####

异步消息返回的结果和同步一样（SendResult），只是不需要及时返回，我们需要实现SendCallback接口

import org.apache.rocketmq.client.producer.SendCallback;
    import org.apache.rocketmq.client.producer.SendResult;
    
    public class RequestCallback implements SendCallback { 
    
        public void onSuccess(SendResult sendResult) { 
            System.out.println("异步返回结果：" + sendResult);
        }
    
        public void onException(Throwable throwable) { 
            System.out.println("啊、我异常了 " + throwable.getMessage());
        }
    }
    
    
    RequestCallback requestCallback = new RequestCallback();
    ProducerUtils.rocketMQTemplate.asyncSend("topic","async-miss", requestCallback);

当然我们也可以使用sendOneWayOrderly来发送顺序消息，一样的加上hashKey就可以。

#### 2-3-7、转换消息 ####

可以理解成某些消息我们需要经过特殊的处理然后再发给消费者。

Map<String, Object> headers 可以理解成特殊处理时候的配置文件。

MessagePostProcessor 特殊处理的处理器

import org.springframework.messaging.Message;
    import org.springframework.messaging.core.MessagePostProcessor;
    import org.springframework.messaging.support.MessageBuilder;
    
    public class MessageHandler implements MessagePostProcessor { 
        public Message<?> postProcessMessage(Message<?> message) { 
            Object o = message.getHeaders().get("123");
            Message<String> message1 = MessageBuilder.withPayload(message.getPayload() + o.toString()).build();
            // 这个message1 才是最终发给消费者的
            return message1;
        }
    }
    
    
    MessageHandler messageHandler = new MessageHandler();
    Map<String, Object> map = new HashMap<String, Object>();
    map.put("123","12345");
    ProducerUtils.rocketMQTemplate.convertAndSend("topic","wa messa",map,messageHandler);

#### 2-3-8、事务消息 ####

我们知道**事务**是保持一致性的，MQ事务，很自然的就想到了是保证生产者发送成功，消费者消费成功吗？其实不然

**RocketMQ事务消息是保持你的业务代码和发消息的一致性**

当我们发送事务消息到了broker（注意这时候消费者是看不到这条消息的），这时候broker会去请求executeLocalTransaction方法执行我们本地的事务代码

*  返回commit，消费者就可以正常消费了
 *  返回rollback，消息就被删除了，不会被消费
 *  返回unknown，消息不会被消费，也不会被删除

如果你返回了unknown或者null，这时候broker就回去访问你的checkLocalTransaction方法，进行检查消息你本地的事务是成功了还是失败了，我们的checkLocalTransaction方法也是返回上面三个状态，操作也是一样的。

**发送事务消息**

Message<String> message = MessageBuilder.withPayload("message").build();
    TransactionSendResult transactionSendResult = ProducerUtils.rocketMQTemplate.sendMessageInTransaction("topic", message, 123);

**事务监听器**

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
    import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
    import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
    import org.springframework.messaging.Message;
    
    @RocketMQTransactionListener
    public class Tran implements RocketMQLocalTransactionListener { 
        public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) { 
    
            // 1、执行你的事务
            System.out.println(message);
            System.out.println(o);
            // 2、返回状态码给MQ
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    
        public RocketMQLocalTransactionState checkLocalTransaction(Message message) { 
            // 检查本地事务执行的结果
            System.out.println(message);
            return RocketMQLocalTransactionState.COMMIT;
        }
    }

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

## 三、源码 ##

关注微信公众号回复：**rocketmq2.0**

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

[Linux_RocketMQ_ESC_CentOS7]: https://www.xdx97.com/article/684051789376389120
[SpringBoot_RocketMQ_rocketmq-spring-boot-starter]: https://www.xdx97.com/article/684525983356682240
[RocketMQ]: https://github.com/apache/rocketmq/tree/master/docs/cn
[Rocketmq-spring-boot-starter]: https://github.com/apache/rocketmq-spring/wiki/%E7%94%A8%E6%88%B7%E6%89%8B%E5%86%8C
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1RvbXdpbGRib2Fy_size_16_color_FFFFFF_t_70]: /images/20221024/e17c82dd8bf743e696945f47ef69517c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1RvbXdpbGRib2Fy_size_16_color_FFFFFF_t_70_pic_center]: /images/20221024/654af8f2de8c4781aa8e652ecde49e9a.png