Spring 整合 RocketMQ

1. 引入jar包

<!-- RocketMQ -->
        <dependency>  
             <groupId>com.alibaba.rocketmq</groupId>  
             <artifactId>rocketmq-all</artifactId>  
             <version>3.2.6</version>  
             <type>pom</type>  
        </dependency>
        <dependency>  
             <groupId>com.alibaba.rocketmq</groupId>  
             <artifactId>rocketmq-client</artifactId>  
             <version>3.2.6</version>  
        </dependency>      

2.Spring bean 配置單例

<bean id="myProducer" class="cn.zno.rocketmq.MyProducer"
            init-method="init"  
            destroy-method="destroy"
            scope="singleton">
　　　　　　<property name="producerGroup" value="MyProducerGroup" />
　　　　　　<property name="namesrvAddr" value="127.0.0.1:9876" />
    </bean>
    <bean class="cn.zno.rocketmq.MyConsumer" 
            init-method="init" 
            destroy-method="destroy"
            scope="singleton">
　　　　　　<property name="consumerGroup" value="MyConsumerGroup" />
　　　　　　<property name="namesrvAddr" value="127.0.0.1:9876" />
    </bean>      

3. 自定義producer

package cn.zno.rocketmq;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;

public class MyProducer {

    private final Logger logger = LoggerFactory.getLogger(MyProducer.class);

    private DefaultMQProducer defaultMQProducer;
    private String producerGroup;
    private String namesrvAddr;

    /**
     * Spring bean init-method
     */
    public void init() throws MQClientException {
        // 參數資訊
        logger.info("DefaultMQProducer initialize!");
        logger.info(producerGroup);
        logger.info(namesrvAddr);

        // 初始化
        defaultMQProducer = new DefaultMQProducer(producerGroup);
        defaultMQProducer.setNamesrvAddr(namesrvAddr);
        defaultMQProducer.setInstanceName(String.valueOf(System.currentTimeMillis()));
        
        defaultMQProducer.start();

　　　　 logger.info("DefaultMQProudcer start success!");

    }

    /**
     * Spring bean destroy-method
     */
    public void destroy() {
        defaultMQProducer.shutdown();
    }

    public DefaultMQProducer getDefaultMQProducer() {
        return defaultMQProducer;
    }

    // ---------------setter -----------------

    public void setProducerGroup(String producerGroup) {
        this.producerGroup = producerGroup;
    }

    public void setNamesrvAddr(String namesrvAddr) {
        this.namesrvAddr = namesrvAddr;
    }

}      

4. 自定義consumer

package cn.zno.rocketmq;

import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;

public class MyConsumer {

    private final Logger logger = LoggerFactory.getLogger(MyConsumer.class);

    private DefaultMQPushConsumer defaultMQPushConsumer;
    private String namesrvAddr;
    private String consumerGroup;

    /**
     * Spring bean init-method
     */
    public void init() throws InterruptedException, MQClientException {

        // 參數資訊
        logger.info("DefaultMQPushConsumer initialize!");
        logger.info(consumerGroup);
        logger.info(namesrvAddr);

        // 一個應用建立一個Consumer，由應用來維護此對象，可以設定為全局對象或者單例<br>
        // 注意：ConsumerGroupName需要由應用來保證唯一
        defaultMQPushConsumer = new DefaultMQPushConsumer(consumerGroup);
        defaultMQPushConsumer.setNamesrvAddr(namesrvAddr);
        defaultMQPushConsumer.setInstanceName(String.valueOf(System.currentTimeMillis()));

        // 訂閱指定MyTopic下tags等于MyTag

        defaultMQPushConsumer.subscribe("MyTopic", "MyTag");

        // 設定Consumer第一次啟動是從隊列頭部開始消費還是隊列尾部開始消費<br>
        // 如果非第一次啟動，那麼按照上次消費的位置繼續消費
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        // 設定為叢集消費(差別于廣播消費)
        defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING);

        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {

            // 預設msgs裡隻有一條消息，可以通過設定consumeMessageBatchMaxSize參數來批量接收消息
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                MessageExt msg = msgs.get(0);
                if (msg.getTopic().equals("MyTopic")) {
                    // TODO 執行Topic的消費邏輯
                    if (msg.getTags() != null && msg.getTags().equals("MyTag")) {
                        // TODO 執行Tag的消費
                    }
                }
                // 如果沒有return success ，consumer會重新消費該消息，直到return success
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // Consumer對象在使用之前必須要調用start初始化，初始化一次即可<br>
        defaultMQPushConsumer.start();

        logger.info("DefaultMQPushConsumer start success!");
    }

    /**
     * Spring bean destroy-method
     */
    public void destroy() {
        defaultMQPushConsumer.shutdown();
    }

    // ----------------- setter --------------------

    public void setNamesrvAddr(String namesrvAddr) {
        this.namesrvAddr = namesrvAddr;
    }

    public void setConsumerGroup(String consumerGroup) {
        this.consumerGroup = consumerGroup;
    }

}      

@Autowired
    private MyProducer myProducer;

    public void sendMessage() {
        Message msg = new Message("MyTopic", "MyTag", (JSONObject.fromObject(someMessage)).getBytes());
        SendResult sendResult = null;
        try {
            sendResult = myProducer.getDefaultMQProducer().send(msg);
        } catch (MQClientException e) {
            logger.error(e.getMessage() + String.valueOf(sendResult));
        }
        // 當消息發送失敗時如何處理
        if (sendResult == null || sendResult.getSendStatus() != SendStatus.SEND_OK) {
            // TODO
        }
    }