imp
try {
long now = System.currentTimeMillis();
org.apache.rocketmq.common.message.Message rocketMsg = convertToRocketMsg(destination, message);
- SendResult sendResult = producer.send(rocketMsg, messageQueueSelector, hashKey, timeout);
- long costTime = System.currentTimeMillis() - now;
- log.debug("send message cost: {} ms, msgId:{}", costTime, sendResult.getMsgId());
- return sendResult;
+ //TODO
+ throw new RuntimeException("暂时未整合阿里云Producer，不要使用");
+// SendResult sendResult = producer.send(rocketMsg, messageQueueSelector, hashKey, timeout);
+// long costTime = System.currentTimeMillis() - now;
+// log.debug("send message cost: {} ms, msgId:{}", costTime, sendResult.getMsgId());
+// return sendResult;
} catch (Exception e) {
log.info("syncSendOrderly failed. destination:{}, message:{} ", destination, message);
throw new MessagingException(e.getMessage(), e);
}
- }
+ }*/
- /**
- * Same to {@link #syncSend(String, Object)} with send orderly with hashKey by specified.
- *
- * @param destination formats: `topicName:tags`
- * @param payload the Object to use as payload
- * @param hashKey use this key to select queue. for example: orderId, productId ...
- * @return {@link SendResult}
- */
- public SendResult syncSendOrderly(String destination, Object payload, String hashKey) {
- return syncSendOrderly(destination, payload, hashKey, producer.getSendMsgTimeout());
- }
/**
- * Same to {@link #syncSendOrderly(String, Object, String)} with send timeout specified in addition.
+ * Same to {@link #syncSend(String, Object)} with send orderly with hashKey by specified.
*
* @param destination formats: `topicName:tags`
* @param payload the Object to use as payload
* @param hashKey use this key to select queue. for example: orderId, productId ...
- * @param timeout send timeout with millis
* @return {@link SendResult}
*/
- public SendResult syncSendOrderly(String destination, Object payload, String hashKey, long timeout) {
- Message message = this.doConvert(payload, null, null);
- return syncSendOrderly(destination, message, hashKey, producer.getSendMsgTimeout());
- }
+// public SendResult syncSendOrderly(String destination, Object payload, String hashKey) {
+// Message message = this.doConvert(payload, null, null);
+// return syncSendOrderly(destination, message, hashKey);
+// }
/**
- * Same to {@link #asyncSend(String, Message, SendCallback)} with send timeout specified in addition.
- *
- * @param destination formats: `topicName:tags`
- * @param message {@link org.springframework.messaging.Message}
- * @param sendCallback {@link SendCallback}
- * @param timeout send timeout with millis
+ * 将公共的sendCallBack转换为阿里云的sendCallBack
+ * @param sendCallback
+ * @return
+ * 2018年3月23日 zhaowg
*/
- public void asyncSend(String destination, Message message, SendCallback sendCallback, long timeout) {
- if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
- log.info("asyncSend failed. destination:{}, message is null ", destination);
- throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
- }
-
- try {
- org.apache.rocketmq.common.message.Message rocketMsg = convertToRocketMsg(destination, message);
- producer.send(rocketMsg, sendCallback, timeout);
- } catch (Exception e) {
- log.info("asyncSend failed. destination:{}, message:{} ", destination, message);
- throw new MessagingException(e.getMessage(), e);
- }
- }
-
+ private com.aliyun.openservices.ons.api.SendCallback aliyunSendCallBackConvert(final SendCallback sendCallback) {
+ com.aliyun.openservices.ons.api.SendCallback aliyunSendCallBack = new com.aliyun.openservices.ons.api.SendCallback() {
+
+ @Override
+ public void onSuccess(com.aliyun.openservices.ons.api.SendResult sendResult) {
+ sendCallback.onSuccess(convertAliyunSendResult(sendResult));
+ }
+
+ @Override
+ public void onException(OnExceptionContext context) {
+ sendCallback.onException(context.getException());
+ }
+ };
+ return aliyunSendCallBack;
+ }
/**
* Send message to broker asynchronously. asynchronous transmission is generally used in response time sensitive
* business scenarios.
@@ -241,97 +214,77 @@ public class RocketMQTemplate extends AbstractMessageSendingTemplate imp
* @param sendCallback {@link SendCallback}
*/
public void asyncSend(String destination, Message message, SendCallback sendCallback) {
- asyncSend(destination, message, sendCallback, producer.getSendMsgTimeout());
- }
-
- /**
- * Same to {@link #asyncSend(String, Object, SendCallback)} with send timeout specified in addition.
- *
- * @param destination formats: `topicName:tags`
- * @param payload the Object to use as payload
- * @param sendCallback {@link SendCallback}
- * @param timeout send timeout with millis
- */
- public void asyncSend(String destination, Object payload, SendCallback sendCallback, long timeout) {
- Message message = this.doConvert(payload, null, null);
- asyncSend(destination, message, sendCallback, timeout);
- }
-
- /**
- * Same to {@link #asyncSend(String, Message, SendCallback)}.
- *
- * @param destination formats: `topicName:tags`
- * @param payload the Object to use as payload
- * @param sendCallback {@link SendCallback}
- */
- public void asyncSend(String destination, Object payload, SendCallback sendCallback) {
- asyncSend(destination, payload, sendCallback, producer.getSendMsgTimeout());
- }
-
- /**
- * Same to {@link #asyncSendOrderly(String, Message, String, SendCallback)} with send timeout specified in
- * addition.
- *
- * @param destination formats: `topicName:tags`
- * @param message {@link org.springframework.messaging.Message}
- * @param hashKey use this key to select queue. for example: orderId, productId ...
- * @param sendCallback {@link SendCallback}
- * @param timeout send timeout with millis
- */
- public void asyncSendOrderly(String destination, Message message, String hashKey, SendCallback sendCallback,
- long timeout) {
if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
- log.info("asyncSendOrderly failed. destination:{}, message is null ", destination);
+ log.info("asyncSend failed. destination:{}, message is null ", destination);
throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
}
try {
- org.apache.rocketmq.common.message.Message rocketMsg = convertToRocketMsg(destination, message);
- producer.send(rocketMsg, messageQueueSelector, hashKey, sendCallback, timeout);
+ if(aliyunProducer != null){
+ com.aliyun.openservices.ons.api.Message aliyunMsg = this.convertToAliyunRocketMsg(destination, message);
+ aliyunProducer.sendAsync(aliyunMsg, aliyunSendCallBackConvert(sendCallback));
+ }else if(defaultProducer != null){
+ org.apache.rocketmq.common.message.Message rocketMsg = convertToRocketMsg(destination, message);
+ defaultProducer.send(rocketMsg, sendCallback);
+ }else{
+ throw new RuntimeException("product为空，请检查配置文件是否配置：spring.rocketmq.aliyun，且值为true或false");
+ }
} catch (Exception e) {
- log.info("asyncSendOrderly failed. destination:{}, message:{} ", destination, message);
+ log.info("asyncSend failed. destination:{}, message:{} ", destination, message);
throw new MessagingException(e.getMessage(), e);
}
}
/**
- * Same to {@link #asyncSend(String, Message, SendCallback)} with send orderly with hashKey by specified.
+ * Same to {@link #asyncSend(String, Message, SendCallback)}.
*
* @param destination formats: `topicName:tags`
- * @param message {@link org.springframework.messaging.Message}
- * @param hashKey use this key to select queue. for example: orderId, productId ...
+ * @param payload the Object to use as payload
* @param sendCallback {@link SendCallback}
*/
- public void asyncSendOrderly(String destination, Message message, String hashKey, SendCallback sendCallback) {
- asyncSendOrderly(destination, message, hashKey, sendCallback, producer.getSendMsgTimeout());
+ public void asyncSend(String destination, Object payload, SendCallback sendCallback) {
+ Message message = this.doConvert(payload, null, null);
+ asyncSend(destination, message, sendCallback);
}
+
/**
- * Same to {@link #asyncSendOrderly(String, Message, String, SendCallback)}.
+ * Same to {@link #asyncSend(String, Message, SendCallback)} with send orderly with hashKey by specified.
*
* @param destination formats: `topicName:tags`
- * @param payload the Object to use as payload
+ * @param message {@link org.springframework.messaging.Message}
* @param hashKey use this key to select queue. for example: orderId, productId ...
* @param sendCallback {@link SendCallback}
*/
- public void asyncSendOrderly(String destination, Object payload, String hashKey, SendCallback sendCallback) {
- asyncSendOrderly(destination, payload, hashKey, sendCallback, producer.getSendMsgTimeout());
- }
+// public void asyncSendOrderly(String destination, Message message, String hashKey, SendCallback sendCallback) {
+// if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
+// log.info("asyncSendOrderly failed. destination:{}, message is null ", destination);
+// throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
+// }
+//
+// try {
+// org.apache.rocketmq.common.message.Message rocketMsg = convertToRocketMsg(destination, message);
+// //TODO zwg
+// throw new RuntimeException("暂时未整合阿里云Producer，不要使用");
+// //producer.send(rocketMsg, messageQueueSelector, hashKey, sendCallback, timeout);
+// } catch (Exception e) {
+// log.info("asyncSendOrderly failed. destination:{}, message:{} ", destination, message);
+// throw new MessagingException(e.getMessage(), e);
+// }
+// }
/**
- * Same to {@link #asyncSendOrderly(String, Object, String, SendCallback)} with send timeout specified in addition.
+ * Same to {@link #asyncSendOrderly(String, Message, String, SendCallback)}.
*
* @param destination formats: `topicName:tags`
* @param payload the Object to use as payload
* @param hashKey use this key to select queue. for example: orderId, productId ...
* @param sendCallback {@link SendCallback}
- * @param timeout send timeout with millis
*/
- public void asyncSendOrderly(String destination, Object payload, String hashKey, SendCallback sendCallback,
- long timeout) {
- Message message = this.doConvert(payload, null, null);
- asyncSendOrderly(destination, message, hashKey, sendCallback, timeout);
- }
+// public void asyncSendOrderly(String destination, Object payload, String hashKey, SendCallback sendCallback) {
+// Message message = this.doConvert(payload, null, null);
+// asyncSendOrderly(destination, message, hashKey, sendCallback);
+// }
/**
* Similar to UDP, this method won't wait for
@@ -349,8 +302,17 @@ public class RocketMQTemplate extends AbstractMessageSendingTemplate imp
}
try {
- org.apache.rocketmq.common.message.Message rocketMsg = convertToRocketMsg(destination, message);
- producer.sendOneway(rocketMsg);
+ if(aliyunProducer !=null){
+ //阿里云环境
+ com.aliyun.openservices.ons.api.Message aliyunMsg = convertToAliyunRocketMsg(destination, message);
+ aliyunProducer.sendOneway(aliyunMsg);
+ }else if(defaultProducer != null){
+ org.apache.rocketmq.common.message.Message rocketMsg = convertToRocketMsg(destination, message);
+ defaultProducer.sendOneway(rocketMsg);
+ }else{
+ throw new RuntimeException("product为空，请检查配置文件是否配置：spring.rocketmq.aliyun，且值为true或false");
+ }
+
} catch (Exception e) {
log.info("sendOneWay failed. destination:{}, message:{} ", destination, message);
throw new MessagingException(e.getMessage(), e);
@@ -375,20 +337,22 @@ public class RocketMQTemplate extends AbstractMessageSendingTemplate imp
* @param message {@link org.springframework.messaging.Message}
* @param hashKey use this key to select queue. for example: orderId, productId ...
*/
- public void sendOneWayOrderly(String destination, Message message, String hashKey) {
- if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
- log.info("sendOneWayOrderly failed. destination:{}, message is null ", destination);
- throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
- }
-
- try {
- org.apache.rocketmq.common.message.Message rocketMsg = convertToRocketMsg(destination, message);
- producer.sendOneway(rocketMsg, messageQueueSelector, hashKey);
- } catch (Exception e) {
- log.info("sendOneWayOrderly failed. destination:{}, message:{}", destination, message);
- throw new MessagingException(e.getMessage(), e);
- }
- }
+// public void sendOneWayOrderly(String destination, Message message, String hashKey) {
+// if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
+// log.info("sendOneWayOrderly failed. destination:{}, message is null ", destination);
+// throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
+// }
+//
+// try {
+// //TODO zwg
+// throw new RuntimeException("暂时未整合阿里云Producer，不要使用");
+// //org.apache.rocketmq.common.message.Message rocketMsg = convertToRocketMsg(destination, message);
+// //producer.sendOneway(rocketMsg, messageQueueSelector, hashKey);
+// } catch (Exception e) {
+// log.info("sendOneWayOrderly failed. destination:{}, message:{}", destination, message);
+// throw new MessagingException(e.getMessage(), e);
+// }
+// }
/**
* Same to {@link #sendOneWayOrderly(String, Message, String)}
@@ -396,21 +360,92 @@ public class RocketMQTemplate extends AbstractMessageSendingTemplate imp
* @param destination formats: `topicName:tags`
* @param payload the Object to use as payload
*/
- public void sendOneWayOrderly(String destination, Object payload, String hashKey) {
- Message message = this.doConvert(payload, null, null);
- sendOneWayOrderly(destination, message, hashKey);
- }
-
+// public void sendOneWayOrderly(String destination, Object payload, String hashKey) {
+// Message message = this.doConvert(payload, null, null);
+// sendOneWayOrderly(destination, message, hashKey);
+// }
+
+ @Override
public void afterPropertiesSet() throws Exception {
- Assert.notNull(producer, "Property 'producer' is required");
- producer.start();
+ if(aliyunProducer != null){
+ log.info("开始启动阿里云环境生产者");
+ aliyunProducer.start();
+ }else if(defaultProducer != null){
+ log.info("开始启动非阿里云环境生产者");
+ defaultProducer.start();
+ }else{
+ throw new RuntimeException("product为空，请检查配置文件是否配置：spring.rocketmq.aliyun，且值为true或false");
+ }
}
protected void doSend(String destination, Message message) {
SendResult sendResult = syncSend(destination, message);
log.debug("send message to `{}` finished. result:{}", destination, sendResult);
}
+ /**
+ * 转换阿里云返回对象
+ * @param aliyunSendResult
+ * @return
+ * 2018年3月23日 zhaowg
+ */
+ private SendResult convertAliyunSendResult(com.aliyun.openservices.ons.api.SendResult aliyunSendResult) {
+ SendResult sendResult = new SendResult();
+ sendResult.setMsgId(aliyunSendResult.getMessageId());
+ MessageQueue messageQueue = new MessageQueue(aliyunSendResult.getTopic(), null, 0);
+ sendResult.setMessageQueue(messageQueue);
+ sendResult.setSendStatus(SendStatus.SEND_OK);
+ return sendResult;
+ }
+ /**
+ * 转换为阿里云发送的消息对象
+ * @param destination formats: `topicName:tags`
+ * @param message {@link org.springframework.messaging.Message}
+ * @return
+ * 2018年3月23日 zhaowg
+ */
+ private com.aliyun.openservices.ons.api.Message convertToAliyunRocketMsg(String destination, Message message) {
+ Object payloadObj = message.getPayload();
+ byte[] payloads;
+
+ if (payloadObj instanceof String) {
+ payloads = ((String) payloadObj).getBytes(Charset.forName(charset));
+ } else {
+ try {
+ String jsonObj = this.objectMapper.writeValueAsString(payloadObj);
+ payloads = jsonObj.getBytes(Charset.forName(charset));
+ } catch (Exception e) {
+ throw new RuntimeException("convert to RocketMQ message failed.", e);
+ }
+ }
+
+ String[] tempArr = destination.split(":", 2);
+ String topic = tempArr[0];
+ String tags = "";
+ if (tempArr.length > 1) {
+ tags = tempArr[1];
+ }
+ com.aliyun.openservices.ons.api.Message rocketMsg = new com.aliyun.openservices.ons.api.Message(topic, tags, payloads);
+
+ MessageHeaders headers = message.getHeaders();
+ if (Objects.nonNull(headers) && !headers.isEmpty()) {
+ Object keys = headers.get(MessageConst.PROPERTY_KEYS);
+ if (!StringUtils.isEmpty(keys)) { // if headers has 'KEYS', set rocketMQ message key
+ rocketMsg.setKey(keys.toString());
+ }
+
+ headers.entrySet().stream()
+ .filter(entry -> !Objects.equals(entry.getKey(), MessageConst.PROPERTY_KEYS)
+ && !Objects.equals(entry.getKey(), "FLAG")
+ && !Objects.equals(entry.getKey(), "WAIT_STORE_MSG_OK")) // exclude "KEYS", "FLAG", "WAIT_STORE_MSG_OK"
+ .forEach(entry -> {
+ rocketMsg.putUserProperties("USERS_" + entry.getKey(), String.valueOf(entry.getValue())); // add other properties with prefix "USERS_"
+ });
+
+ }
+
+ return rocketMsg;
+ }
/**
* Convert spring message to rocketMQ message
*
@@ -508,8 +543,11 @@ public class RocketMQTemplate extends AbstractMessageSendingTemplate imp
@Override
public void destroy() {
- if (Objects.nonNull(producer)) {
- producer.shutdown();
+ if (Objects.nonNull(defaultProducer)) {
+ defaultProducer.shutdown();
+ }
+ if(Objects.nonNull(aliyunProducer)){
+ aliyunProducer.shutdown();
}
}
}