Alibaba阿里 RocketMQ入门实例(四种消息类型)

系统管理员 2022-02-25 14:30 733阅读 0赞

## 发送顺序消息 ##

public class ProducerClient {
        public static void main(String[] args) {
            Properties properties = new Properties();
            // 您在控制台创建的 Group ID
            properties.put(PropertyKeyConst.GROUP_ID, "XXX");
            // AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
            properties.put(PropertyKeyConst.AccessKey, "XXX");
            // SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
            properties.put(PropertyKeyConst.SecretKey, "XXX");
            // 设置 TCP 接入域名，进入控制台的实例管理页面的“获取接入点信息”区域查看
            properties.put(PropertyKeyConst.NAMESRV_ADDR,
              "XXX");
            OrderProducer producer = ONSFactory.createOrderProducer(properties);
            // 在发送消息前，必须调用 start 方法来启动 Producer，只需调用一次即可。
            producer.start();
            for (int i = 0; i < 1000; i++) {
                String orderId = "biz_" + i % 10;
                Message msg = new Message(//
                        // Message 所属的 Topic
                        "Order_global_topic",
                        // Message Tag, 可理解为 Gmail 中的标签，对消息进行再归类，方便 Consumer 指定过滤条件在消息队列 RocketMQ 的服务器过滤
                        "TagA",
                        // Message Body 可以是任何二进制形式的数据， 消息队列 RocketMQ 不做任何干预，需要 Producer 与 Consumer 协商好一致的序列化和反序列化方式
                        "send order global msg".getBytes()
                );
                // 设置代表消息的业务关键属性，请尽可能全局唯一。
                // 以方便您在无法正常收到消息情况下，可通过控制台查询消息并补发。
                // 注意：不设置也不会影响消息正常收发
                msg.setKey(orderId);
                // 分区顺序消息中区分不同分区的关键字段，sharding key 于普通消息的 key 是完全不同的概念。
                // 全局顺序消息，该字段可以设置为任意非空字符串。
                String shardingKey = String.valueOf(orderId);
                try {
                    SendResult sendResult = producer.send(msg, shardingKey);
                    // 发送消息，只要不抛异常就是成功
                    if (sendResult != null) {
                        System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId());
                    }
                }
                catch (Exception e) {
                    // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理
                    System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
                    e.printStackTrace();
                }
            }
            // 在应用退出前，销毁 Producer 对象
            // 注意：如果不销毁也没有问题
            producer.shutdown();
        }

## 订阅顺序消息 ##

public static void main(String[] args) {
            Properties properties = new Properties();
            // 您在控制台创建的 Group ID
            properties.put(PropertyKeyConst.GROUP_ID, "XXX");
            // AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
            properties.put(PropertyKeyConst.AccessKey, "XXX");
            // SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
            properties.put(PropertyKeyConst.SecretKey, "XXX");
            // 设置 TCP 接入域名，进入控制台的实例管理页面的“获取接入点信息”区域查看
            properties.put(PropertyKeyConst.NAMESRV_ADDR,
              "XXX");
              // 顺序消息消费失败进行重试前的等待时间，单位(毫秒)，取值范围: 10 毫秒 ~ 1800 毫秒
            properties.put(PropertyKeyConst.SuspendTimeMillis, "100"); 
            // 消息消费失败时的最大重试次数
            properties.put(PropertyKeyConst.MaxReconsumeTimes, "20"); 
            // 在订阅消息前，必须调用 start 方法来启动 Consumer，只需调用一次即可。
            OrderConsumer consumer = ONSFactory.createOrderedConsumer(properties);
            consumer.subscribe(
                    // Message 所属的 Topic
                    "Jodie_Order_Topic",
                    // 订阅指定 Topic 下的 Tags：
                    // 1. * 表示订阅所有消息
                    // 2. TagA || TagB || TagC 表示订阅 TagA 或 TagB 或 TagC 的消息
                    "*",
                    new MessageOrderListener() {
                        /**
                         * 1. 消息消费处理失败或者处理出现异常，返回 OrderAction.Suspend<br>
                         * 2. 消息处理成功，返回 OrderAction.Success
                         */
                        @Override
                        public OrderAction consume(Message message, ConsumeOrderContext context) {
                            System.out.println(message);
                            return OrderAction.Success;
                        }
                    });
            consumer.start();
        }
    }

