springboot环境下的rokectMQ多数据源实现

业务原因，必要在一个项目中与多方MQ举行业务通信；
步调一，复制一份RocketMQProperties配置文件，制止与原来的冲突

1. package com.heit.road.web.config;
3. import org.apache.rocketmq.common.topic.TopicValidator;
5. import java.util.HashMap;
6. import java.util.Map;
8. public class MultipleRocketMQProperties {
10.     /**
11.      * The name server for rocketMQ, formats: `host:port;host:port`.
12.      */
13.     private String nameServer;
15.     /**
16.      * Enum type for accessChannel, values: LOCAL, CLOUD
17.      */
18.     private String accessChannel;
20.     private org.apache.rocketmq.spring.autoconfigure.RocketMQProperties.Producer producer;
22.     /**
23.      * Configure enable listener or not.
24.      * In some particular cases, if you don't want the the listener is enabled when container startup,
25.      * the configuration pattern is like this :
26.      * rocketmq.consumer.listeners.<group-name>.<topic-name>.enabled=<boolean value, true or false>
27.      * <p>
28.      * the listener is enabled by default.
29.      */
30.     private org.apache.rocketmq.spring.autoconfigure.RocketMQProperties.Consumer consumer = new org.apache.rocketmq.spring.autoconfigure.RocketMQProperties.Consumer();
32.     public String getNameServer() {
33.         return nameServer;
34.     }
36.     public void setNameServer(String nameServer) {
37.         this.nameServer = nameServer;
38.     }
40.     public String getAccessChannel() {
41.         return accessChannel;
42.     }
44.     public void setAccessChannel(String accessChannel) {
45.         this.accessChannel = accessChannel;
46.     }
48.     public org.apache.rocketmq.spring.autoconfigure.RocketMQProperties.Producer getProducer() {
49.         return producer;
50.     }
52.     public void setProducer(org.apache.rocketmq.spring.autoconfigure.RocketMQProperties.Producer producer) {
53.         this.producer = producer;
54.     }
56.     public static class Producer {
58.         /**
59.          * Group name of producer.
60.          */
61.         private String group;
63.         /**
64.          * Millis of send message timeout.
65.          */
66.         private int sendMessageTimeout = 3000;
68.         /**
69.          * Compress message body threshold, namely, message body larger than 4k will be compressed on default.
70.          */
71.         private int compressMessageBodyThreshold = 1024 * 4;
73.         /**
74.          * Maximum number of retry to perform internally before claiming sending failure in synchronous mode.
75.          * This may potentially cause message duplication which is up to application developers to resolve.
76.          */
77.         private int retryTimesWhenSendFailed = 2;
79.         /**
80.          * <p> Maximum number of retry to perform internally before claiming sending failure in asynchronous mode. </p>
81.          * This may potentially cause message duplication which is up to application developers to resolve.
82.          */
83.         private int retryTimesWhenSendAsyncFailed = 2;
85.         /**
86.          * Indicate whether to retry another broker on sending failure internally.
87.          */
88.         private boolean retryNextServer = false;
90.         /**
91.          * Maximum allowed message size in bytes.
92.          */
93.         private int maxMessageSize = 1024 * 1024 * 4;
95.         /**
96.          * The property of "access-key".
97.          */
98.         private String accessKey;
100.         /**
101.          * The property of "secret-key".
102.          */
103.         private String secretKey;
105.         /**
106.          * Switch flag instance for message trace.
107.          */
108.         private boolean enableMsgTrace = true;
110.         /**
111.          * The name value of message trace topic.If you don't config,you can use the default trace topic name.
112.          */
113.         private String customizedTraceTopic = TopicValidator.RMQ_SYS_TRACE_TOPIC;
115.         public String getGroup() {
116.             return group;
117.         }
119.         public void setGroup(String group) {
120.             this.group = group;
121.         }
123.         public int getSendMessageTimeout() {
124.             return sendMessageTimeout;
125.         }
127.         public void setSendMessageTimeout(int sendMessageTimeout) {
128.             this.sendMessageTimeout = sendMessageTimeout;
129.         }
131.         public int getCompressMessageBodyThreshold() {
132.             return compressMessageBodyThreshold;
133.         }
135.         public void setCompressMessageBodyThreshold(int compressMessageBodyThreshold) {
136.             this.compressMessageBodyThreshold = compressMessageBodyThreshold;
137.         }
139.         public int getRetryTimesWhenSendFailed() {
140.             return retryTimesWhenSendFailed;
141.         }
143.         public void setRetryTimesWhenSendFailed(int retryTimesWhenSendFailed) {
144.             this.retryTimesWhenSendFailed = retryTimesWhenSendFailed;
145.         }
147.         public int getRetryTimesWhenSendAsyncFailed() {
148.             return retryTimesWhenSendAsyncFailed;
149.         }
151.         public void setRetryTimesWhenSendAsyncFailed(int retryTimesWhenSendAsyncFailed) {
152.             this.retryTimesWhenSendAsyncFailed = retryTimesWhenSendAsyncFailed;
153.         }
155.         public boolean isRetryNextServer() {
156.             return retryNextServer;
157.         }
159.         public void setRetryNextServer(boolean retryNextServer) {
160.             this.retryNextServer = retryNextServer;
161.         }
163.         public int getMaxMessageSize() {
164.             return maxMessageSize;
165.         }
167.         public void setMaxMessageSize(int maxMessageSize) {
168.             this.maxMessageSize = maxMessageSize;
169.         }
171.         public String getAccessKey() {
172.             return accessKey;
173.         }
175.         public void setAccessKey(String accessKey) {
176.             this.accessKey = accessKey;
177.         }
179.         public String getSecretKey() {
180.             return secretKey;
181.         }
183.         public void setSecretKey(String secretKey) {
184.             this.secretKey = secretKey;
185.         }
187.         public boolean isEnableMsgTrace() {
188.             return enableMsgTrace;
189.         }
191.         public void setEnableMsgTrace(boolean enableMsgTrace) {
192.             this.enableMsgTrace = enableMsgTrace;
193.         }
195.         public String getCustomizedTraceTopic() {
196.             return customizedTraceTopic;
197.         }
199.         public void setCustomizedTraceTopic(String customizedTraceTopic) {
200.             this.customizedTraceTopic = customizedTraceTopic;
201.         }
202.     }
204.     public org.apache.rocketmq.spring.autoconfigure.RocketMQProperties.Consumer getConsumer() {
205.         return consumer;
206.     }
208.     public void setConsumer(org.apache.rocketmq.spring.autoconfigure.RocketMQProperties.Consumer consumer) {
209.         this.consumer = consumer;
210.     }
212.     public static final class Consumer {
213.         /**
214.          * listener configuration container
215.          * the pattern is like this:
216.          * group1.topic1 = false
217.          * group2.topic2 = true
218.          * group3.topic3 = false
219.          */
220.         private Map<String, Map<String, Boolean>> listeners = new HashMap<>();
222.         public Map<String, Map<String, Boolean>> getListeners() {
223.             return listeners;
224.         }
226.         public void setListeners(Map<String, Map<String, Boolean>> listeners) {
227.             this.listeners = listeners;
228.         }
229.     }
231. }

