RocketMQ快速入门：集成java客户端实现各类消息发送｜异步、同步、顺序、单向、延迟、事务（五）附带源码

目录

• 0. 引言
• 1. 集成rocketMQ
• • 1.1 消息同步发送
  • • 1.1.1 实现步骤
    • 1.1.1 消息重试设置
    • 1.2 消息异步发送
    • 1.3 消息单向发送
    • 1.4 消息顺序发送
    • 1.5 消息批量发送
    • 1.6 消息延迟发送
    • 1.7 事务消息发送
    • 3. 总结

      0. 引言

      前面的章节中，我们已经针对rocketmq的基本概念和消息发送、消费流程进行了讲解，但实际在开发中如何实现rocketmq的接入、实现消息发送、消费还没有落实，那么今天，我们继续来学习如何基于java client集成rocketMQ

      1. 集成rocketMQ

      1、spring或其他java框架集成rocektmq很简单，只需要引入rocketmq依赖即可， 这里的版本号与你的rocketmq保持一致

       org.apache.rocketmq rocketmq-client 4.8.0

      1.1 消息同步发送

      官方文档：https://rocketmq.apache.org/zh/docs/4.x/producer/02message1/

      1.1.1 实现步骤

      1、实现消息发送需要利用生产者类DefaultMQProducer实现，以下示例简单消息的同步发送

      public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException { // 声明group
              DefaultMQProducer producer = new DefaultMQProducer("group_test");
              // 声明namesrv地址
              producer.setNamesrvAddr("localhost:9876");
              // 启动实例
              producer.start();
              // 设置消息的topic,tag以及消息体
              Message msg = new Message("topic_test", "tag_test", "消息内容".getBytes(StandardCharsets.UTF_8));
              // 发送消息，并设置10s连接超时
              SendResult send = producer.send(msg, 10000);
              System.out.println("发送结果："+send);
              // 关闭实例
              producer.shutdown();
          }
      

      2、当然，实际生产中，最好不要每次发送消息都创建新的生产者，于是我们需要将其封装成一个bean，实现起来也很简单

      @Configuration
      public class RocketMQConfig { // 配置项大家自定义即可
          @Value("${rocketmq.name-server}")
          private String nameSrvAddr;
          @Value("${rocketmq.producer.group}")
          private String producerGroup;
          private DefaultMQProducer producer;
          @Bean
          public DefaultMQProducer mqProducer(){ producer = new DefaultMQProducer(producerGroup);
              producer.setNamesrvAddr(nameSrvAddr);
              try { producer.start();
              } catch (MQClientException e) { e.printStackTrace();
              }
              return producer;
          }
      }
      

      3、调用：

       @Resource
          private DefaultMQProducer mqProducer;
          @GetMapping("send2")
          public String send2(){ org.apache.rocketmq.common.message.Message msg = new org.apache.rocketmq.common.message.Message("topic_test", "tag_test", "消息内容".getBytes(StandardCharsets.UTF_8));
              try { mqProducer.send(msg, 1000);
              } catch (Exception e) { e.printStackTrace();
                  return "fail";
              }
              return "success";
          }
      

      4、因为没有实现DefaultMQProducer的自动关闭，如果要进一步优化，大家可以利用InitializingBean, DisposableBean两个接口实现afterPropertiesSet和destroy，实现资源的自动启动、自动关闭，比如通过destroy实现producer的自动关闭：

       @Override
          public void destroy() { if (Objects.nonNull(this.producer)) { this.producer.shutdown();
              }
          }
      

      5、发送结果如下所示

      SendResult [sendStatus=SEND_OK, msgId=7F00000135DA18B4AAC20E15FE090000, offsetMsgId=C0A8F40100002A9F000000000000C307, messageQueue=MessageQueue [topic=topic_test, brokerName=broker, queueId=15], queueOffset=7]

      1.1.1 消息重试设置

      实际生产时，可能因为网络波动等不确定原因，导致服务连接出现短时问题，如果恰巧这时我们发送了消息，就会导致发送失败。所以一般我们还会设置消息的重试，只需要调用setRetryTimesWhenSendFailed方法即可

      // 设置发送失败时重试次数
      producer.setRetryTimesWhenSendFailed(2);
      

      1.2 消息异步发送

      消息同步发送可以保证顺序性，但是部分场景我们无需考虑发送次序，并且同步发送如果前面的发送阻塞了，反而会影响后续的发送效率。于是异步发送就产生了。

      1、通过查看DefaultMQProducer类可以看到，发送方法中有一个SendCallback参数，这实际就是异步发送的回调方法参数，通过他我们就可以实现异步发送了

      2、实现代码

      import org.apache.rocketmq.client.exception.MQBrokerException;
      import org.apache.rocketmq.client.exception.MQClientException;
      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 org.apache.rocketmq.remoting.exception.RemotingException;
      import java.nio.charset.StandardCharsets;
      /**
       * @author benjamin_5
       * @Description
       * @date 2024/2/25
       */
      public class Producer2AsyncDemo { public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException { DefaultMQProducer producer = new DefaultMQProducer("group_test");
              producer.setNamesrvAddr("127.0.0.1:9876");
              // 启动实例
              producer.start();
              for (int i = 0; i < 10; i++) { Message msg = new Message("topic_test", "tag_test", ("hello "+ i).getBytes(StandardCharsets.UTF_8));
                  producer.send(msg, new SendCallback() { @Override
                      public void onSuccess(SendResult sendResult) { System.out.println(new String(msg.getBody(), StandardCharsets.UTF_8)+"发送结果："+sendResult);
                      }
                      @Override
                      public void onException(Throwable e) { System.out.println("异常："+e);
                          e.printStackTrace();
                      }
                  });
              }
              // 等待发送完再结束主线程
              Thread.sleep(10000);
              producer.shutdown();
          }
      }
      

      3、通过观察发送结果，我们也能看出是无序的异步发送

      1.3 消息单向发送

      单向消息主要适用于非核心日志等允许消息丢失的场景，其方法是没有返回值的，也就是拿不到响应结果，当然好处就是：快！

      单向消息支持同步和异步，通过参数可以了解

      示例代码如下：

      public class Producer5OneWayDemo { public static void main(String[] args) throws Exception{ // 声明group
              DefaultMQProducer producer = new DefaultMQProducer("group_test");
              // 声明namesrv地址
              producer.setNamesrvAddr("localhost:9876");
              // 设置重试次数
              producer.setRetryTimesWhenSendFailed(2);
              // 启动实例
              producer.start();
              // 设置消息的topic,tag以及消息体
              Message msg = new Message("topic_test", "tag_test", "单向消息".getBytes(StandardCharsets.UTF_8));
              // 发送消息
              producer.sendOneway(msg);
              // 关闭实例
              producer.shutdown();
          }
      }
      

      1.4 消息顺序发送

      官方文档：https://rocketmq.apache.org/zh/docs/4.x/producer/03message2

      顺序消息在对流程有严格顺序要求的业务场景有广泛应用，比如订单的创建、付款、发货、签收，这几个状态流程必须保持顺序。

      而顺序消息的实现也分成两部分：顺序发送、顺序消费。本章我们主要针对顺序发送进行讲解，顺序消费我们将在下一章继续说明。这里需要注意的是，顺序发送不等于同步发送，虽然同步发送也保持一定的顺序性，但想象如果多个线程发送多个同步消息，那消息还具有顺序性吗？ 这就不一定了。那顺序发送就是单线程的同步发送吗？

      1、单线程的同步发送其实也不能满足顺序性，比如如下的场景，生产者单线程同步发送了两条消息A1,A2, MQ会将其负载到不同的队列存储，然后消费者可能有多个，消费时就容易产生同时从这两个队列消费，从而就不能保证A1一定在A2之前被消费掉。于是单线程的同步发送其实并不能保证顺序性，那么怎么处理呢？

      2、既然会发送到不同的队列导致同时消费的可能性，那我们就确保同一数据发送到同一队列，比如同一订单的创建、付款、签收都发送到一个队列中，如此来保证发送的顺序性

      3、所以，顺序发送的重点，也在于让同类消息发送到同一队列上，rocketmq中是通过队列选择器MessageQueueSelector来实现的

      我们查看send方法，需要提供3个参数：消息、队列选择器和一个arg。前两个参数可以理解，那么这个arg是什么呢，其实就是我们用于区分消息类型的标识，比如为了防止上述的混淆，我们要将同一订单的不同状态的消息都发送到同一个队列中，那么我们就可以以订单号作为这个标识，其目的就是将同一类型的消息通过这个标识进行区分。

      4、在发送消息之间，我们需要创建队列，并且将队列指定为顺序队列，即创建队列时指定–order参数为true

      （1）我们需要通过namesrv内置的mqadmin工具来实现指定

      （2）进入namesrv，在其安装目录的bin目录下执行（如果你和我一样是通过docker安装的rocketmq，那直接进入docker namesrv的容器即可， 进入后的当前目录就是bin目录）

      ./mqadmin updateTopic -c DefaultCluster -t topic_order -o true -n localhost:9876
      

      -n NameServer的地址和端口

      -c 指定集群名称

      -t 指定主题名称

      -w 指定队列的数量，如果要保证全局顺序性，可以设置队列数为1，以此来避免多队列产生的非顺序性问题

      5、其次如果需要保证严格的顺序性，还需要在namesrv中配置orderMessageEnable 和 returnOrderTopicConfigToBroker 是 true

      （注：默认配置文件为namesrv.properties（通过./mqnamesrv -p即可查看配置文件路径），也可通过创建自定义配置文件namesrv.conf， 启动namesrv时指定配置文件nohup sh bin/mqnamesrv -c conf/namesrv.conf &， 本文示例无需配置该项也可保证顺序性，但生产时建议配置）

      orderMessageEnable=true
      returnOrderTopicConfigToBroker=true
      

      6、整体的顺序发送代码如下，这里我简单使用一个orderId作为区分标识（也叫区分键），将奇数和偶数消息分别视为一种消息，大家实际应用时可以根据自己的业务调整。

      其MessageQueueSelector对象的定义，主要是实现其select方法，这里就是通过arg参数，做取余运算，进行队列的选择。当然大家也可以用arg的hashcode来作为标识处理

       public static void main(String[] args) throws Exception{ DefaultMQProducer producer = new DefaultMQProducer("group_test");
              // 声明namesrv地址
              producer.setNamesrvAddr("localhost:9876");
              // 设置重试次数
              producer.setRetryTimesWhenSendFailed(2);
              // 启动实例
              producer.start();
              // 设置消息的topic,tag以及消息体
              Message msg = new Message("topic_test", "tag_test", "消息内容".getBytes(StandardCharsets.UTF_8));
              // 要求发送顺序：i为偶数先发，然后按照由小到大顺序发送
              for (int i = 0; i < 10; i++) { // 模拟偶数、奇数分别属于一类消息
                  int orderId = i % 2;
                  SendResult result = producer.send(msg, new MessageQueueSelector() { /**
                       *
                       * @param list 消息队列集合
                       * @param message 消息
                       * @param arg send方法中传入的第三参数，即orderId参数，orderId可以是Object类型
                       * @return
                       */
                      @Override
                      public MessageQueue select(List list, Message message, Object arg) { Integer orderId = (Integer) arg;
                          int index = orderId % list.size();
                          return list.get(index);
                      }
                  }, orderId);
                  System.out.println("发送结果："+result.toString());
              }
              producer.shutdown();
          }
      

      7、提前用顺序消费的代码来消费验证下

      public class Consumer2OrderDemo { public static void main(String[] args) throws InterruptedException, MQClientException { DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group_test");
              consumer.setNamesrvAddr("127.0.0.1:9876");
              // 集群消费模式
              consumer.setMessageModel(MessageModel.CLUSTERING);
              // 设置topic
              consumer.subscribe("topic_order", "*");
              // 注册回调函数，处理消息
              consumer.registerMessageListener(new MessageListenerOrderly() { @Override
                    public ConsumeOrderlyStatus consumeMessage(List list, ConsumeOrderlyContext consumeOrderlyContext) { byte[] body = list.get(0).getBody();
                        System.out.println("接收消息："+new String(body, StandardCharsets.UTF_8));
                        return ConsumeOrderlyStatus.SUCCESS;
                    }
              });
              // 启动消费者实例
              consumer.start();
              Thread.sleep(10000);
          }
      }
      

      消费到的消息如下所示，可以看到奇偶是分别保持顺序的，即：0,2,4,6,8 和1,3,5,7,9

      最后总结一下，顺序发送的前提要满足两点：

      • 单一生产者：不同的生产者部署在不同的服务器，哪怕使用相同的区分键，也可能导致不同生产者之间的消息无法区分顺序性
      • 串行发送：多线程发送可能会导致并发发送，从而导致破坏消息顺序性

        因此rocketmq的顺序消息，实际是基于队列的顺序消息，不同队列之间的消息是不满足顺序性的。这一点大家要注意。

        1.5 消息批量发送

        官方文档：https://rocketmq.apache.org/zh/docs/4.x/producer/05message4

        批量消息相对比较简单，从发送方法中可以看到支持集合形式的message对象

        因此我们只需要传入List即可

        public class Producer4BatchDemo { public static void main(String[] args) throws Exception { try { DefaultMQProducer producer = new DefaultMQProducer("group_test");
                    // 声明namesrv地址
                    producer.setNamesrvAddr("localhost:9876");
                    // 设置重试次数
                    producer.setRetryTimesWhenSendFailed(2);
                    producer.start();
                    String topic = "topic_test";
                    List messages = new ArrayList<>();
                    messages.add(new Message(topic, "Tag",  "消息1".getBytes()));
                    messages.add(new Message(topic, "Tag",  "消息2".getBytes()));
                    messages.add(new Message(topic, "Tag",  "消息3".getBytes()));
                    SendResult send = producer.send(messages);
                    System.out.println("发送结果："+send);
                    producer.shutdown();
                } catch (Exception e) { e.printStackTrace();
                }
            }
        }
        

        需要注意的是:

        （1）批量消息的大小不能超过 1MiB（否则需要自行分割）

        （2）同一批 batch 中 topic 必须相同

        （3）批量消息是不支持延迟消息的

        1.6 消息延迟发送

        官方文档：https://rocketmq.apache.org/zh/docs/4.x/producer/04message3

        rocketMQ的延迟消息时间不能自定义，但是可以在预设的16个等级中选择，也能够满足我们大部分的业务场景

        延迟消息的发送，只需要在message中指定延迟时间等级即可，通过setDelayTimeLevel方法实现，示例代码如下

        public class Producer6ScheduledDemo { public static void main(String[] args) { try { DefaultMQProducer producer = new DefaultMQProducer("group_test");
                    // 声明namesrv地址
                    producer.setNamesrvAddr("localhost:9876");
                    // 设置重试次数
                    producer.setRetryTimesWhenSendFailed(2);
                    producer.start();
                    int totalMessagesToSend = 100;
                    for (int i = 0; i < totalMessagesToSend; i++) { Message message = new Message("topic_test", ("延迟消息 " + i).getBytes());
                        
                        message.setDelayTimeLevel(3);
                        // Send the message
                        SendResult send = producer.send(message);
                        System.out.println("发送结果："+send);
                    }
                    
                    producer.shutdown();
                } catch (Exception e) { e.printStackTrace();
                }
            }
        }
        

        1.7 事务消息发送

        rocketmq的一大特性就是支持事务消息，他的原理是利用half message来实现的，这个具体原理我们后续单独讲解，今天先关注他的使用

        事务消息的发送不再使用 DefaultMQProducer，而是使用 TransactionMQProducer 进行发送，我们还可以设置事务回查的线程池，当然如果不设置也会默认生成一个

        1、首先需要实现 TransactionListener 接口，并传入 TransactionMQProducer

        事务消息和数据库事务类似，也是通过两阶段提交的形式来实现的，所以其实现方法中要重写executeLocalTransaction和 checkLocalTransaction方法

        executeLocalTransaction方法是用于执行消费消息以及后续的其他事务事件，然后返回一个事物状态，用于告诉rocketmq这条事务消息是可以正常提交了还是需要回滚。

        2、在讲具体实现代码前我们先理清楚几个概念，帮助大家理解为什么要书写这些代码，首先查看状态枚举LocalTransactionState可以知道，executeLocalTransaction方法返回值一共有3种状态：

        • COMMIT_MESSAGE

          提交状态，事务正常进行，一般是本地事务执行成功后进行设置。告知broker提交该事务消息，然后消费者可以消费该消息，当然此时消费者已经执行完本地事务了，再消费可以根据业务逻辑进行后续的逻辑处理，如果没有相关逻辑了忽略消息即可

          • ROLLBACK_MESSAGE

            回滚状态，事务撤回，broker将删除当前half消息，一般是本地事务执行失败后进行设置

            • UNKNOW

              未知状态，固定时间后Broker端会通过checkLocalTransaction方法进行消息回查，根据回查结果来判断该消息是提交还是回滚

              3、要理解checkLocalTransaction方法，我们还得先理解什么是消息回查？

              有一种场景：当本地事务执行完成后，在返回COMMIT_MESSAGE或ROLLBACK_MESSAGE状态时，因为网络波动或者broker服务挂了，导致broker没有正常收到这个状态，从而没有及时把half message进行提交或回滚，这时就需要有个定时巡查机制，来检查这些没有正常收到提交状态的消息的实际状态到底是什么，这个巡查机制就是消息回查。因此checkLocalTransaction方法中就要书写检查本地事务状态的方法，比如事务是对订单提交消息的消费，那么就去查询订单状态，如果已经是提交状态那么就返回COMMIT_MESSAGE，否则就返回ROLLBACK_MESSAGE

              最后总结一下，executeLocalTransaction方法用于书写消费消息，执行本地事务的代码，并最终返回一个状态，本地事务执行成功则返回COMMIT_MESSAGE状态，执行失败则返回ROLLBACK_MESSAGE，UNKNOW状态实际书写时是基本用不到的。checkLocalTransaction状态用于消息状态回查，需要在该方法中提供一个查询本地事务执行状态的方法，然后返回实际执行状态，本地事务执行成功返回COMMIT，执行失败返回ROLLBACK

              4、实现代码，申明TransactionListener接口，实现executeLocalTransaction和checkLocalTransaction方法，生产中可以将TransactionListenerImpl申明为bean, 然后引入其他dao类或者service类处理本地事务

              class TransactionListenerImpl implements TransactionListener { @Override
                      public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { // TODO 执行本地事务（书写你自己的本地事务逻辑）
                          try{ String body = new String(msg.getBody());
                              int i = Integer.parseInt(body);
                              // 模拟偶数执行成功，奇数执行失败
                              if(i % 2 == 0){ System.out.println("本地事务执行成功："+body);
                                  // 执行成功
                                  return LocalTransactionState.COMMIT_MESSAGE;
                              }else{ System.out.println("本地事务执行失败："+body);
                                  // 执行失败
                                  return LocalTransactionState.ROLLBACK_MESSAGE;
                              }
                          }catch (Exception e){ e.printStackTrace();
                              // 执行失败
                              return LocalTransactionState.ROLLBACK_MESSAGE;
                          }
                      }
                      @Override
                      public LocalTransactionState checkLocalTransaction(MessageExt msg) { // TODO 去缓存或者数据库查询当前消息的实际状态
                          // 模拟查询到状态为1
                          Integer status = 1;
                          // 不同实际状态对应的消息状态
                          if (null != status) { switch (status) { case 1:
                                      return LocalTransactionState.COMMIT_MESSAGE;
                                  case 2:
                                      return LocalTransactionState.ROLLBACK_MESSAGE;
                                  default:
                                      return LocalTransactionState.COMMIT_MESSAGE;
                              }
                          }
                          return LocalTransactionState.COMMIT_MESSAGE;
                      }
                  }
              

              6、消息发送方法，事务消息不能再用DefaultMQProducer了，得用TransactionMQProducer

              这里额外申明了回查线程池executorService，当然如果不申明rocektmq也会默认创建一个的

              public static void main(String[] args) throws MQClientException, InterruptedException { TransactionListener transactionListener = new TransactionListenerImpl();
                      TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name");
                      ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue(2000), new ThreadFactory() { @Override
                          public Thread newThread(Runnable r) { Thread thread = new Thread(r);
                              thread.setName("transaction-msg-check-thread");
                              return thread;
                          }
                      });
                      producer.setExecutorService(executorService);
                      producer.setTransactionListener(transactionListener);
                      producer.setNamesrvAddr("localhost:9876");
                      producer.start();
                      String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
                      for (int i = 0; i < 10; i++) { try { Message msg =
                                      new Message("topic_test", tags[i % tags.length], "KEY" + i,
                                              (""+i).getBytes());
                              SendResult sendResult = producer.sendMessageInTransaction(msg, null);
                              System.out.printf("%s%n", sendResult);
                              Thread.sleep(10);
                          } catch (MQClientException e) { e.printStackTrace();
                          }
                      }
                      for (int i = 0; i < 100000; i++) { Thread.sleep(1000);
                      }
                      producer.shutdown();
                  }
              

              7、执行消息发送结果

              8、消费消息，可以看到只有偶数消息正常提交消费了，奇数都被回滚了

              3. 总结

              至此我们针对java client实现各类消息发送的方法就梳理完成了，但实际工作中，我们现在更加常用的是基于springboot框架，而rocektmq也有专门针对springboot框架进行集成，实现起来更加简单，下一期我们重点讲解springboot集成实现消息发送

              本文演示源码见：https://gitee.com/wuhanxue/wu_study/tree/master/demo/rocketmq_demo