###  收发事务消息 ###

## 交互流程 ##

消息队列 RocketMQ 的事务消息交互流程如下图所示：

[![事务消息交互流程][E4_BA_8B_E5_8A_A1_E6_B6_88_E6_81_AF_E5_9B_BE_E7_89_87.png]][E4_BA_8B_E5_8A_A1_E6_B6_88_E6_81_AF_E5_9B_BE_E7_89_87.png]

## 发送事务消息 ##

发送事务消息包含以下两个步骤：

1.  发送半消息（Half Message）及执行本地事务。 示例代码如下：
    
    1.      public class TransactionProducerClient {
            private final static Logger log = ClientLogger.getLog(); // 您需要设置自己的日志，便于排查问题
            
            public static void main(String[] args) throws InterruptedException {
            final BusinessService businessService = new BusinessService(); // 本地业务
            Properties properties = new Properties();
            // 您在控制台创建的 Group ID 注意：事务消息的 Group ID 不能与其他类型消息的 Group ID 共用
            properties.put(PropertyKeyConst.GROUP_ID, "XXX");
            // 阿里云身份验证，在阿里云服务器管理控制台创建
            properties.put(PropertyKeyConst.AccessKey, "XXX");
            // 阿里云身份验证，在阿里云服务器管理控制台创建
            properties.put(PropertyKeyConst.SecretKey, "XXX");
            // 设置 TCP 接入域名，进入控制台的实例管理页面的“获取接入点信息”区域查看
            properties.put(PropertyKeyConst.NAMESRV_ADDR,
            "XXX");
            
            TransactionProducer producer = ONSFactory.createTransactionProducer(properties,
            new LocalTransactionCheckerImpl());
            producer.start();
            Message msg = new Message("Topic", "TagA", "Hello MQ transaction===".getBytes());
            try {
            SendResult sendResult = producer.send(msg, new LocalTransactionExecuter() {
            @Override
            public TransactionStatus execute(Message msg, Object arg) {
            // 消息 ID（有可能消息体一样，但消息 ID 不一样，当前消息 ID 在控制台无法查询）
            String msgId = msg.getMsgID();
            // 消息体内容进行 crc32，也可以使用其它的如 MD5
            long crc32Id = HashUtil.crc32Code(msg.getBody());
            // 消息 ID 和 crc32id 主要是用来防止消息重复
            // 如果业务本身是幂等的，可以忽略，否则需要利用 msgId 或 crc32Id 来做幂等
            // 如果要求消息绝对不重复，推荐做法是对消息体 body 使用 crc32 或 MD5 来防止重复消息
            Object businessServiceArgs = new Object();
            TransactionStatus transactionStatus = TransactionStatus.Unknow;
            try {
            boolean isCommit =
            businessService.execbusinessService(businessServiceArgs);
            if (isCommit) {
            // 本地事务成功则提交消息
            transactionStatus = TransactionStatus.CommitTransaction;
            } else {
            // 本地事务失败则回滚消息
            transactionStatus = TransactionStatus.RollbackTransaction;
            }
            } catch (Exception e) {
            log.error("Message Id:{}", msgId, e);
            }
            System.out.println(msg.getMsgID());
            log.warn("Message Id:{}transactionStatus:{}", msgId, transactionStatus.name());
            return transactionStatus;
            }
            }, null);
            }
            catch (Exception e) {
            // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理
            System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
            e.printStackTrace();
            }
            // demo example 防止进程退出(实际使用不需要这样)
            TimeUnit.MILLISECONDS.sleep(Integer.MAX_VALUE);
            }
            }
        
         
2.  提交事务消息状态
    
    当本地事务执行完成（执行成功或执行失败），需要通知服务器当前消息的事务状态。 通知方式有以下两种：
    
     *  执行本地事务完成后提交。
     *  执行本地事务一直没提交状态，等待服务器回查消息的事务状态。
    
    事务状态有以下三种：
    
     *  TransactionStatus.CommitTransaction 提交事务，允许订阅方消费该消息。
     *  TransactionStatus.RollbackTransaction 回滚事务，消息将被丢弃不允许消费。
     *  TransactionStatus.Unknow 无法判断状态，期待消息队列 RocketMQ 的 Broker 向发送方再次询问该消息对应的本地事务的状态。

[试用][Link 1]

