RocketMq 顺序消费、分区消息、延迟发送消息、Topic、tag分类实战（消费者）（三）

消费端配置
如下所示：是消费者的配置类，有以下几点需要注意的地方
1、是TargetMessageListener这个监听类（下文会把这个监听类的具体代码贴出来），需要把这个监听类订阅。
2、rocketMqDcProperties.getTargetProperties()这个方法里面有相关的配置信息（这里面有绑定消费者是那个组，因为一类Producer或Consumer，这类Producer或Consumer通常生产或消费同一类消息，且消息发布或订阅的逻辑一致），具体代码见下文
3、subscription.setTopic(rocketMqDcProperties.getOrderTopic()) 是绑定Topic，这个Topic跟上篇文章生产者的Topic是一致的，这样就能保证消费者能准确消费到生产者发送的消息
4、subscription.setExpression(MqTagEnum.target.name()); 这个是一个消息的过滤，也是一个对消息的具体分类。详细用法请参考链接：
https://help.aliyun.com/zh/apsaramq-for-rocketmq/cloud-message-queue-rocketmq-4-x-series/developer-reference/message-filtering?spm=a2c4g.11186623.0.i38#concept-2047069
综上所述：
这段代码主要是用于配置一个 OrderConsumerBean 对象，设置其属性和订阅关系

@Configuration
public class TargetConsumerClient {@Autowiredprivate RocketMqDcProperties rocketMqDcProperties;@Autowiredprivate TargetMessageListener messageListener;@ConditionalOnProperty(name = "rocket.mq.dc.enabled", havingValue = "true", matchIfMissing = true)@Bean(initMethod = "start", destroyMethod = "shutdown",name = {"buildTargetConsumer"})public OrderConsumerBean buildTargetConsumer() {OrderConsumerBean orderConsumerBean = new OrderConsumerBean();//配置文件orderConsumerBean.setProperties(rocketMqDcProperties.getTargetProperties());//订阅关系Map<Subscription, MessageOrderListener> subscriptionTable = new HashMap<Subscription, MessageOrderListener>();Subscription subscription = new Subscription();subscription.setTopic(rocketMqDcProperties.getOrderTopic());subscription.setExpression(MqTagEnum.target.name());subscriptionTable.put(subscription, messageListener);orderConsumerBean.setSubscriptionTable(subscriptionTable);return orderConsumerBean;}
}

