收发顺序消息

更新时间: 2023-07-25 14:18:53

顺序消息(FIFO消息)是云消息队列 RocketMQ 版提供的一种严格按照顺序来发布和消费的消息类型。本文提供使用HTTP协议下的Java SDK收发顺序消息的示例代码。

背景信息

顺序消息分为两类:

  • 全局顺序:对于指定的一个Topic,所有消息按照严格的先入先出FIFO(First In First Out)的顺序进行发布和消费。

  • 分区顺序:对于指定的一个Topic,所有消息根据Sharding Key进行区块分区。同一个分区内的消息按照严格的FIFO顺序进行发布和消费。Sharding Key是顺序消息中用来区分不同分区的关键字段,和普通消息的Key是完全不同的概念。

更多信息,请参见顺序消息

前提条件

您已完成以下操作:

  • 安装Java SDK。更多信息,请参见准备环境

  • 创建资源。代码中涉及的资源信息,例如实例、Topic和Group ID等,需要在控制台上提前创建。更多信息,请参见创建资源

  • 获取阿里云访问密钥AccessKey ID和AccessKey Secret。更多信息,请参见创建AccessKey

发送顺序消息

重要

云消息队列 RocketMQ 版服务端判定消息产生的顺序性是参照单一生产者、单一线程并发下消息发送的时序。如果发送方有多个生产者或者有多个线程并发发送消息,则此时只能以到达云消息队列 RocketMQ 版服务端的时序作为消息顺序的依据,和业务侧的发送顺序未必一致。

发送顺序消息的示例代码如下。

import com.aliyun.mq.http.MQClient;
import com.aliyun.mq.http.MQProducer;
import com.aliyun.mq.http.model.TopicMessage;

import java.util.Date;

public class OrderProducer {

    public static void main(String[] args) {
        MQClient mqClient = new MQClient(
                // 设置HTTP协议客户端接入点,进入消息队列RocketMQ版控制台实例详情页面的接入点区域查看。
                "${HTTP_ENDPOINT}",
                // 请确保环境变量ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已设置。
                // AccessKey ID,阿里云身份验证标识。
	              System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
	              // AccessKey Secret,阿里云身份验证密钥。
	              System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
        );

        // 消息所属的Topic,在消息队列RocketMQ版控制台创建。
        // 不同消息类型的Topic不能混用,例如普通消息的Topic只能用于收发普通消息,不能用于收发其他类型的消息。
        final String topic = "${TOPIC}";
        // Topic所属的实例ID,在消息队列RocketMQ版控制台创建。
        // 若实例有命名空间,则实例ID必须传入;若实例无命名空间,则实例ID传入null空值或字符串空值。实例的命名空间可以在消息队列RocketMQ版控制台的实例详情页面查看。
        final String instanceId = "${INSTANCE_ID}";

        // 获取Topic的生产者。
        MQProducer producer;
        if (instanceId != null && instanceId != "") {
            producer = mqClient.getProducer(instanceId, topic);
        } else {
            producer = mqClient.getProducer(topic);
        }

        try {
            // 循环发送8条消息。
            for (int i = 0; i < 8; i++) {
                TopicMessage pubMsg = new TopicMessage(
                        // 消息内容。
                        "hello mq!".getBytes(),
                        // 消息标签。
                        "A"
                );
                // 设置分区顺序消息的Sharding Key,用于标识不同的分区。Sharding Key与消息的Key是完全不同的概念。
                pubMsg.setShardingKey(String.valueOf(i % 2));
                pubMsg.getProperties().put("a", String.valueOf(i));
                // 同步发送消息,只要不抛异常就是成功。
                TopicMessage pubResultMsg = producer.publishMessage(pubMsg);

                // 同步发送消息,只要不抛异常就是成功。
                System.out.println(new Date() + " Send mq message success. Topic is:" + topic + ", msgId is: " + pubResultMsg.getMessageId()
                        + ", bodyMD5 is: " + pubResultMsg.getMessageBodyMD5());
            }
        } catch (Throwable e) {
            // 消息发送失败,需要进行重试处理,可重新发送这条消息或持久化这条数据进行补偿处理。
            System.out.println(new Date() + " Send mq message failed. Topic is:" + topic);
            e.printStackTrace();
        }

        mqClient.close();
    }

}

订阅顺序消息

订阅顺序消息的示例代码如下。