复制代码

步调二，复制一份@RocketMQMessageListener，并新增数据源参数soruce
，这里不接纳原来的nameServer参数，大概是版本原因，这个参数现在并不支持多数据源

1. /*
2. * Licensed to the Apache Software Foundation (ASF) under one or more
3. * contributor license agreements.  See the NOTICE file distributed with
4. * this work for additional information regarding copyright ownership.
5. * The ASF licenses this file to You under the Apache License, Version 2.0
6. * (the "License"); you may not use this file except in compliance with
7. * the License.  You may obtain a copy of the License at
8. *
9. *     http://www.apache.org/licenses/LICENSE-2.0
10. *
11. * Unless required by applicable law or agreed to in writing, software
12. * distributed under the License is distributed on an "AS IS" BASIS,
13. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14. * See the License for the specific language governing permissions and
15. * limitations under the License.
16. */
18. package com.heit.road.web.config;
20. import org.apache.rocketmq.spring.annotation.ConsumeMode;
21. import org.apache.rocketmq.spring.annotation.MessageModel;
22. import org.apache.rocketmq.spring.annotation.SelectorType;
24. import java.lang.annotation.*;
26. @Target(ElementType.TYPE)
27. @Retention(RetentionPolicy.RUNTIME)
28. @Documented
29. public @interface MultipleRocketMQMessageListener {
31.     String NAME_SERVER_PLACEHOLDER = "";
32.     String ACCESS_KEY_PLACEHOLDER = "";
33.     String SECRET_KEY_PLACEHOLDER = "";
34.     String TRACE_TOPIC_PLACEHOLDER = "";
35.     String ACCESS_CHANNEL_PLACEHOLDER = "";
37.     /**
38.      * Consumers of the same role is required to have exactly same subscriptions and consumerGroup to correctly achieve
39.      * load balance. It's required and needs to be globally unique.
40.      * <p>
41.      * <p>
42.      * See <a target="_blank" href="http://rocketmq.apache.org/docs/core-concept/">here</a> for further discussion.
43.      */
44.     String soruce();
46.     String consumerGroup();
48.     /**
49.      * Topic name.
50.      */
51.     String topic();
53.     /**
54.      * Control how to selector message.
55.      *
56.      * @see SelectorType
57.      */
58.     SelectorType selectorType() default SelectorType.TAG;
60.     /**
61.      * Control which message can be select. Grammar please see {@link SelectorType#TAG} and {@link SelectorType#SQL92}
62.      */
63.     String selectorExpression() default "*";
65.     /**
66.      * Control consume mode, you can choice receive message concurrently or orderly.
67.      */
68.     ConsumeMode consumeMode() default ConsumeMode.CONCURRENTLY;
70.     /**
71.      * Control message mode, if you want all subscribers receive message all message, broadcasting is a good choice.
72.      */
73.     MessageModel messageModel() default MessageModel.CLUSTERING;
75.     /**
76.      * Max consumer thread number.
77.      */
78.     int consumeThreadMax() default 64;
80.     /**
81.      * Maximum amount of time in minutes a message may block the consuming thread.
82.      */
83.     long consumeTimeout() default 15L;
85.     /**
86.      * The property of "access-key".
87.      */
88.     String accessKey() default ACCESS_KEY_PLACEHOLDER;
90.     /**
91.      * The property of "secret-key".
92.      */
93.     String secretKey() default SECRET_KEY_PLACEHOLDER;
95.     /**
96.      * Switch flag instance for message trace.
97.      */
98.     boolean enableMsgTrace() default true;
100.     /**
101.      * The name value of message trace topic.If you don't config,you can use the default trace topic name.
102.      */
103.     String customizedTraceTopic() default TRACE_TOPIC_PLACEHOLDER;
105.     /**
106.      * The property of "name-server".
107.      */
108.     String nameServer() default NAME_SERVER_PLACEHOLDER;
110.     /**
111.      * The property of "access-channel".
112.      */
113.     String accessChannel() default ACCESS_CHANNEL_PLACEHOLDER;
114. }

复制代码

步调三，复制一份RocketMQListenerContainer，替换参数RocketMQMessageListener

