淺談Springboot整合RocketMQ使用心得

一、阿里雲官網---幫助文檔

https://help.aliyun.com/document_detail/29536.html?spm=5176.doc29535.6.555.WWTIUh

按照官網步驟，創建Topic、申請發布（生產者）、申請訂閱（消費者）

二、代碼

1、配置：

 public class MqConfig { /** * 啟動測試之前請替換如下XXX 為您的配置*/ public static final String PUBLIC_TOPIC = "test";//公網測試public static final String PUBLIC_PRODUCER_ID = "PID_SCHEDULER"; public static final String PUBLIC_CONSUMER_ID = "CID_SERVICE"; public static final String ACCESS_KEY = "123"; public static final String SECRET_KEY = "123"; public static final String TAG = ""; public static final String THREAD_NUM = "25";//消費端線程數/** * 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 */ public static final String ONSADDR = "http://onsaddr-internal.aliyun.com:8080/rocketmq/nsaddr4client-internal";}

ONSADDR 阿里雲用公有云生產，測試用公網

不同的業務可以設置不同的tag，但是如果發送消息量大的話，建議新建TOPIC

2、生產者

方式1：

配置文件：producer.xml

 <?xml version="1.0" encoding="UTF-8"?><!DOCTYPE beans PUBLIC "-//SPRING//DTD BEAN//EN" "http://www.springframework.org/dtd/spring-beans.dtd"><beans> <bean id="producer" init-method="start" destroy-method="shutdown"> <property name="properties"> <map> <entry key="ProducerId" value="" /> <!-- PID,請替換--> <entry key="AccessKey" value="" /> <!-- ACCESS_KEY,請替換--> <entry key="SecretKey" value="" /> <!-- SECRET_KEY,請替換--> <!--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 --> <entry key="ONSAddr" value="http://onsaddr-internal.aliyun.com:8080/rocketmq/nsaddr4client-internal"/> </map> </property> </bean></beans>

啟動方式1，在使用類的全局裡設置：

 //初始化生產者private ApplicationContext ctx; private ProducerBean producer; @Value("${producerConfig.enabled}")//開關，spring配置項，true為開啟，false關閉private boolean producerConfigEnabled; @PostConstruct public void init(){ if (true == producerConfigEnabled) { ctx = new ClassPathXmlApplicationContext("producer.xml"); producer = (ProducerBean) ctx.getBean("producer"); } }

PS：最近發現一個坑，如果producer用上面這種方式啟動的話，一旦啟動的多了，會造成fullGC，所以可以換成下面這種註解方式啟動，在用到的地方手動start、shutdown

方式2：配置類（不需要xml）

 @Configurationpublic class ProducerBeanConfig { @Value("${openservices.ons.producerBean.producerId}") private String producerId; @Value("${openservices.ons.producerBean.accessKey}") private String accessKey; @Value("${openservices.ons.producerBean.secretKey}") private String secretKey; private ProducerBean producerBean; @Value("${openservices.ons.producerBean.ONSAddr}") private String ONSAddr; @Bean public ProducerBean oneProducer() { ProducerBean producerBean = new ProducerBean(); Properties properties = new Properties(); properties.setProperty(PropertyKeyConst.ProducerId, producerId); properties.setProperty(PropertyKeyConst.AccessKey, accessKey); properties.setProperty(PropertyKeyConst.SecretKey, secretKey); properties.setProperty(PropertyKeyConst.ONSAddr, ONSAddr); producerBean.setProperties(properties); return producerBean; }}

PS：經過這次雙11發現，以上2種方式在大數據量，多線程情況下都不太適用， 性能很差，所以推薦用3

方式3：（不需要xml）

 @Componentpublic class ProducerBeanSingleTon { @Value("${openservices.ons.producerBean.producerId}") private String producerId; @Value("${openservices.ons.producerBean.accessKey}") private String accessKey; @Value("${openservices.ons.producerBean.secretKey}") private String secretKey; @Value("${openservices.ons.producerBean.ONSAddr}") private String ONSAddr; private static Producer producer; private static class SingletonHolder { private static final ProducerBeanSingleTon INSTANCE = new ProducerBeanSingleTon(); } private ProducerBeanSingleTon (){} public static final ProducerBeanSingleTon getInstance() { return SingletonHolder.INSTANCE; } @PostConstruct public void init(){ // producer 實例配置初始化Properties properties = new Properties(); //您在控制台創建的Producer ID properties.setProperty(PropertyKeyConst.ProducerId, producerId); // AccessKey 阿里雲身份驗證，在阿里雲服務器管理控制台創建properties.setProperty(PropertyKeyConst.AccessKey, accessKey); // SecretKey 阿里雲身份驗證，在阿里雲服務器管理控制台創建properties.setProperty(PropertyKeyConst.SecretKey, secretKey); //設置發送超時時間，單位毫秒properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000"); // 設置TCP 接入域名（此處以公共雲生產環境為例） properties.setProperty(PropertyKeyConst.ONSAddr, ONSAddr); producer = ONSFactory.createProducer(properties); // 在發送消息前，必須調用start方法來啟動Producer，只需調用一次即可producer.start(); } public Producer getProducer(){ return producer; }}

spring配置

spring.jpa.properties.hibernate.dialect = org.hibernate.dialect.MySQL5DialectconsumerConfig.enabled = trueproducerConfig.enabled = true #方式1：scheduling.enabled = false#方式2、3：rocketMQ /u516C/u7F51/u914D/u7F6Eopenservices.ons.producerBean.producerId = pidopenservices.ons.producerBean.accessKey = openservices.ons.producerBean.secretKey = openservices.ons.producerBean.ONSAddr = 公網、杭州公有云生產

方式1投遞消息代碼：

 try { String jsonC = JsonUtils.toJson(elevenMessage); Message message = new Message(MqConfig.TOPIC, MqConfig.TAG, jsonC.getBytes()); SendResult sendResult = producer.send(message); if (sendResult != null) { logger.info(".Send mq message success!”; } else { logger.warn(".sendResult is null........."); } } catch (Exception e) { logger.warn("DoubleElevenAllPreService"); Thread.sleep(1000);//如果有異常，休眠1秒}

方式2投遞消息代碼：（可以每發1000個啟動/關閉一次）

 producerBean.start();try { String jsonC = JsonUtils.toJson(elevenMessage); Message message = new Message(MqConfig.TOPIC, MqConfig.TAG, jsonC.getBytes()); SendResult sendResult = producer.send(message); if (sendResult != null) { logger.info(".Send mq message success!”; } else { logger.warn(".sendResult is null........."); } } catch (Exception e) { logger.warn("DoubleElevenAllPreService"); Thread.sleep(1000);//如果有異常，休眠1秒} producerBean.shutdown();

方式3：投遞消息

try { String jsonC = JsonUtils.toJson(elevenMessage); Message message = new Message(MqConfig.TOPIC, MqConfig.TAG, jsonC.getBytes()); Producer producer = ProducerBeanSingleTon.getInstance().getProducer(); SendResult sendResult = producer.send(message); if (sendResult != null) { logger.info("DoubleElevenMidService.Send mq message success! Topic is:"”; } else { logger.warn("DoubleElevenMidService.sendResult is null........."); } } catch (Exception e) { logger.error("DoubleElevenMidService Thread.sleep 1 s___error is "+e.getMessage(), e); Thread.sleep(1000);//如果有異常，休眠1秒}

發送消息的代碼一定要捕獲異常，不然會重複發送。

這裡的TOPIC用自己創建的，elevenMessage是要發送的內容，我這裡是自己建的對象

3、消費者

配置啟動類：

 @Configuration@ConditionalOnProperty(value = "consumerConfig.enabled", havingValue = "true", matchIfMissing = true)public class ConsumerConfig { private Logger logger = LoggerFactory.getLogger(LoggerAppenderType.smsdist.name()); @Bean public Consumer consumerFactory(){//不同消費者這裡不能重名Properties consumerProperties = new Properties(); consumerProperties.setProperty(PropertyKeyConst.ConsumerId, MqConfig.CONSUMER_ID); consumerProperties.setProperty(PropertyKeyConst.AccessKey, MqConfig.ACCESS_KEY); consumerProperties.setProperty(PropertyKeyConst.SecretKey, MqConfig.SECRET_KEY); //consumerProperties.setProperty(PropertyKeyConst.ConsumeThreadNums,MqConfig.THREAD_NUM); consumerProperties.setProperty(PropertyKeyConst.ONSAddr, MqConfig.ONSADDR); Consumer consumer = ONSFactory.createConsumer(consumerProperties); consumer.subscribe(MqConfig.TOPIC, MqConfig.TAG, new DoubleElevenMessageListener());//new對應的監聽器consumer.start(); logger.info("ConsumerConfig start success."); return consumer; }}

CID和ONSADDR一點要選對，用自己的，消費者線程數等可以在這裡配置

創建消息監聽器類，消費消息：

 @Componentpublic class MessageListener implements MessageListener { private Logger logger = LoggerFactory.getLogger("remind"); protected static ElevenReposity elevenReposity; @Resource public void setElevenReposity(ElevenReposity elevenReposity){ MessageListener .elevenReposity=elevenReposity; } @Override public Action consume(Message message, ConsumeContext consumeContext) { if(message.getTopic().equals("自己的TOPIC")){//避免消費到其他消息json轉換報錯try { byte[] body = message.getBody(); String res = new String(body); //res 是生產者傳過來的消息內容//業務代碼}else{ logger.warn("!"); } } catch (Exception e) { logger.error("MessageListener.consume error:" + e.getMessage(), e); } logger.info("MessageListener.Receive message”); //如果想測試消息重投的功能,可以將Action.CommitMessage 替換成Action.ReconsumeLater return Action.CommitMessage; }else{ logger.warn(); return Action.ReconsumeLater; } }

注意，由於消費者是多線程的，所以對像要用static+set注入，把對象的級別提升到進程，這樣多個線程就可以共用，但是無法調用父類的方法和變量

消費者狀態可以查看消費者是否連接成功，消費是否延遲，消費速度等

重置消費位點可以清空所有消息

三、注意事項

1、發送的消息體最大為256KB

2、消息最多存在3天

3、消費端默認線程數是20

4、如果運行過程中出現java掛掉或者cpu佔用異常高，可以在發送消息的時候，每發送1000條讓線程休息1s

5、本地測試或啟動的時候，把ONSADDR換成公網，不然報錯無法啟動

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持武林網。