public class LocalTransactionCheckerImpl implements LocalTransactionChecker {
    private final static Logger log = ClientLogger.getLog();
    final BusinessService businessService = new BusinessService();
    
    @Override
    public TransactionStatus check(Message msg) {
    //消息 ID（有可能消息体一样，但消息 ID 不一样，当前消息属于半消息，所以消息 ID 在控制台无法查询）
    String msgId = msg.getMsgID();
    //消息体内容进行 crc32，也可以使用其它的方法如 MD5
    long crc32Id = HashUtil.crc32Code(msg.getBody());
    //消息 ID、消息本 crc32Id 主要是用来防止消息重复
    //如果业务本身是幂等的，可以忽略，否则需要利用 msgId 或 crc32Id 来做幂等
    //如果要求消息绝对不重复，推荐做法是对消息体使用 crc32 或 MD5 来防止重复消息
    //业务自己的参数对象，这里只是一个示例，需要您根据实际情况来处理
    Object businessServiceArgs = new Object();
    TransactionStatus transactionStatus = TransactionStatus.Unknow;
    try {
    boolean isCommit = businessService.checkbusinessService(businessServiceArgs);
    if (isCommit) {
    //本地事务已成功则提交消息
    transactionStatus = TransactionStatus.CommitTransaction;
    } else {
    //本地事务已失败则回滚消息
    transactionStatus = TransactionStatus.RollbackTransaction;
    }
    } catch (Exception e) {
    log.error("Message Id:{}", msgId, e);
    }
    log.warn("Message Id:{}transactionStatus:{}", msgId, transactionStatus.name());
    return transactionStatus;
    }
    }

**工具类**

import java.util.zip.CRC32;
    public class HashUtil {
    public static long crc32Code(byte[] bytes) {
    CRC32 crc32 = new CRC32();
    crc32.update(bytes);
    return crc32.getValue();
    }
    }

**事务回查机制说明**

*  发送事务消息为什么必须要实现回查 Check 机制？
    
    当步骤（1）中半消息发送完成，但本地事务返回状态为 TransactionStatus.Unknow，或者应用退出导致本地事务未提交任何状态时，从 Broker 的角度看，这条 Half 状态的消息的状态是未知的。 因此 Broker 会定期要求发送方能 Check 该半状态消息，并上报其最终状态。
 *  Check 被回调时，业务逻辑都需要做些什么？
    
    事务消息的 Check 方法里面，应该写一些检查事务一致性的逻辑。 消息队列 RocketMQ 发送事务消息时需要实现 LocalTransactionChecker 接口，用来处理 Broker 主动发起的本地事务状态回查请求；因此在事务消息的 Check 方法中，需要完成两件事情：
    
    (1) 检查该半消息对应的本地事务的状态（committed or rollback）。
    
    (2) 向 Broker 提交该半消息本地事务的状态。

# 收发延时消息 #

延时消息用于指定消息发送到消息队列 RocketMQ 的服务器端后，延时一段时间才被投递到客户端进行消费（例如 3 秒后才被消费），适用于解决一些消息生产和消费有时间窗口要求的场景，或者通过消息触发延迟任务的场景，类似于延迟队列。

关于延时消息的概念介绍及使用过程中的注意事项，请参见[延时消息][Link 2]。