1. /*
2. * Licensed to the Apache Software Foundation (ASF) under one or more
3. * contributor license agreements.  See the NOTICE file distributed with
4. * this work for additional information regarding copyright ownership.
5. * The ASF licenses this file to You under the Apache License, Version 2.0
6. * (the "License"); you may not use this file except in compliance with
7. * the License.  You may obtain a copy of the License at
8. *
9. *     http://www.apache.org/licenses/LICENSE-2.0
10. *
11. * Unless required by applicable law or agreed to in writing, software
12. * distributed under the License is distributed on an "AS IS" BASIS,
13. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14. * See the License for the specific language governing permissions and
15. * limitations under the License.
16. */
18. package com.heit.road.web.config;
20. import org.apache.rocketmq.client.AccessChannel;
21. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
22. import org.apache.rocketmq.client.consumer.MessageSelector;
23. import org.apache.rocketmq.client.consumer.listener.*;
24. import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
25. import org.apache.rocketmq.client.exception.MQClientException;
26. import org.apache.rocketmq.client.producer.SendCallback;
27. import org.apache.rocketmq.client.producer.SendResult;
28. import org.apache.rocketmq.client.producer.SendStatus;
29. import org.apache.rocketmq.client.utils.MessageUtil;
30. import org.apache.rocketmq.common.message.MessageExt;
31. import org.apache.rocketmq.remoting.RPCHook;
32. import org.apache.rocketmq.remoting.exception.RemotingException;
33. import org.apache.rocketmq.spring.annotation.ConsumeMode;
34. import org.apache.rocketmq.spring.annotation.MessageModel;
35. import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
36. import org.apache.rocketmq.spring.annotation.SelectorType;
37. import org.apache.rocketmq.spring.core.RocketMQListener;
38. import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
39. import org.apache.rocketmq.spring.core.RocketMQReplyListener;
40. import org.apache.rocketmq.spring.support.RocketMQListenerContainer;
41. import org.apache.rocketmq.spring.support.RocketMQUtil;
42. import org.slf4j.Logger;
43. import org.slf4j.LoggerFactory;
44. import org.springframework.aop.framework.AopProxyUtils;
45. import org.springframework.beans.BeansException;
46. import org.springframework.beans.factory.InitializingBean;
47. import org.springframework.context.ApplicationContext;
48. import org.springframework.context.ApplicationContextAware;
49. import org.springframework.context.SmartLifecycle;
50. import org.springframework.core.MethodParameter;
51. import org.springframework.messaging.Message;
52. import org.springframework.messaging.MessageHeaders;
53. import org.springframework.messaging.converter.MessageConversionException;
54. import org.springframework.messaging.converter.MessageConverter;
55. import org.springframework.messaging.converter.SmartMessageConverter;
56. import org.springframework.messaging.support.MessageBuilder;
57. import org.springframework.util.Assert;
58. import org.springframework.util.MimeTypeUtils;
60. import java.lang.reflect.Method;
61. import java.lang.reflect.ParameterizedType;
62. import java.lang.reflect.Type;
63. import java.nio.charset.Charset;
64. import java.util.List;
65. import java.util.Objects;
67. @SuppressWarnings("WeakerAccess")
68. public class MultipleRocketMQListenerContainer implements InitializingBean,
69.         RocketMQListenerContainer, SmartLifecycle, ApplicationContextAware {
70.     private final static Logger log = LoggerFactory.getLogger(MultipleRocketMQListenerContainer.class);
72.     private ApplicationContext applicationContext;
74.     /**
75.      * The name of the DefaultRocketMQListenerContainer instance
76.      */
77.     private String name;
79.     private long suspendCurrentQueueTimeMillis = 1000;
81.     /**
82.      * Message consume retry strategy<br> -1,no retry,put into DLQ directly<br> 0,broker control retry frequency<br>
83.      * >0,client control retry frequency.
84.      */
85.     private int delayLevelWhenNextConsume = 0;
87.     private String nameServer;
89.     private AccessChannel accessChannel = AccessChannel.LOCAL;
91.     private String consumerGroup;
93.     private String topic;
95.     private int consumeThreadMax = 64;
97.     private String charset = "UTF-8";
99.     private MessageConverter messageConverter;
101.     private RocketMQListener rocketMQListener;
103.     private RocketMQReplyListener rocketMQReplyListener;
105.     private MultipleRocketMQMessageListener rocketMQMessageListener;
107.     private DefaultMQPushConsumer consumer;
109.     private Type messageType;
111.     private MethodParameter methodParameter;
113.     private boolean running;
115.     // The following properties came from @RocketMQMessageListener.
116.     private ConsumeMode consumeMode;
117.     private SelectorType selectorType;
118.     private String selectorExpression;
119.     private MessageModel messageModel;
120.     private long consumeTimeout;
122.     public long getSuspendCurrentQueueTimeMillis() {
123.         return suspendCurrentQueueTimeMillis;
124.     }
126.     public void setSuspendCurrentQueueTimeMillis(long suspendCurrentQueueTimeMillis) {
127.         this.suspendCurrentQueueTimeMillis = suspendCurrentQueueTimeMillis;
128.     }
130.     public int getDelayLevelWhenNextConsume() {
131.         return delayLevelWhenNextConsume;
132.     }
134.     public void setDelayLevelWhenNextConsume(int delayLevelWhenNextConsume) {
135.         this.delayLevelWhenNextConsume = delayLevelWhenNextConsume;
136.     }
138.     public String getNameServer() {
139.         return nameServer;
140.     }
142.     public void setNameServer(String nameServer) {
143.         this.nameServer = nameServer;
144.     }
146.     public AccessChannel getAccessChannel() {
147.         return accessChannel;
148.     }
150.     public void setAccessChannel(AccessChannel accessChannel) {
151.         this.accessChannel = accessChannel;
152.     }
154.     public String getConsumerGroup() {
155.         return consumerGroup;
156.     }
158.     public void setConsumerGroup(String consumerGroup) {
159.         this.consumerGroup = consumerGroup;
160.     }
162.     public String getTopic() {
163.         return topic;
164.     }
166.     public void setTopic(String topic) {
167.         this.topic = topic;
168.     }
170.     public int getConsumeThreadMax() {
171.         return consumeThreadMax;
172.     }
174.     public String getCharset() {
175.         return charset;
176.     }
178.     public void setCharset(String charset) {
179.         this.charset = charset;
180.     }
182.     public MessageConverter getMessageConverter() {
183.         return messageConverter;
184.     }
186.     public MultipleRocketMQListenerContainer setMessageConverter(MessageConverter messageConverter) {
187.         this.messageConverter = messageConverter;
188.         return this;
189.     }
191.     public RocketMQListener getRocketMQListener() {
192.         return rocketMQListener;
193.     }
195.     public void setRocketMQListener(RocketMQListener rocketMQListener) {
196.         this.rocketMQListener = rocketMQListener;
197.     }
199.     public RocketMQReplyListener getRocketMQReplyListener() {
200.         return rocketMQReplyListener;
201.     }
203.     public void setRocketMQReplyListener(RocketMQReplyListener rocketMQReplyListener) {
204.         this.rocketMQReplyListener = rocketMQReplyListener;
205.     }
207.     public MultipleRocketMQMessageListener getRocketMQMessageListener() {
208.         return rocketMQMessageListener;
209.     }
211.     public void setRocketMQMessageListener(MultipleRocketMQMessageListener anno) {
212.         this.rocketMQMessageListener = anno;
214.         this.consumeMode = anno.consumeMode();
215.         this.consumeThreadMax = anno.consumeThreadMax();
216.         this.messageModel = anno.messageModel();
217.         this.selectorType = anno.selectorType();
218.         this.selectorExpression = anno.selectorExpression();
219.         this.consumeTimeout = anno.consumeTimeout();
220.     }
222.     public ConsumeMode getConsumeMode() {
223.         return consumeMode;
224.     }
226.     public SelectorType getSelectorType() {
227.         return selectorType;
228.     }
230.     public void setSelectorExpression(String selectorExpression) {
231.         this.selectorExpression = selectorExpression;
232.     }
234.     public String getSelectorExpression() {
235.         return selectorExpression;
236.     }
238.     public MessageModel getMessageModel() {
239.         return messageModel;
240.     }
242.     public DefaultMQPushConsumer getConsumer() {
243.         return consumer;
244.     }
246.     public void setConsumer(DefaultMQPushConsumer consumer) {
247.         this.consumer = consumer;
248.     }
250.     @Override
251.     public void destroy() {
252.         this.setRunning(false);
253.         if (Objects.nonNull(consumer)) {
254.             consumer.shutdown();
255.         }
256.         log.info("container destroyed, {}", this.toString());
257.     }
259.     @Override
260.     public boolean isAutoStartup() {
261.         return true;
262.     }
264.     @Override
265.     public void stop(Runnable callback) {
266.         stop();
267.         callback.run();
268.     }
270.     @Override
271.     public void start() {
272.         if (this.isRunning()) {
273.             throw new IllegalStateException("container already running. " + this.toString());
274.         }
276.         try {
277.             consumer.start();
278.         } catch (MQClientException e) {
279.             throw new IllegalStateException("Failed to start RocketMQ push consumer", e);
280.         }
281.         this.setRunning(true);
283.         log.info("running container: {}", this.toString());
284.     }
286.     @Override
287.     public void stop() {
288.         if (this.isRunning()) {
289.             if (Objects.nonNull(consumer)) {
290.                 consumer.shutdown();
291.             }
292.             setRunning(false);
293.         }
294.     }
296.     @Override
297.     public boolean isRunning() {
298.         return running;
299.     }
301.     private void setRunning(boolean running) {
302.         this.running = running;
303.     }
305.     @Override
306.     public int getPhase() {
307.         // Returning Integer.MAX_VALUE only suggests that
308.         // we will be the first bean to shutdown and last bean to start
309.         return Integer.MAX_VALUE;
310.     }
312.     @Override
313.     public void afterPropertiesSet() throws Exception {
314.         initRocketMQPushConsumer();
316.         this.messageType = getMessageType();
317.         this.methodParameter = getMethodParameter();
318.         log.debug("RocketMQ messageType: {}", messageType);
319.     }
321.     @Override
322.     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
323.         this.applicationContext = applicationContext;
324.     }
326.     @Override
327.     public String toString() {
328.         return "DefaultRocketMQListenerContainer{" +
329.             "consumerGroup='" + consumerGroup + '\'' +
330.             ", nameServer='" + nameServer + '\'' +
331.             ", topic='" + topic + '\'' +
332.             ", consumeMode=" + consumeMode +
333.             ", selectorType=" + selectorType +
334.             ", selectorExpression='" + selectorExpression + '\'' +
335.             ", messageModel=" + messageModel +
336.             '}';
337.     }
339.     public void setName(String name) {
340.         this.name = name;
341.     }
343.     public class DefaultMessageListenerConcurrently implements MessageListenerConcurrently {
345.         @SuppressWarnings("unchecked")
346.         @Override
347.         public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
348.             for (MessageExt messageExt : msgs) {
349.                 log.debug("received msg: {}", messageExt);
350.                 try {
351.                     long now = System.currentTimeMillis();
352.                     handleMessage(messageExt);
353.                     long costTime = System.currentTimeMillis() - now;
354.                     log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
355.                 } catch (Exception e) {
356.                     log.warn("consume message failed. messageExt:{}, error:{}", messageExt, e);
357.                     context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);
358.                     return ConsumeConcurrentlyStatus.RECONSUME_LATER;
359.                 }
360.             }
362.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
363.         }
364.     }
366.     public class DefaultMessageListenerOrderly implements MessageListenerOrderly {
368.         @SuppressWarnings("unchecked")
369.         @Override
370.         public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
371.             for (MessageExt messageExt : msgs) {
372.                 log.debug("received msg: {}", messageExt);
373.                 try {
374.                     long now = System.currentTimeMillis();
375.                     handleMessage(messageExt);
376.                     long costTime = System.currentTimeMillis() - now;
377.                     log.info("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
378.                 } catch (Exception e) {
379.                     log.warn("consume message failed. messageExt:{}", messageExt, e);
380.                     context.setSuspendCurrentQueueTimeMillis(suspendCurrentQueueTimeMillis);
381.                     return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
382.                 }
383.             }
385.             return ConsumeOrderlyStatus.SUCCESS;
386.         }
387.     }
389.     private void handleMessage(
390.         MessageExt messageExt) throws MQClientException, RemotingException, InterruptedException {
391.         if (rocketMQListener != null) {
392.             rocketMQListener.onMessage(doConvertMessage(messageExt));
393.         } else if (rocketMQReplyListener != null) {
394.             Object replyContent = rocketMQReplyListener.onMessage(doConvertMessage(messageExt));
395.             Message<?> message = MessageBuilder.withPayload(replyContent).build();
397.             org.apache.rocketmq.common.message.Message replyMessage = MessageUtil.createReplyMessage(messageExt, convertToBytes(message));
398.             consumer.getDefaultMQPushConsumerImpl().getmQClientFactory().getDefaultMQProducer().send(replyMessage, new SendCallback() {
399.                 @Override public void onSuccess(SendResult sendResult) {
400.                     if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
401.                         log.error("Consumer replies message failed. SendStatus: {}", sendResult.getSendStatus());
402.                     } else {
403.                         log.info("Consumer replies message success.");
404.                     }
405.                 }
407.                 @Override public void onException(Throwable e) {
408.                     log.error("Consumer replies message failed. error: {}", e.getLocalizedMessage());
409.                 }
410.             });
411.         }
412.     }
414.     private byte[] convertToBytes(Message<?> message) {
415.         Message<?> messageWithSerializedPayload = doConvert(message.getPayload(), message.getHeaders());
416.         Object payloadObj = messageWithSerializedPayload.getPayload();
417.         byte[] payloads;
418.         try {
419.             if (null == payloadObj) {
420.                 throw new RuntimeException("the message cannot be empty");
421.             }
422.             if (payloadObj instanceof String) {
423.                 payloads = ((String) payloadObj).getBytes(Charset.forName(charset));
424.             } else if (payloadObj instanceof byte[]) {
425.                 payloads = (byte[]) messageWithSerializedPayload.getPayload();
426.             } else {
427.                 String jsonObj = (String) this.messageConverter.fromMessage(messageWithSerializedPayload, payloadObj.getClass());
428.                 if (null == jsonObj) {
429.                     throw new RuntimeException(String.format(
430.                         "empty after conversion [messageConverter:%s,payloadClass:%s,payloadObj:%s]",
431.                         this.messageConverter.getClass(), payloadObj.getClass(), payloadObj));
432.                 }
433.                 payloads = jsonObj.getBytes(Charset.forName(charset));
434.             }
435.         } catch (Exception e) {
436.             throw new RuntimeException("convert to bytes failed.", e);
437.         }
438.         return payloads;
439.     }
441.     private Message<?> doConvert(Object payload, MessageHeaders headers) {
442.         Message<?> message = this.messageConverter instanceof SmartMessageConverter ?
443.             ((SmartMessageConverter) this.messageConverter).toMessage(payload, headers, null) :
444.             this.messageConverter.toMessage(payload, headers);
445.         if (message == null) {
446.             String payloadType = payload.getClass().getName();
447.             Object contentType = headers != null ? headers.get(MessageHeaders.CONTENT_TYPE) : null;
448.             throw new MessageConversionException("Unable to convert payload with type='" + payloadType +
449.                 "', contentType='" + contentType + "', converter=[" + this.messageConverter + "]");
450.         }
451.         MessageBuilder<?> builder = MessageBuilder.fromMessage(message);
452.         builder.setHeaderIfAbsent(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN);
453.         return builder.build();
454.     }
456.     @SuppressWarnings("unchecked")
457.     private Object doConvertMessage(MessageExt messageExt) {
458.         if (Objects.equals(messageType, MessageExt.class)) {
459.             return messageExt;
460.         } else {
461.             String str = new String(messageExt.getBody(), Charset.forName(charset));
462.             if (Objects.equals(messageType, String.class)) {
463.                 return str;
464.             } else {
465.                 // If msgType not string, use objectMapper change it.
466.                 try {
467.                     if (messageType instanceof Class) {
468.                         //if the messageType has not Generic Parameter
469.                         return this.getMessageConverter().fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) messageType);
470.                     } else {
471.                         //if the messageType has Generic Parameter, then use SmartMessageConverter#fromMessage with third parameter "conversionHint".
472.                         //we have validate the MessageConverter is SmartMessageConverter in this#getMethodParameter.
473.                         return ((SmartMessageConverter) this.getMessageConverter()).fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) ((ParameterizedType) messageType).getRawType(), methodParameter);
474.                     }
475.                 } catch (Exception e) {
476.                     log.info("convert failed. str:{}, msgType:{}", str, messageType);
477.                     throw new RuntimeException("cannot convert message to " + messageType, e);
478.                 }
479.             }
480.         }
481.     }
483.     private MethodParameter getMethodParameter() {
484.         Class<?> targetClass;
485.         if (rocketMQListener != null) {
486.             targetClass = AopProxyUtils.ultimateTargetClass(rocketMQListener);
487.         } else {
488.             targetClass = AopProxyUtils.ultimateTargetClass(rocketMQReplyListener);
489.         }
490.         Type messageType = this.getMessageType();
491.         Class clazz = null;
492.         if (messageType instanceof ParameterizedType && messageConverter instanceof SmartMessageConverter) {
493.             clazz = (Class) ((ParameterizedType) messageType).getRawType();
494.         } else if (messageType instanceof Class) {
495.             clazz = (Class) messageType;
496.         } else {
497.             throw new RuntimeException("parameterType:" + messageType + " of onMessage method is not supported");
498.         }
499.         try {
500.             final Method method = targetClass.getMethod("onMessage", clazz);
501.             return new MethodParameter(method, 0);
502.         } catch (NoSuchMethodException e) {
503.             e.printStackTrace();
504.             throw new RuntimeException("parameterType:" + messageType + " of onMessage method is not supported");
505.         }
506.     }
508.     private Type getMessageType() {
509.         Class<?> targetClass;
510.         if (rocketMQListener != null) {
511.             targetClass = AopProxyUtils.ultimateTargetClass(rocketMQListener);
512.         } else {
513.             targetClass = AopProxyUtils.ultimateTargetClass(rocketMQReplyListener);
514.         }
515.         Type matchedGenericInterface = null;
516.         while (Objects.nonNull(targetClass)) {
517.             Type[] interfaces = targetClass.getGenericInterfaces();
518.             if (Objects.nonNull(interfaces)) {
519.                 for (Type type : interfaces) {
520.                     if (type instanceof ParameterizedType &&
521.                         (Objects.equals(((ParameterizedType) type).getRawType(), RocketMQListener.class) || Objects.equals(((ParameterizedType) type).getRawType(), RocketMQReplyListener.class))) {
522.                         matchedGenericInterface = type;
523.                         break;
524.                     }
525.                 }
526.             }
527.             targetClass = targetClass.getSuperclass();
528.         }
529.         if (Objects.isNull(matchedGenericInterface)) {
530.             return Object.class;
531.         }
533.         Type[] actualTypeArguments = ((ParameterizedType) matchedGenericInterface).getActualTypeArguments();
534.         if (Objects.nonNull(actualTypeArguments) && actualTypeArguments.length > 0) {
535.             return actualTypeArguments[0];
536.         }
537.         return Object.class;
538.     }
540.     private void initRocketMQPushConsumer() throws MQClientException {
541.         if (rocketMQListener == null && rocketMQReplyListener == null) {
542.             throw new IllegalArgumentException("Property 'rocketMQListener' or 'rocketMQReplyListener' is required");
543.         }
544.         Assert.notNull(consumerGroup, "Property 'consumerGroup' is required");
545.         Assert.notNull(nameServer, "Property 'nameServer' is required");
546.         Assert.notNull(topic, "Property 'topic' is required");
548.         RPCHook rpcHook = RocketMQUtil.getRPCHookByAkSk(applicationContext.getEnvironment(),
549.             this.rocketMQMessageListener.accessKey(), this.rocketMQMessageListener.secretKey());
550.         boolean enableMsgTrace = rocketMQMessageListener.enableMsgTrace();
551.         if (Objects.nonNull(rpcHook)) {
552.             consumer = new DefaultMQPushConsumer(consumerGroup, rpcHook, new AllocateMessageQueueAveragely(),
553.                 enableMsgTrace, this.applicationContext.getEnvironment().
554.                 resolveRequiredPlaceholders(this.rocketMQMessageListener.customizedTraceTopic()));
555.             consumer.setVipChannelEnabled(false);
556.             consumer.setInstanceName(RocketMQUtil.getInstanceName(rpcHook, consumerGroup));
557.         } else {
558.             log.debug("Access-key or secret-key not configure in " + this + ".");
559.             consumer = new DefaultMQPushConsumer(consumerGroup, enableMsgTrace,
560.                 this.applicationContext.getEnvironment().
561.                     resolveRequiredPlaceholders(this.rocketMQMessageListener.customizedTraceTopic()));
562.         }
564.         String customizedNameServer = this.applicationContext.getEnvironment().resolveRequiredPlaceholders(this.rocketMQMessageListener.nameServer());
565.         if (customizedNameServer != null) {
566.             consumer.setNamesrvAddr(customizedNameServer);
567.         } else {
568.             consumer.setNamesrvAddr(nameServer);
569.         }
570.         if (accessChannel != null) {
571.             consumer.setAccessChannel(accessChannel);
572.         }
573.         consumer.setConsumeThreadMax(consumeThreadMax);
574.         if (consumeThreadMax < consumer.getConsumeThreadMin()) {
575.             consumer.setConsumeThreadMin(consumeThreadMax);
576.         }
577.         consumer.setConsumeTimeout(consumeTimeout);
579.         switch (messageModel) {
580.             case BROADCASTING:
581.                 consumer.setMessageModel(org.apache.rocketmq.common.protocol.heartbeat.MessageModel.BROADCASTING);
582.                 break;
583.             case CLUSTERING:
584.                 consumer.setMessageModel(org.apache.rocketmq.common.protocol.heartbeat.MessageModel.CLUSTERING);
585.                 break;
586.             default:
587.                 throw new IllegalArgumentException("Property 'messageModel' was wrong.");
588.         }
590.         switch (selectorType) {
591.             case TAG:
592.                 consumer.subscribe(topic, selectorExpression);
593.                 break;
594.             case SQL92:
595.                 consumer.subscribe(topic, MessageSelector.bySql(selectorExpression));
596.                 break;
597.             default:
598.                 throw new IllegalArgumentException("Property 'selectorType' was wrong.");
599.         }
601.         switch (consumeMode) {
602.             case ORDERLY:
603.                 consumer.setMessageListener(new DefaultMessageListenerOrderly());
604.                 break;
605.             case CONCURRENTLY:
606.                 consumer.setMessageListener(new DefaultMessageListenerConcurrently());
607.                 break;
608.             default:
609.                 throw new IllegalArgumentException("Property 'consumeMode' was wrong.");
610.         }
612.         if (rocketMQListener instanceof RocketMQPushConsumerLifecycleListener) {
613.             ((RocketMQPushConsumerLifecycleListener) rocketMQListener).prepareStart(consumer);
614.         } else if (rocketMQReplyListener instanceof RocketMQPushConsumerLifecycleListener) {
615.             ((RocketMQPushConsumerLifecycleListener) rocketMQReplyListener).prepareStart(consumer);
616.         }
618.     }
620. }

