在maven中使用RocketMQ发布消息和消费消息

一. 创建maven项目并导入依赖

PS: 需要自行安装RocketMQ并启动服务

<dependencies>
     <dependency>
         <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-client</artifactId>
         <version>4.9.4</version>
     </dependency>
 </dependencies>

二. 同步消息

同步发送是指发送的消息需要等待获取到返回结果后再执行第二条消息发送,采用的是一种阻塞模型，安全，但效率会降低一丢丢。

1. 生产者

   package com.lqs.demo01;
   
   import org.apache.rocketmq.client.producer.DefaultMQProducer;
   import org.apache.rocketmq.client.producer.SendResult;
   import org.apache.rocketmq.common.message.Message;
   
   import java.nio.charset.StandardCharsets;
   
   /**
    * @author lqs
    * @date 2022/11/3 9:07
    */
   public class Producer { // 生产者
       public static void main(String[] args) throws Exception {
           // 创建生产者 
           DefaultMQProducer producer = new DefaultMQProducer("producer-demo");
           // 连接服务
           producer.setNamesrvAddr("127.0.0.1:9876");
           producer.start();
   		// 消息参数依次是主题、标签、发送的消息
           Message message = new Message("demo-topic", "demo", "aaa I am a monster! nice to see ya!!!".getBytes(StandardCharsets.UTF_8));
           // 返回结果
           SendResult result = producer.send(message);
           System.out.println(result);
   		// 关闭生产者
           producer.shutdown();
       }
   }

2. 消费者

   package com.lqs.demo01;
   
   import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
   import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
   import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
   import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
   import org.apache.rocketmq.client.exception.MQClientException;
   import org.apache.rocketmq.common.message.MessageExt;
   
   import java.util.Arrays;
   import java.util.List;
   
   /**
    * @author lqs
    * @date 2022/11/3 9:25
    */
   public class Consumer {// 消费者
   
       public static void main(String[] args) throws Exception {
           // 创建消费者
           DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("demo-consumer");
           consumer.setNamesrvAddr("127.0.0.1:9876");
           // 订阅需要消费的主题和标签下的消息
           consumer.subscribe("demo-topic", "demo");
           // 实现消息监听接口
           consumer.registerMessageListener(new MessageListenerConcurrently() {
               @Override
               public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                   try {
                       list.forEach(l -> {
                           // 获取消息
                           System.out.println("body:" +new String(l.getBody()));
                           System.out.println("context:" + consumeConcurrentlyContext);
                       });
                       return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                   } catch (Exception e) {
                       return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                   }
               }
           });
           consumer.start();
           System.out.println("消息消息启动成功");
       }
   }

三. 异步消息

异步消息是指发送的消息到MQ队列不用等待其返回结果，继续发送其它消息，效率快。

1. 生产者

   package com.lqs.demo02;
   
   import org.apache.rocketmq.client.producer.DefaultMQProducer;
   import org.apache.rocketmq.client.producer.SendCallback;
   import org.apache.rocketmq.client.producer.SendResult;
   import org.apache.rocketmq.common.message.Message;
   
   import java.nio.charset.StandardCharsets;
   
   /**
    * @author lqs
    * @date 2022/11/3 9:38
    */
   public class Producer {// 异步生产者发送消息
   
       public static void main(String[] args) throws Exception {
           DefaultMQProducer producer = new DefaultMQProducer("producer-demo");
           producer.setNamesrvAddr("127.0.0.1:9876");
           producer.start();
   
           Message message = new Message("demo-topic", "demo-async", "this is a async message and lqs is freaking awesome!".getBytes(StandardCharsets.UTF_8));
           // 设置延时(过了一定时间再发送到MQ对列) 5秒
         //  message.setDelayTimeLevel(2);// 指定延迟等级 比如2 表示5秒
           
           // 异步回调获取返回结果
           producer.send(message, new SendCallback() {
               @Override
               public void onSuccess(SendResult sendResult) {
                   System.out.println("发送结果:"+sendResult);
               }
   
               @Override
               public void onException(Throwable throwable) {
                   System.out.println("失败原因:"+throwable.getMessage());
               }
           });
       }
   }

2. 消费者

   package com.lqs.demo02;
   
   import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
   import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
   import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
   import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
   import org.apache.rocketmq.client.exception.MQClientException;
   import org.apache.rocketmq.common.message.MessageExt;
   
   import java.util.List;
   
   /**
    * @author lqs
    * @date 2022/11/3 9:47
    */
   public class Consumer {
       public static void main(String[] args) throws Exception {
           DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("demo-consumer");
           consumer.setNamesrvAddr("127.0.0.1:9876");
           consumer.subscribe("demo-topic", "demo-async");
   
           // 实现消息监听接口
           consumer.registerMessageListener(new MessageListenerConcurrently() {
               @Override
               public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                   try {
                       list.forEach(l -> {
                           // 获取消息
                           System.out.println("body:" +new String(l.getBody()));
                           System.out.println("context:" + consumeConcurrentlyContext);
                       });
                       // 返回状态表示消息已经被消费，不可再次消费
                       return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                   } catch (Exception e) {
                       // 消费失败 可再次消费
                       return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                   }
               }
           });
           consumer.start();
           System.out.println("消息消息启动成功");
       }
   }