import com.aliyun.mq.http.MQClient;
import com.aliyun.mq.http.MQConsumer;
import com.aliyun.mq.http.common.AckMessageException;
import com.aliyun.mq.http.model.Message;

import java.util.ArrayList;
import java.util.List;

public class OrderConsumer {

    public static void main(String[] args) {
        MQClient mqClient = new MQClient(
                // 设置HTTP协议客户端接入点,进入消息队列RocketMQ版控制台实例详情页面的接入点区域查看。
                "${HTTP_ENDPOINT}",
                // 请确保环境变量ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已设置。
                // AccessKey ID,阿里云身份验证标识。
	              System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
	              // AccessKey Secret,阿里云身份验证密钥。
	              System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
        );

        // 消息所属的Topic,在消息队列RocketMQ版控制台创建。
        //不同消息类型的Topic不能混用,例如普通消息的Topic只能用于收发普通消息,不能用于收发其他类型的消息。
        final String topic = "${TOPIC}";
        // 您在消息队列RocketMQ版控制台创建的Group ID。
        final String groupId = "${GROUP_ID}";
        // Topic所属的实例ID,在消息队列RocketMQ版控制台创建。
        // 若实例有命名空间,则实例ID必须传入;若实例无命名空间,则实例ID传入null空值或字符串空值。实例的命名空间可以在消息队列RocketMQ版控制台的实例详情页面查看。
        final String instanceId = "${INSTANCE_ID}";

        final MQConsumer consumer;
        if (instanceId != null && instanceId != "") {
            consumer = mqClient.getConsumer(instanceId, topic, groupId, null);
        } else {
            consumer = mqClient.getConsumer(topic, groupId);
        }

        // 在当前线程循环消费消息,建议多开个几个线程并发消费消息。
        do {
            List<Message> messages = null;

            try {
                // 长轮询顺序消费消息, 拉取到的消息可能是多个分区的(对于分区顺序消息),一个分区内的消息一定是顺序的。
                // 对于分区顺序消息,只要一个分区内存在没有被确认消费的消息,那么该分区下次还会消费到相同的消息。
                // 对于一个分区,只有所有消息确认消费成功才能消费下一批消息。
                // 长轮询表示如果Topic没有消息,则请求会在服务端挂起3s,3s内如果有消息可以消费则服务端立即返回响应。
                messages = consumer.consumeMessageOrderly(
                        3,  // 一次最多消费3条消息(最多可设置为16条)。
                        3   // 长轮询时间为3秒(最多可设置为30秒)。
                );
            } catch (Throwable e) {
                e.printStackTrace();
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
            }
            // Topic中没有消息可消费。
            if (messages == null || messages.isEmpty()) {
                System.out.println(Thread.currentThread().getName() + ": no new message, continue!");
                continue;
            }

            // 处理业务逻辑。
            System.out.println("Receive " + messages.size() + " messages:");
            for (Message message : messages) {
                System.out.println(message);
                System.out.println("ShardingKey: " + message.getShardingKey() + ", a:" + message.getProperties().get("a"));
            }

            // 消息重试时间到达前若不确认消息消费成功,则消息会被重复消费。
            // 消息句柄有时间戳,同一条消息每次消费拿到的都不一样。
            {
                List<String> handles = new ArrayList<String>();
                for (Message message : messages) {
                    handles.add(message.getReceiptHandle());
                }

                try {
                    consumer.ackMessage(handles);
                } catch (Throwable e) {
                    // 某些消息的句柄可能超时,会导致消息消费状态确认不成功。
                    if (e instanceof AckMessageException) {
                        AckMessageException errors = (AckMessageException) e;
                        System.out.println("Ack message fail, requestId is:" + errors.getRequestId() + ", fail handles:");
                        if (errors.getErrorMessages() != null) {
                            for (String errorHandle :errors.getErrorMessages().keySet()) {
                                System.out.println("Handle:" + errorHandle + ", ErrorCode:" + errors.getErrorMessages().get(errorHandle).getErrorCode()
                                        + ", ErrorMsg:" + errors.getErrorMessages().get(errorHandle).getErrorMessage());
                            }
                        }
                        continue;
                    }
                    e.printStackTrace();
                }
            }
        } while (true);
    }
}
上一篇: 收发普通消息 下一篇: 收发定时消息和延时消息