复制代码

步调四，配置文件添加参数,与原rocketmq参数区分开

1. multiplerocketmq:
2.   xdt:
3.     name-server: xdtIP:xdtPort
4.     producer:
5.       group: xdt_groups
6.   road:
7.     name-server: roadIP:roadPort
8.     producer:
9.       group: road_groups

复制代码

步调五，通过配置文件加载多数据源

1. package com.heit.road.web.config;
3. import cn.hutool.core.date.DateUtil;
4. import org.apache.rocketmq.client.AccessChannel;
5. import org.apache.rocketmq.client.producer.DefaultMQProducer;
6. import org.apache.rocketmq.spring.annotation.ConsumeMode;
7. import org.apache.rocketmq.spring.annotation.MessageModel;
8. import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
9. import org.apache.rocketmq.spring.core.RocketMQListener;
10. import org.apache.rocketmq.spring.core.RocketMQReplyListener;
11. import org.apache.rocketmq.spring.core.RocketMQTemplate;
12. import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
13. import org.apache.rocketmq.spring.support.RocketMQUtil;
14. import org.slf4j.Logger;
15. import org.slf4j.LoggerFactory;
16. import org.springframework.aop.framework.AopProxyUtils;
17. import org.springframework.aop.scope.ScopedProxyUtils;
18. import org.springframework.beans.BeansException;
19. import org.springframework.beans.factory.SmartInitializingSingleton;
20. import org.springframework.beans.factory.annotation.Qualifier;
21. import org.springframework.beans.factory.support.BeanDefinitionValidationException;
22. import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
23. import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
24. import org.springframework.boot.context.properties.ConfigurationProperties;
25. import org.springframework.context.ApplicationContext;
26. import org.springframework.context.ApplicationContextAware;
27. import org.springframework.context.ConfigurableApplicationContext;
28. import org.springframework.context.annotation.Bean;
29. import org.springframework.context.annotation.Configuration;
30. import org.springframework.context.support.GenericApplicationContext;
31. import org.springframework.core.env.StandardEnvironment;
32. import org.springframework.util.Assert;
33. import org.springframework.util.StringUtils;
35. import java.util.Collections;
36. import java.util.Map;
37. import java.util.concurrent.atomic.AtomicLong;
38. import java.util.stream.Collectors;
40. @Configuration
41. public class MultipleRocketMQConfig implements ApplicationContextAware, SmartInitializingSingleton {
42.     private final static Logger log = LoggerFactory.getLogger(MultipleRocketMQConfig.class);
43.     private ConfigurableApplicationContext applicationContext;
44.     private AtomicLong counter = new AtomicLong(0);
45.     private StandardEnvironment environment;
46.     private RocketMQMessageConverter rocketMQMessageConverter;
48.     public MultipleRocketMQConfig(RocketMQMessageConverter rocketMQMessageConverter,
49.                                   StandardEnvironment environment) {
50.         this.rocketMQMessageConverter = rocketMQMessageConverter;
51.         this.environment = environment;
52.     }
54.     @Bean("road")
55.     @ConditionalOnProperty(prefix = "multiplerocketmq.road", value = {"name-server"})
56.     @ConfigurationProperties(prefix = "multiplerocketmq.road")
57.     public MultipleRocketMQProperties road() {
58.         return new MultipleRocketMQProperties();
59.     }
61.     @Bean(value = "roadmq", destroyMethod = "destroy")
62.     @ConditionalOnMissingBean(name = "roadmq")
63.     @ConditionalOnProperty(prefix = "multiplerocketmq.road", value = {"name-server", "producer.group"})
64.     public RocketMQTemplate roadMQProducer(@Qualifier("road") MultipleRocketMQProperties rocketMQProperties, RocketMQMessageConverter rocketMQMessageConverter) {
65.         return createRocketMQTemplate(rocketMQProperties, rocketMQMessageConverter);
66.     }
68.     @Bean("xdt")
69.     @ConditionalOnProperty(prefix = "multiplerocketmq.xdt", value = {"name-server"})
70.     @ConfigurationProperties(prefix = "multiplerocketmq.xdt")
71.     public MultipleRocketMQProperties xdt() {
72.         return new MultipleRocketMQProperties();
73.     }
75.     @Bean(value = "xdtmq", destroyMethod = "destroy")
76.     @ConditionalOnMissingBean(name = "xdtmq")
77.     @ConditionalOnProperty(prefix = "multiplerocketmq.xdt", value = {"name-server", "producer.group"})
78.     public RocketMQTemplate xdtMQProducer(@Qualifier("xdt") MultipleRocketMQProperties rocketMQProperties, RocketMQMessageConverter rocketMQMessageConverter) {
79.         return createRocketMQTemplate(rocketMQProperties, rocketMQMessageConverter);
80.     }
82.     private RocketMQTemplate createRocketMQTemplate(MultipleRocketMQProperties rocketMQProperties, RocketMQMessageConverter rocketMQMessageConverter) {
83.         RocketMQProperties.Producer producerConfig = rocketMQProperties.getProducer();
84.         String nameServer = rocketMQProperties.getNameServer();
85.         String groupName = producerConfig.getGroup();
86.         Assert.hasText(nameServer, "[rocketmq.name-server] must not be null");
87.         Assert.hasText(groupName, "[rocketmq.producer.group] must not be null");
89.         String accessChannel = rocketMQProperties.getAccessChannel();
91.         String ak = rocketMQProperties.getProducer().getAccessKey();
92.         String sk = rocketMQProperties.getProducer().getSecretKey();
93.         boolean isEnableMsgTrace = rocketMQProperties.getProducer().isEnableMsgTrace();
94.         String customizedTraceTopic = rocketMQProperties.getProducer().getCustomizedTraceTopic();
95.         DefaultMQProducer producer = RocketMQUtil.createDefaultMQProducer(groupName, ak, sk, isEnableMsgTrace, customizedTraceTopic);
97.         producer.setNamesrvAddr(nameServer);
98.         if (!StringUtils.isEmpty(accessChannel)) {
99.             producer.setAccessChannel(AccessChannel.valueOf(accessChannel));
100.         }
101.         producer.setSendMsgTimeout(producerConfig.getSendMessageTimeout());
102.         producer.setRetryTimesWhenSendFailed(producerConfig.getRetryTimesWhenSendFailed());
103.         producer.setRetryTimesWhenSendAsyncFailed(producerConfig.getRetryTimesWhenSendAsyncFailed());
104.         producer.setMaxMessageSize(producerConfig.getMaxMessageSize());
105.         producer.setCompressMsgBodyOverHowmuch(producerConfig.getCompressMessageBodyThreshold());
106.         producer.setRetryAnotherBrokerWhenNotStoreOK(producerConfig.isRetryNextServer());
107.         producer.setInstanceName(producer.getProducerGroup() + DateUtil.now());
108.         RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
109.         rocketMQTemplate.setProducer(producer);
110.         rocketMQTemplate.setMessageConverter(rocketMQMessageConverter.getMessageConverter());
111.         return rocketMQTemplate;
112.     }
114.     @Override
115.     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
116.         this.applicationContext = (ConfigurableApplicationContext) applicationContext;
117.     }
119.     @Override
120.     public void afterSingletonsInstantiated() {
121.         Map<String, Object> beans = this.applicationContext.getBeansWithAnnotation(MultipleRocketMQMessageListener.class)
122.                 .entrySet().stream().filter(entry -> !ScopedProxyUtils.isScopedTarget(entry.getKey()))
123.                 .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
125.         beans.forEach(this::registerContainer);
126.     }
128.     private void registerContainer(String beanName, Object bean) {
129.         Class<?> clazz = AopProxyUtils.ultimateTargetClass(bean);
131.         if (RocketMQListener.class.isAssignableFrom(bean.getClass()) && RocketMQReplyListener.class.isAssignableFrom(bean.getClass())) {
132.             throw new IllegalStateException(clazz + " cannot be both instance of " + RocketMQListener.class.getName() + " and " + RocketMQReplyListener.class.getName());
133.         }
135.         if (!RocketMQListener.class.isAssignableFrom(bean.getClass()) && !RocketMQReplyListener.class.isAssignableFrom(bean.getClass())) {
136.             throw new IllegalStateException(clazz + " is not instance of " + RocketMQListener.class.getName() + " or " + RocketMQReplyListener.class.getName());
137.         }
139.         MultipleRocketMQMessageListener annotation = clazz.getAnnotation(MultipleRocketMQMessageListener.class);
140.         String topic = annotation.topic();
141.         String consumerGroup = annotation.consumerGroup();
142.         String soruce = annotation.soruce();
143.         MultipleRocketMQProperties rocketMQProperties = this.applicationContext.getBean(soruce, MultipleRocketMQProperties.class);
144.         boolean listenerEnabled =
145.                 (boolean) rocketMQProperties.getConsumer().getListeners().getOrDefault(consumerGroup, Collections.EMPTY_MAP)
146.                         .getOrDefault(topic, true);
148.         if (!listenerEnabled) {
149.             log.debug(
150.                     "Consumer Listener (group:{},topic:{}) is not enabled by configuration, will ignore initialization.",
151.                     consumerGroup, topic);
152.             return;
153.         }
154.         validate(annotation);
156.         String containerBeanName = String.format("%s_%s_%s", topic, consumerGroup,
157.                 counter.incrementAndGet());
158.         GenericApplicationContext genericApplicationContext = (GenericApplicationContext) applicationContext;
160.         genericApplicationContext.registerBean(containerBeanName, MultipleRocketMQListenerContainer.class,
161.                 () -> createRocketMQListenerContainer(containerBeanName, bean, annotation));
162.         MultipleRocketMQListenerContainer container = genericApplicationContext.getBean(containerBeanName,
163.                 MultipleRocketMQListenerContainer.class);
164.         if (!container.isRunning()) {
165.             try {
166.                 container.getConsumer().setInstanceName(containerBeanName);
167.                 container.getConsumer().setNamesrvAddr(rocketMQProperties.getNameServer());
168.                 container.start();
169.             } catch (Exception e) {
170.                 throw new RuntimeException(e);
171.             }
172.         }
173.     }
175.     private MultipleRocketMQListenerContainer createRocketMQListenerContainer(String name, Object bean,
176.                                                                               MultipleRocketMQMessageListener annotation) {
177.         MultipleRocketMQListenerContainer container = new MultipleRocketMQListenerContainer();
179.         container.setRocketMQMessageListener(annotation);
180.         String soruce = annotation.soruce();
181.         if (StringUtils.isEmpty(soruce)) {
182.             throw new RuntimeException(name + " 未指定数据源");
183.         }
184.         MultipleRocketMQProperties rocketMQProperties = this.applicationContext.getBean(soruce, MultipleRocketMQProperties.class);
185.         container.setAccessChannel(AccessChannel.CLOUD);
186.         container.setTopic(annotation.topic());
187.         container.setNameServer(rocketMQProperties.getNameServer());
188.         String tags = annotation.selectorExpression();
189.         if (!StringUtils.isEmpty(tags)) {
190.             container.setSelectorExpression(tags);
191.         }
192.         container.setConsumerGroup(annotation.consumerGroup());
193.         if (RocketMQListener.class.isAssignableFrom(bean.getClass())) {
194.             container.setRocketMQListener((RocketMQListener) bean);
195.         } else if (RocketMQReplyListener.class.isAssignableFrom(bean.getClass())) {
196.             container.setRocketMQReplyListener((RocketMQReplyListener) bean);
197.         }
198.         container.setMessageConverter(rocketMQMessageConverter.getMessageConverter());
199.         return container;
200.     }
202.     private void validate(MultipleRocketMQMessageListener annotation) {
203.         if (annotation.consumeMode() == ConsumeMode.ORDERLY &&
204.                 annotation.messageModel() == MessageModel.BROADCASTING) {
205.             throw new BeanDefinitionValidationException(
206.                     "Bad annotation definition in @RocketMQMessageListener, messageModel BROADCASTING does not support ORDERLY message!");
207.         }
208.     }
210. }

