阿里MQ普通+顺序+延时消息整合Spring

傷城~ 2022-06-12 20:49 285阅读 0赞

# 前言 #

由于公司项目需要，研究了下AliWareMQ。阿里mq的普通消息和延时消息还是挺简单的。不过在顺序消息的时候出现了一些瓶颈。后来查阅源码和依据demo整理了一版融合Spring的版本。

# 实例 #

## mq配置文件(Spring) ##

主要是顺序消息的配置，以及多实例的配置(需要在控制台配置p/c)

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation=" http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.2.xsd">
        
        <bean id="producer" class="com.aliyun.openservices.ons.api.bean.ProducerBean" init-method="start" destroy-method="shutdown">
            <property name="properties">
                <map>
                    <entry key="ProducerId" value="${ProducerId}"/> 
                    <entry key="AccessKey" value="${AccessKey}"/> 
                    <entry key="SecretKey" value="${SecretKey}"/> 
                    <entry key="ONSAddr" value="${ONSAddr}"/>
                </map>
            </property>
        </bean>
        
        <bean id="mqOrderProducer" class="com.rqbao.mq.listenter.MQOrderProducer" init-method="start" destroy-method="shutdown">
            <constructor-arg name="producerProperties">
                <props>
                    <prop key="ProducerId">${OrderProducerId}</prop>
                    <prop key="AccessKey">${AccessKey}</prop>
                    <prop key="SecretKey">${SecretKey}</prop>
                    <prop key="ONSAddr">${ONSAddr}</prop>
                    <prop key="ORDER_TOPIC">${ORDER_TOPIC}</prop>
                    <prop key="TAG">1111</prop>
                </props>
            </constructor-arg>
        </bean>
        
    
        
        <bean id="consumer" class="com.aliyun.openservices.ons.api.bean.ConsumerBean" init-method="start" destroy-method="shutdown">
            <property name="properties">
                <map>
                    <entry key="ConsumerId" value="${OrderConsumerId}"/> 
                    <entry key="AccessKey" value="${AccessKey}"/> 
                    <entry key="SecretKey" value="${SecretKey}"/>
                    <entry key="ONSAddr" value="${ONSAddr}"/>
                </map>
            </property>
    
            <property name="subscriptionTable">
                <map>
                    <entry value-ref="messageListener">
                        <key>
                            <bean class="com.aliyun.openservices.ons.api.bean.Subscription">
                                <property name="topic" value="${topic}"/> 
                                <property name="expression" value="*"/>
                            </bean>
                        </key>
                    </entry>
                </map>
    
            </property>
        </bean>
        
        <bean id="consumerOrder" class="com.rqbao.mq.listenter.MQOrderConsumer" init-method="start" destroy-method="shutdown">
            <constructor-arg name="consumerProperties">
                <props>
                    <prop key="ConsumerId">${ConsumerId}</prop>
                    <prop key="AccessKey">${AccessKey}</prop>
                    <prop key="SecretKey">${SecretKey}</prop>
                    <prop key="ONSAddr">${ONSAddr}</prop>
                    <prop key="ORDER_TOPIC">${ORDER_TOPIC}</prop>
                    <prop key="TAG">1111</prop>
                </props>
            </constructor-arg>
        </bean>
        
        <bean id="messageListener" class="com.rqbao.mq.listenter.MQListener"/>
        
    
        
    </beans>

## 顺序生产者 ##

package com.rqbao.mq.listenter;
    
    import com.aliyun.openservices.ons.api.ONSFactory;
    import com.aliyun.openservices.ons.api.order.OrderProducer;
    
    import java.util.Properties;
    
    /** * Created by ricky on 2017/7/2. */
    public class MQOrderProducer { 
        private  Properties producerProperties = new Properties();
        private  OrderProducer producer;
    
        public MQOrderProducer(Properties producerProperties) {
            this.producerProperties = producerProperties;
            producer = ONSFactory.createOrderProducer(producerProperties);
        }
        public OrderProducer getOrderProducer(){
            return  producer;
        }
        public  void  start(){
            new MQOrderProducer(producerProperties);
            producer.start();
        }
        public  void  shutdown(){
            producer.shutdown();
        }
    
        public Properties getProducerProperties() {
            return producerProperties;
        }
    
        public void setProducerProperties(Properties producerProperties) {
            this.producerProperties = producerProperties;
        }
    
        public OrderProducer getProducer() {
            return producer;
        }
    
        public void setProducer(OrderProducer producer) {
            this.producer = producer;
        }
    }

## 顺序消费者 ##