	public Properties getTargetProperties() {return this.getProperties(this.targetGroupId);}private Properties getProperties(String groupId) {Properties properties = new Properties();properties.setProperty("AccessKey", this.accessKey);properties.setProperty("SecretKey", this.secretKey);properties.setProperty("NAMESRV_ADDR", this.nameSrvAddr);properties.setProperty("GROUP_ID", groupId);properties.setProperty("ConsumeThreadNums", this.getConsumeThreadNums().toString());properties.setProperty("maxReconsumeTimes", this.getMaxReconsumeTimes().toString());properties.setProperty("consumeTimeout", this.getConsumeTimeout().toString());properties.setProperty("suspendTimeMillis", this.getSuspendTimeMillis().toString());return properties;}

消费端代码
如下所示：TargetMessageListener 实现了MessageOrderListener接口，并如上文所示，其和OrderConsumerBean也绑定了订阅关系。

@Slf4j
@Component
public class TargetMessageListener implements MessageOrderListener {@Autowiredprivate MqMessageRecordDao mqMessageRecordDao;@Autowiredprivate TargetService targetServiceImpl;@Overridepublic OrderAction consume(final Message message, final ConsumeOrderContext context) {log.info("MQ消息消费者监听消息内容：{}", message);MqMessageRecordMo mqMessageRecordMo = new MqMessageRecordMo();try {String body = new String(message.getBody());String tag = message.getTag();mqMessageRecordMo.setMsgId(message.getMsgID());mqMessageRecordMo.setOrderTopic(message.getTopic());mqMessageRecordMo.setProducerIp(message.getBornHost());mqMessageRecordMo.setTag(tag);//mqMessageRecordMo.setMessageKey(message.getKey());mqMessageRecordMo.setShardingKey(message.getShardingKey());mqMessageRecordMo.setBodyJson(body)mqMessageRecordMo.setProducerTime(LocalDateLUtils.timestampToDatetime(message.getBornTimestamp()));mqMessageRecordMo.setCreateTime(LocalDateTime.now());mqMessageRecordMo.setPMsgId(message.getUserProperties("pMsgId"));log.info("MQ消费者消息消费成功，解析并处理相应的业务逻辑, tag = {}，key = {}，messageId = {}", tag, message.getShardingKey(),message.getMsgID());DataBaseEnum dataBase = DataBaseEnum.getEnum(message.getShardingKey());targetServiceImpl.process(message.getMsgID(),dataBase,body);log.info("MQ消息体消费监听解析结果：{}", body);mqMessageRecordMo.setIsSuccess(true);return OrderAction.Success;} catch (Exception e) {//消费失败，挂起当前队列// 存储错误消息，重试，记录日志/*log.error("target MQ消费者消息监听消息业务逻辑处理失败:",e);mqMessageRecordMo.setIsSuccess(false);mqMessageRecordMo.setErrorMsg(e.getMessage());return OrderAction.Success;} finally {mqMessageRecordDao.save(mqMessageRecordMo);}}
}

在这里想讲下顺序消息
顺序消息
顺序消息可以保证消息的消费顺序和发送的顺序一致，即先发送的先消费，后发送的后消费，常用于金融证券、电商业务等对消息指令顺序有严格要求的场景。本文介绍云消息队列 RocketMQ 版顺序消息的概念、适用场景、实现原理以及使用过程中的注意事项。
什么是顺序消息
顺序消息是云消息队列 RocketMQ 版提供的一种对消息发送和消费顺序有严格要求的消息。对于一个指定的Topic，消息严格按照先进先出（FIFO）的原则进行消息发布和消费，即先发布的消息先消费，后发布的消息后消费。
顺序消息分为分区顺序消息和全局顺序消息。
分区顺序消息
对于指定的一个Topic，所有消息根据Sharding Key进行区块分区，同一个分区内的消息按照严格的先进先出（FIFO）原则进行发布和消费。同一分区内的消息保证顺序，不同分区之间的消息顺序不做要求。
适用场景
适用于性能要求高，以Sharding Key作为分区字段，在同一个区块中严格地按照先进先出（FIFO）原则进行消息发布和消费的场景。
示例
用户注册需要发送验证码，以用户ID作为Sharding Key，那么同一个用户发送的消息都会按照发布的先后顺序来消费。
电商的订单创建，以订单ID作为Sharding Key，那么同一个订单相关的创建订单消息、订单支付消息、订单退款消息、订单物流消息都会按照发布的先后顺序来消费。
阿里巴巴集团内部电商系统均使用分区顺序消息，既保证业务的顺序，同时又能保证业务的高性能。
全局顺序消息
对于指定的一个Topic，所有消息按照严格的先入先出（FIFO）的顺序来发布和消费。
适用场景
适用于性能要求不高，所有的消息严格按照FIFO原则来发布和消费的场景。
示例
在证券处理中，以人民币兑换美元为Topic，在价格相同的情况下，先出价者优先处理，则可以按照FIFO的方式发布和消费全局顺序消息。
说明
全局顺序消息实际上是一种特殊的分区顺序消息，即Topic中只有一个分区，因此全局顺序和分区顺序的实现原理相同。因为分区顺序消息有多个分区，所以分区顺序消息比全局顺序消息的并发度和性能更高
如何实现顺序消息
在这里插入图片描述
在云消息队列 RocketMQ 版中，消息的顺序需要由以下三个阶段保证：
消息发送
如上图所示，A1、B1、A2、A3、B2、B3是订单A和订单B的消息产生的顺序，业务上要求同一订单的消息保持顺序，例如订单A的消息发送和消费都按照A1、A2、A3的顺序。如果是普通消息，订单A的消息可能会被轮询发送到不同的队列中，不同队列的消息将无法保持顺序，而顺序消息发送时云消息队列 RocketMQ 版支持将Sharding Key相同（例如同一订单号）的消息序路由到一个队列中。
云消息队列 RocketMQ 版服务端判定消息产生的顺序性是参照同一生产者发送消息的时序。不同生产者、不同线程并发产生的消息，云消息队列 RocketMQ 版服务端无法判定消息的先后顺序。
消息存储
如上图所示，顺序消息的Topic中，每个逻辑队列对应一个物理队列，当消息按照顺序发送到Topic中的逻辑队列时，每个分区的消息将按照同样的顺序存储到对应的物理队列中。
消息消费
云消息队列 RocketMQ 版按照存储的顺序将消息投递给Consumer，Consumer收到消息后也不对消息顺序做任何处理，按照接收到的顺序进行消费。
Consumer消费消息时，同一Sharding Key的消息使用单线程消费，保证消息消费顺序和存储顺序一致，最终实现消费顺序和发布顺序的一致
注意事项
a、同一个Group ID只对应一种类型的Topic，即不同时用于顺序消息和无序消息的收发。
b、对于全局顺序消息，建议消息不要有阻塞。同时运行多个实例，是为了防止工作实例意外退出而导致业务中断。当工作实例退出时，其他实例可以立即接手工作，不会导致业务中断，实际工作的只会有一个实例。
c、云消息队列 RocketMQ 版服务端判定消息产生的顺序性是参照单一生产者、单一线程并发下消息发送的时序。如果发送方有多个生产者或者有多个线程并发发送消息，则此时只能以到达云消息队列 RocketMQ 版服务端的时序作为消息顺序的依据，和业务侧的发送顺序未必一致

顺序消息常见问题
a、同一条消息是否可以既是顺序消息，又是定时消息和事务消息？
不可以。顺序消息、定时消息、事务消息是不同的消息类型，三者是互斥关系，不能叠加在一起使用。

b、顺序消息支持哪些地域？
支持云消息队列 RocketMQ 版所有公共云地域和金融云地域。

c、为什么全局顺序消息性能一般？
全局顺序消息是严格按照FIFO的消息阻塞原则，即上一条消息没有被成功消费，那么下一条消息会一直被存储到Topic队列中。如果想提高全局顺序消息的TPS，可以升级实例配置，同时消息客户端应用尽量减少处理本地业务逻辑的耗时。

d、顺序消息支持哪种消息发送方式？
顺序消息只支持可靠同步发送方式，不支持异步发送方式，否则将无法严格保证顺序。

e、顺序消息是否支持集群消费和广播消费？
顺序消息暂时仅支持集群消费模式，不支持广播消费模式。
以上文档链接来源：
https://help.aliyun.com/zh/apsaramq-for-rocketmq/cloud-message-queue-rocketmq-4-x-series/developer-reference/ordered-messages?spm=a2c4g.11186623.0.0.34b428e5LL1Jlh