复制代码

步调六，准备多数据源listener

1. package com.heit.road.web.listener;
3. import com.heit.road.web.config.MultipleRocketMQMessageListener;
4. import org.apache.rocketmq.spring.core.RocketMQListener;
5. import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
6. import org.springframework.stereotype.Component;
8. @Component
9. @ConditionalOnProperty(prefix = "multiplerocketmq.road", value = {"name-server"})
10. @MultipleRocketMQMessageListener(soruce = "road", consumerGroup = "RoadRocketMqlistener6", topic = "road_test")
11. public class RoadRocketMqlistener implements RocketMQListener<String> {
12.     @Override
13.     public void onMessage(String message) {
14.         System.out.println("RoadRocketMqlistener 收到信息==》 " + message);
15.     }
16. }

复制代码

1. package com.heit.road.web.listener;
3. import com.heit.road.web.config.MultipleRocketMQMessageListener;
4. import org.apache.rocketmq.spring.core.RocketMQListener;
5. import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
6. import org.springframework.stereotype.Component;
8. @Component
9. @ConditionalOnProperty(prefix = "multiplerocketmq.xdt", value = {"name-server"})
10. @MultipleRocketMQMessageListener(soruce = "xdt", consumerGroup = "XdtRocketMqlistener6", topic = "road_test")
11. public class XdtRocketMqlistener implements RocketMQListener<String> {
12.     @Override
13.     public void onMessage(String message) {
14.         System.out.println("XdtRocketMqlistener 收到信息==》 " + message);
15.     }
16. }