发送延时消息
    发送延时消息的示例代码如下：
    
    试用
    import com.aliyun.openservices.ons.api.Message;
    import com.aliyun.openservices.ons.api.ONSFactory;
    import com.aliyun.openservices.ons.api.Producer;
    import com.aliyun.openservices.ons.api.PropertyKeyConst;
    import com.aliyun.openservices.ons.api.SendResult;
    import java.util.Properties;
    public class ProducerDelayTest {
        public static void main(String[] args) {
            Properties properties = new Properties();
            // AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
            properties.put(PropertyKeyConst.AccessKey, "XXX");
            // SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
            properties.put(PropertyKeyConst.SecretKey, "XXX");
            // 设置 TCP 接入域名，进入控制台的实例管理页面的“获取接入点信息”区域查看
            properties.put(PropertyKeyConst.NAMESRV_ADDR,
              "XXX");
            Producer producer = ONSFactory.createProducer(properties);
            // 在发送消息前，必须调用 start 方法来启动 Producer，只需调用一次即可。
            producer.start();
            Message msg = new Message( //
                    // 您在控制台创建的 Topic
                    "Topic",
                    // Message Tag, 可理解为 Gmail 中的标签，对消息进行再归类，方便 Consumer 指定过滤条件在消息队列 RocketMQ 服务器过滤
                    "tag",
                    // Message Body 可以是任何二进制形式的数据，消息队列 RocketMQ 不做任何干预，需要 Producer 与 Consumer 协商好一致的序列化和反序列化方式
                    "Hello MQ".getBytes());
            // 设置代表消息的业务关键属性，请尽可能全局唯一。
            // 以方便您在无法正常收到消息情况下，可通过控制台查询消息并补发。
            // 注意：不设置也不会影响消息正常收发
            msg.setKey("ORDERID_100");
            try {
                // 延时消息，单位毫秒（ms），在指定延迟时间（当前时间之后）进行投递，例如消息在 3 秒后投递
                long delayTime = System.currentTimeMillis() + 3000;
                // 设置消息需要被投递的时间
                msg.setStartDeliverTime(delayTime);
                SendResult sendResult = producer.send(msg);
                // 同步发送消息，只要不抛异常就是成功
                if (sendResult != null) {
                System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId());
                }
                } catch (Exception e) {
                // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理
                System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
                e.printStackTrace();
            }
            // 在应用退出前，销毁 Producer 对象<br>
            // 注意：如果不销毁也没有问题
            producer.shutdown();
        }
    }

###  收发定时消息 ###

定时消息可以做到在指定时间戳之后才可被消费者消费，适用于对消息生产和消费有时间窗口要求，或者利用消息出发定时任务的场景。

关于定时消息的概念介绍及使用过程中的注意事项

public class ProducerDelayTest {
        public static void main(String[] args) {
            Properties properties = new Properties();
            // 阿里云身份验证，在阿里云服务器管理控制台创建
            properties.put(PropertyKeyConst.AccessKey, "XXX");
            // 阿里云身份验证，在阿里云服务器管理控制台创建
            properties.put(PropertyKeyConst.SecretKey, "XXX");
            // 设置 TCP 接入域名，进入控制台的实例管理页面的“获取接入点信息”区域查看
            properties.put(PropertyKeyConst.NAMESRV_ADDR,
              "XXX");
            Producer producer = ONSFactory.createProducer(properties);
            // 在发送消息前，必须调用 start 方法来启动 Producer，只需调用一次即可。
            producer.start();
            Message msg = new Message( //
                    // Message 所属的 Topic
                    "Topic",
                    // Message Tag 可理解为 Gmail 中的标签，对消息进行再归类，方便 Consumer 指定过滤条件在消息队列 RocketMQ 的服务器过滤
                    "tag",
                    // Message Body 可以是任何二进制形式的数据， 消息队列 RocketMQ 不做任何干预，需要 Producer 与 Consumer 协商好一致的序列化和反序列化方式
                    "Hello MQ".getBytes());
            // 设置代表消息的业务关键属性，请尽可能全局唯一
            // 以方便您在无法正常收到消息情况下，可通过控制台查询消息并补发。
            // 注意：不设置也不会影响消息正常收发
            msg.setKey("ORDERID_100");
            try {
                // 定时消息，单位毫秒（ms），在指定时间戳（当前时间之后）进行投递，例如 2016-03-07 16:21:00 投递。如果被设置成当前时间戳之前的某个时刻，消息将立刻投递给消费者。
                long timeStamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse("2016-03-07 16:21:00").getTime();
                msg.setStartDeliverTime(timeStamp);
                // 发送消息，只要不抛异常就是成功
                SendResult sendResult = producer.send(msg);
                System.out.println("Message Id:" + sendResult.getMessageId());
            }
            catch (Exception e) {
                // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理
                System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
                e.printStackTrace();
            }
            // 在应用退出前，销毁 Producer 对象
            // 注意：如果不销毁也没有问题
            producer.shutdown();
        }

[E4_BA_8B_E5_8A_A1_E6_B6_88_E6_81_AF_E5_9B_BE_E7_89_87.png]: /images/20220225/13e31c2d9f1f4f178b14faab56c4abb4.png
[Link 1]: https://api.aliyun.com/new#/tutorial?id=29548
[Link 2]: https://help.aliyun.com/document_detail/43349.html