四. 事务消息

1. 事务监听

   package com.lqs.demo4;
   
   import org.apache.rocketmq.client.producer.LocalTransactionState;
   import org.apache.rocketmq.client.producer.TransactionListener;
   import org.apache.rocketmq.common.message.Message;
   import org.apache.rocketmq.common.message.MessageExt;
   
   /**
    * @author lqs
    * @date 2022/11/3 10:12
    */
   public class MyTransactionCheckListener implements TransactionListener {
       /**
        * @param message
        * @param o       事物消息携带的额外参数
        * @return
        */
       @Override
       public LocalTransactionState executeLocalTransaction(Message message, Object o) {
           try {
               System.out.println("事务发布消息:" + message);
               System.out.println("事务参数（额外数据）:" + o);
               // 事务返回状态
               return LocalTransactionState.COMMIT_MESSAGE;
           } catch (Exception e) {
               e.printStackTrace();
               return LocalTransactionState.ROLLBACK_MESSAGE;
           }
       }
   
       @Override
       public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
           return null;
       }
   }

2. 生成者

   package com.lqs.demo4;
   
   import org.apache.rocketmq.client.exception.MQClientException;
   import org.apache.rocketmq.client.producer.DefaultMQProducer;
   import org.apache.rocketmq.client.producer.TransactionMQProducer;
   import org.apache.rocketmq.common.message.Message;
   
   import java.nio.charset.StandardCharsets;
   import java.util.concurrent.ExecutorService;
   import java.util.concurrent.Executors;
   
   /**
    * @author lqs
    * @date 2022/11/3 10:13
    */
   public class Producer {// 事务消息生产者
   
       public static void main(String[] args) throws Exception {
           //1.创建生产者对象
           TransactionMQProducer producer = new TransactionMQProducer("trans-demo");
           //2.连接NameServer
           producer.setNamesrvAddr("127.0.0.1:9876");
           //3.创建线程
           ExecutorService executorService = Executors.newFixedThreadPool(2);
           producer.setExecutorService(executorService);
           //4.指定线程的事物监听
           producer.setTransactionListener(new MyTransactionCheckListener());
           //3.启动
           producer.start();
   
           Message message = new Message("demo-topic", "demo-trans", "this is a trans message and lqs is freaking awesome!".getBytes(StandardCharsets.UTF_8));
           producer.sendMessageInTransaction(message, "事务demo参数");
   
       }
   }

3. 消费者

   package com.lqs.demo4;
   
   import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
   import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
   import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
   import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
   import org.apache.rocketmq.common.message.MessageExt;
   
   import java.util.List;
   
   /**
    * @author lqs
    * @date 2022/11/3 9:47
    */
   public class Consumer {
       public static void main(String[] args) throws Exception {
           DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("demo-trans");
           consumer.setNamesrvAddr("127.0.0.1:9876");
           consumer.subscribe("demo-topic", "demo-trans");
   
           // 实现消息监听接口
           consumer.registerMessageListener(new MessageListenerConcurrently() {
               @Override
               public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                   try {
                       list.forEach(l -> {
                           // 获取消息
                           System.out.println("body:" +new String(l.getBody()));
                           System.out.println("context:" + consumeConcurrentlyContext);
                       });
                       return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                   } catch (Exception e) {
                       return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                   }
               }
           });
           consumer.start();
           System.out.println("消息消息启动成功");
       }
   }

五. RocketMQ事务消息工作流程

• 事务消息解决的就是，事务发送方执行本地事务和事务消息的原子性。

1. 首先事务发起方:往MQ中发送一个事务消息-half半消息(准备消息)，该消息不可被消费;

2. 然后事务发送方执行本地事务，并发送commit给MQ，如果是执行失败就会发送rollback给MQ;

3. MQ搜到commit指令，代表本地事务执行成功，之前的half消息就会成为可以被commit提交的消

   息，否则就是Rollback消息;

4. 事务参与者，也就是消费者会消费这个消息，然后执行相关的逻辑处理。如果是Rollback消息就不

   会被消费，而是丢弃;

5. 如果事务参与方并没有发送commit或者rollback指令MQ，MQ回调用事务发送方的回到方法来检

   查，我们需要去实现这个本地事务检查接口，通过返回commit或者rollback来告知MQ本地事务是

   否执行成功。