复制代码

步调七，编写测试方法，多数据源发送数据

1. //    @Qualifier("xdtmq")
2.     @Lazy
3.     @Resource
4.     private RocketMQTemplate xdtmq;
6. //    @Qualifier("roadmq")
7.     @Lazy
8.     @Resource
9.     private RocketMQTemplate roadmq;
11.     @ApiOperation("测试用表-MQ信息测试")
12.     @PostMapping("sendTest")
13.     public BaseBack<?> sendTest() {
14.         xdtmq.asyncSend("road_test", "xdtmq123", new SendCallback() {
15.             @Override
16.             public void onSuccess(SendResult sendResult) {
17.                 System.out.println("尝试xdt================================================");
18.                 System.out.println("信息发送成功");
19.                 System.out.println("sendResult = " + sendResult);
20.             }
22.             @Override
23.             public void onException(Throwable e) {
24.                 System.out.println("尝试xdt================================================");
25.                 System.out.println("信息发送失败");
26.                 e.printStackTrace();
27.             }
28.         });
30.         roadmq.asyncSend("road_test", "roadmq123", new SendCallback() {
32.             @Override
33.             public void onSuccess(SendResult sendResult) {
34.                 System.out.println("尝试road================================================");
35.                 System.out.println("信息发送成功");
36.                 System.out.println("sendResult = " + sendResult);
37.             }
39.             @Override
40.             public void onException(Throwable e) {
41.                 System.out.println("尝试road================================================");
42.                 System.out.println("信息发送失败");
43.                 e.printStackTrace();
44.             }
45.         });
46.         return null;
47.     }

复制代码

效果如下：

 

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。