package com.rqbao.mq.listenter;
    
    import com.aliyun.openservices.ons.api.Message;
    import com.aliyun.openservices.ons.api.ONSFactory;
    import com.aliyun.openservices.ons.api.order.ConsumeOrderContext;
    import com.aliyun.openservices.ons.api.order.MessageOrderListener;
    import com.aliyun.openservices.ons.api.order.OrderAction;
    import com.aliyun.openservices.ons.api.order.OrderConsumer;
    
    import java.util.Properties;
    
    /** * Created by ricky on 2017/7/2. */
    public class MQOrderConsumer { 
        private  Properties consumerProperties = new Properties();
        private  OrderConsumer consumer;
    
        public MQOrderConsumer(Properties consumerProperties) {
            this.consumerProperties = consumerProperties;
            consumer = ONSFactory.createOrderedConsumer(consumerProperties);
            consumer.subscribe(consumerProperties.get("ORDER_TOPIC").toString(), consumerProperties.get("TAG").toString(),  new MessageOrderListener() {
                @Override
                public OrderAction consume(final Message message, final ConsumeOrderContext context) {
                    System.out.println(message);
                    return OrderAction.Success;
                }
            });
        }
        public  void  start(){
            new MQOrderProducer(consumerProperties);
            consumer.start();
        }
        public  void  shutdown(){
            consumer.shutdown();
        }
    
        public Properties getConsumerProperties() {
            return consumerProperties;
        }
    
        public void setConsumerProperties(Properties consumerProperties) {
            this.consumerProperties = consumerProperties;
        }
    
        public OrderConsumer getConsumer() {
            return consumer;
        }
    
        public void setConsumer(OrderConsumer consumer) {
            this.consumer = consumer;
        }
    }

## 参数文件 ##

#ALiMQ相关参数
    #入钥
    AccessKey=XXXXXX
    #秘钥
    SecretKey=XXXXXXX
    #阿里云MQ线上地址
    #PropertyKeyConst.ONSAddr 请根据不同Region进行配置
    #公网测试: http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet
    #公有云生产: http://onsaddr-internal.aliyun.com:8080/rocketmq/nsaddr4client-internal
    #杭州金融云: http://jbponsaddr-internal.aliyun.com:8080/rocketmq/nsaddr4client-internal
    #深圳金融云: http://mq4finance-sz.addr.aliyun.com:8080/rocketmq/nsaddr4client-internal
    ONSAddr=http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet
    #主题
    topic=RICKY_CGBUSINESS_LOCAL
    ORDER_TOPIC=RICKY_CGBUSINESS_TEST
    #生产者Id
    ProducerId=PID_RICKYCG_LOCAL
    OrderProducerId=PID_RICKYCG_TEST
    #消费者
    ConsumerId=CID_RICKYCG_LOCAL
    OrderConsumerId=CID_RICKYCG_TEST

## 顺序消息生产 ##

mport com.aliyun.openservices.ons.api.Message;
    import com.aliyun.openservices.ons.api.SendResult;
    import com.aliyun.openservices.ons.api.order.OrderProducer;
    import com.rqbao.mq.common.config.Global;
    import org.apache.log4j.Logger;
    import org.springframework.beans.factory.InitializingBean;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    import java.util.Date;
    import java.util.UUID;
    
    /** * Created by ricky on 2017/6/8. *MQ消息生产功能，可能附加一些其他功能 */
    @Service
    public class MQOrderService implements InitializingBean{ 
        private static final Logger logger = Logger.getLogger(MQOrderService.class);
        @Autowired
        MQOrderProducer mqOrderProducer;
    
        /** * 发送顺序MQ消息 * @param TAG 业务tag tag将使用CGUTILS中的存管接口变量 * @param msg 需传递对应接口需要参数，格式待考究 * delayTime 延迟时间 3000--->3s * @return */
        public String sendOrderMQMsg(String TAG,String msg){
            OrderProducer producer=mqOrderProducer.getOrderProducer();
            Message message = new Message(Global.getConfig("ORDER_TOPIC"), TAG,msg.getBytes());
            // 设置代表消息的业务关键属性，请尽可能全局唯一。
            String orderId = "rqb_" + UUID.randomUUID().toString().replaceAll("-","");
            message.setKey(orderId);
            // 分区顺序消息中区分不同分区的关键字段，sharding key于普通消息的key是完全不同的概念。
            // 全局顺序消息，该字段可以设置为任意非空字符串。
            String shardingKey = String.valueOf(orderId);
            SendResult sendResult = producer.send(message, shardingKey);
            if (sendResult != null) {
                System.out.println(new Date() + " Send mq message success! Topic is:" + Global.getConfig("topic") + "msgId is: " + sendResult.getMessageId());
            }
            return sendResult.toString();
        }
        @Override
        public void afterPropertiesSet() throws Exception {
    
        }
    }

## 顺序消息消费 ##

具体在MQOrderConsumer中写消费处理

## 关于延时和定时消息 ##

依据官方说明，延时消息必须是无序的topic，切记。  
![这里写图片描述][SouthEast]

可参阅[官方文档][Link 1]

# 问题 #

MQ存在重复消费的问题，我这里的解决方案是进行MQ消息记录，比如消息有tag再加上body中相关的业务id，联合组成唯一索引，避免重复消费问题。

# 附demo地址： #

[http://git.oschina.net/bgt0314/alimq\_demo][http_git.oschina.net_bgt0314_alimq_demo]

# 交流群 #

244930845 不懂的欢迎进群咨询。

[SouthEast]: /images/20220613/4f039ca120da404696d40ce28b5c9160.png
[Link 1]: https://help.aliyun.com/document_detail/49319.html
[http_git.oschina.net_bgt0314_alimq_demo]: http://git.oschina.net/bgt0314/alimq_demo