study/Spring-Cloud-Alibaba
1
0
Fork 0
mirror of https://gitee.com/mirrors/Spring-Cloud-Alibaba.git synced 2021-06-26 13:25:11 +08:00
Code Issues Releases Wiki Activity

rocketmq binder support transactional message and update example, docs for 1.x branch

Browse Source
This commit is contained in:
fangjian0423
2018-12-12 16:09:14 +08:00
parent 8f1b45b2e6
commit 8f24bd979b
6 changed files with 514 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQBinderConstants.java
View File

@@ -21,7 +21,7 @@ package org.springframework.cloud.stream.binder.rocketmq;
*/
public interface RocketMQBinderConstants {
String ENDPOINT_ID = "rocketmq_binder";
String ENDPOINT_ID = "rocketmq_binder";
/**
* Header key
@@ -32,6 +32,8 @@ public interface RocketMQBinderConstants {
String ROCKET_SEND_RESULT = "ROCKETMQ_SEND_RESULT";
String ROCKET_TRANSACTIONAL_ARG = "ROCKETMQ_TRANSACTIONAL_ARG";
String ACKNOWLEDGEMENT_KEY = "ACKNOWLEDGEMENT";
/**
@@ -39,23 +41,23 @@ public interface RocketMQBinderConstants {
*/
String LASTSEND_TIMESTAMP = "lastSend.timestamp";
interface Metrics {
interface Producer {
String PREFIX = "scs-rocketmq.producer.";
String TOTAL_SENT = "totalSent";
String TOTAL_SENT_FAILURES = "totalSentFailures";
String SENT_PER_SECOND = "sentPerSecond";
String SENT_FAILURES_PER_SECOND = "sentFailuresPerSecond";
}
interface Metrics {
interface Producer {
String PREFIX = "scs-rocketmq.producer.";
String TOTAL_SENT = "totalSent";
String TOTAL_SENT_FAILURES = "totalSentFailures";
String SENT_PER_SECOND = "sentPerSecond";
String SENT_FAILURES_PER_SECOND = "sentFailuresPerSecond";
}
interface Consumer {
String GROUP_PREFIX = "scs-rocketmq.consumerGroup.";
String PREFIX = "scs-rocketmq.consumer.";
String TOTAL_CONSUMED = "totalConsumed";
String CONSUMED_PER_SECOND = "consumedPerSecond";
String TOTAL_CONSUMED_FAILURES = "totalConsumedFailures";
String CONSUMED_FAILURES_PER_SECOND = "consumedFailuresPerSecond";
}
}
interface Consumer {
String GROUP_PREFIX = "scs-rocketmq.consumerGroup.";
String PREFIX = "scs-rocketmq.consumer.";
String TOTAL_CONSUMED = "totalConsumed";
String CONSUMED_PER_SECOND = "consumedPerSecond";
String TOTAL_CONSUMED_FAILURES = "totalConsumedFailures";
String CONSUMED_FAILURES_PER_SECOND = "consumedFailuresPerSecond";
}
}
}

65
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java
View File

@@ -16,6 +16,9 @@
package org.springframework.cloud.stream.binder.rocketmq;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.LocalTransactionExecuter;
import org.apache.rocketmq.client.producer.TransactionCheckListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
@@ -36,6 +39,7 @@ import org.springframework.cloud.stream.provisioning.ProducerDestination;
import org.springframework.integration.core.MessageProducer;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.ClassUtils;
/**
* @author Timur Valiev
@@ -71,9 +75,24 @@ public class RocketMQMessageChannelBinder extends
ExtendedProducerProperties<RocketMQProducerProperties> producerProperties,
MessageChannel errorChannel) throws Exception {
if (producerProperties.getExtension().getEnabled()) {
return new RocketMQMessageHandler(destination.getName(),
producerProperties.getExtension(),
RocketMQMessageHandler messageHandler = new RocketMQMessageHandler(
destination.getName(), producerProperties,
rocketBinderConfigurationProperties, instrumentationManager);
if (producerProperties.getExtension().getTransactional()) {
// transaction message check LocalTransactionExecuter
messageHandler.setLocalTransactionExecuter(
getClassConfiguration(destination.getName(),
producerProperties.getExtension().getExecuter(),
LocalTransactionExecuter.class));
// transaction message check TransactionCheckListener
messageHandler.setTransactionCheckListener(
getClassConfiguration(destination.getName(),
producerProperties.getExtension()
.getTransactionCheckListener(),
TransactionCheckListener.class));
}
return messageHandler;
}
else {
throw new RuntimeException("Binding for channel " + destination.getName()
@@ -120,4 +139,46 @@ public class RocketMQMessageChannelBinder extends
public RocketMQProducerProperties getExtendedProducerProperties(String channelName) {
return extendedBindingProperties.getExtendedProducerProperties(channelName);
}
private <T> T getClassConfiguration(String destName, String className,
Class<T> interfaceClass) {
if (StringUtils.isEmpty(className)) {
throw new RuntimeException("Binding for channel " + destName
+ " using transactional message, should set "
+ interfaceClass.getSimpleName() + " configuration"
+ interfaceClass.getSimpleName() + " should be set, like "
+ "'spring.cloud.stream.rocketmq.bindings.output.producer.xxx=TheFullClassNameOfYour"
+ interfaceClass.getSimpleName() + "'");
}
else if (StringUtils.isNotEmpty(className)) {
Class fieldClass;
// check class exists
try {
fieldClass = ClassUtils.forName(className,
RocketMQMessageChannelBinder.class.getClassLoader());
}
catch (ClassNotFoundException e) {
throw new RuntimeException("Binding for channel " + destName
+ " using transactional message, but " + className
+ " class is not found");
}
// check interface incompatible
if (!interfaceClass.isAssignableFrom(fieldClass)) {
throw new RuntimeException("Binding for channel " + destName
+ " using transactional message, but " + className
+ " is incompatible with " + interfaceClass.getSimpleName()
+ " interface");
}
try {
return (T) fieldClass.newInstance();
}
catch (Exception e) {
throw new RuntimeException("Binding for channel " + destName
+ " using transactional message, but " + className
+ " instance error", e);
}
}
return null;
}
}

154
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageHeaderAccessor.java
View File

@@ -16,6 +16,12 @@
package org.springframework.cloud.stream.binder.rocketmq;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ACKNOWLEDGEMENT_KEY;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ORIGINAL_ROCKET_MESSAGE;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKET_FLAG;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKET_SEND_RESULT;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKET_TRANSACTIONAL_ARG;
import java.util.HashMap;
import java.util.Map;
@@ -29,95 +35,105 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageHeaderAccessor;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ACKNOWLEDGEMENT_KEY;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ORIGINAL_ROCKET_MESSAGE;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKET_FLAG;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKET_SEND_RESULT;
/**
* @author Timur Valiev
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/
public class RocketMQMessageHeaderAccessor extends MessageHeaderAccessor {
public RocketMQMessageHeaderAccessor() {
super();
}
public RocketMQMessageHeaderAccessor() {
super();
}
public RocketMQMessageHeaderAccessor(Message<?> message) {
super(message);
}
public RocketMQMessageHeaderAccessor(Message<?> message) {
super(message);
}
public Acknowledgement getAcknowledgement(Message message) {
return message.getHeaders().get(ACKNOWLEDGEMENT_KEY, Acknowledgement.class);
}
public Acknowledgement getAcknowledgement(Message message) {
return message.getHeaders().get(ACKNOWLEDGEMENT_KEY, Acknowledgement.class);
}
public RocketMQMessageHeaderAccessor withAcknowledgment(Acknowledgement acknowledgment) {
setHeader(ACKNOWLEDGEMENT_KEY, acknowledgment);
return this;
}
public RocketMQMessageHeaderAccessor withAcknowledgment(
Acknowledgement acknowledgment) {
setHeader(ACKNOWLEDGEMENT_KEY, acknowledgment);
return this;
}
public String getTags() {
return getMessageHeaders().get(MessageConst.PROPERTY_TAGS) == null ? "" : (String) getMessageHeaders().get(MessageConst.PROPERTY_TAGS);
}
public String getTags() {
return getMessageHeaders().get(MessageConst.PROPERTY_TAGS) == null ? ""
: (String) getMessageHeaders().get(MessageConst.PROPERTY_TAGS);
}
public RocketMQMessageHeaderAccessor withTags(String tag) {
setHeader(MessageConst.PROPERTY_TAGS, tag);
return this;
}
public RocketMQMessageHeaderAccessor withTags(String tag) {
setHeader(MessageConst.PROPERTY_TAGS, tag);
return this;
}
public String getKeys() {
return getMessageHeaders().get(MessageConst.PROPERTY_KEYS) == null ? "" : (String) getMessageHeaders().get(MessageConst.PROPERTY_KEYS);
}
public String getKeys() {
return getMessageHeaders().get(MessageConst.PROPERTY_KEYS) == null ? ""
: (String) getMessageHeaders().get(MessageConst.PROPERTY_KEYS);
}
public RocketMQMessageHeaderAccessor withKeys(String keys) {
setHeader(MessageConst.PROPERTY_KEYS, keys);
return this;
}
public RocketMQMessageHeaderAccessor withKeys(String keys) {
setHeader(MessageConst.PROPERTY_KEYS, keys);
return this;
}
public MessageExt getRocketMessage() {
return getMessageHeaders().get(ORIGINAL_ROCKET_MESSAGE, MessageExt.class);
}
public MessageExt getRocketMessage() {
return getMessageHeaders().get(ORIGINAL_ROCKET_MESSAGE, MessageExt.class);
}
public RocketMQMessageHeaderAccessor withRocketMessage(MessageExt message) {
setHeader(ORIGINAL_ROCKET_MESSAGE, message);
return this;
}
public RocketMQMessageHeaderAccessor withRocketMessage(MessageExt message) {
setHeader(ORIGINAL_ROCKET_MESSAGE, message);
return this;
}
public Integer getDelayTimeLevel() {
return NumberUtils.toInt((String)getMessageHeaders().get(MessageConst.PROPERTY_DELAY_TIME_LEVEL), 0);
}
public Integer getDelayTimeLevel() {
return NumberUtils.toInt(
(String) getMessageHeaders().get(MessageConst.PROPERTY_DELAY_TIME_LEVEL),
0);
}
public RocketMQMessageHeaderAccessor withDelayTimeLevel(Integer delayTimeLevel) {
setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, delayTimeLevel);
return this;
}
public RocketMQMessageHeaderAccessor withDelayTimeLevel(Integer delayTimeLevel) {
setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, delayTimeLevel);
return this;
}
public Integer getFlag() {
return NumberUtils.toInt((String)getMessageHeaders().get(ROCKET_FLAG), 0);
}
public Integer getFlag() {
return NumberUtils.toInt((String) getMessageHeaders().get(ROCKET_FLAG), 0);
}
public RocketMQMessageHeaderAccessor withFlag(Integer delayTimeLevel) {
setHeader(ROCKET_FLAG, delayTimeLevel);
return this;
}
public RocketMQMessageHeaderAccessor withFlag(Integer delayTimeLevel) {
setHeader(ROCKET_FLAG, delayTimeLevel);
return this;
}
public SendResult getSendResult() {
return getMessageHeaders().get(ROCKET_SEND_RESULT, SendResult.class);
}
public Object getTransactionalArg() {
return getMessageHeaders().get(ROCKET_TRANSACTIONAL_ARG);
}
public static void putSendResult(MutableMessage message, SendResult sendResult) {
message.getHeaders().put(ROCKET_SEND_RESULT, sendResult);
}
public Object withTransactionalArg(Object arg) {
setHeader(ROCKET_TRANSACTIONAL_ARG, arg);
return this;
}
public Map<String, String> getUserProperties() {
Map<String, String> result = new HashMap<>();
for (Map.Entry<String, Object> entry : this.toMap().entrySet()) {
if (entry.getValue() instanceof String && !MessageConst.STRING_HASH_SET.contains(entry.getKey()) && !entry
.getKey().equals(MessageHeaders.CONTENT_TYPE)) {
result.put(entry.getKey(), (String)entry.getValue());
}
}
return result;
}
public SendResult getSendResult() {
return getMessageHeaders().get(ROCKET_SEND_RESULT, SendResult.class);
}
public static void putSendResult(MutableMessage message, SendResult sendResult) {
message.getHeaders().put(ROCKET_SEND_RESULT, sendResult);
}
public Map<String, String> getUserProperties() {
Map<String, String> result = new HashMap<>();
for (Map.Entry<String, Object> entry : this.toMap().entrySet()) {
if (entry.getValue() instanceof String
&& !MessageConst.STRING_HASH_SET.contains(entry.getKey())
&& !entry.getKey().equals(MessageHeaders.CONTENT_TYPE)) {
result.put(entry.getKey(), (String) entry.getValue());
}
}
return result;
}
}

49
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQMessageHandler.java
View File

@@ -21,10 +21,14 @@ import java.util.Map;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.LocalTransactionExecuter;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.TransactionCheckListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants;
import org.springframework.cloud.stream.binder.rocketmq.RocketMQMessageHeaderAccessor;
import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
@@ -47,16 +51,20 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
private InstrumentationManager instrumentationManager;
private final RocketMQProducerProperties producerProperties;
private LocalTransactionExecuter localTransactionExecuter;
private TransactionCheckListener transactionCheckListener;
private final ExtendedProducerProperties<RocketMQProducerProperties> producerProperties;
private final String destination;
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
protected volatile boolean running = false;
private volatile boolean running = false;
public RocketMQMessageHandler(String destination,
RocketMQProducerProperties producerProperties,
ExtendedProducerProperties<RocketMQProducerProperties> producerProperties,
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
InstrumentationManager instrumentationManager) {
this.destination = destination;
@@ -67,7 +75,16 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
@Override
public void start() {
producer = new DefaultMQProducer(destination);
if (producerProperties.getExtension().getTransactional()) {
producer = new TransactionMQProducer(destination);
if (transactionCheckListener != null) {
((TransactionMQProducer) producer)
.setTransactionCheckListener(transactionCheckListener);
}
}
else {
producer = new DefaultMQProducer(destination);
}
if (instrumentationManager != null) {
producerInstrumentation = instrumentationManager
@@ -77,8 +94,9 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
producer.setNamesrvAddr(rocketBinderConfigurationProperties.getNamesrvAddr());
if (producerProperties.getMaxMessageSize() > 0) {
producer.setMaxMessageSize(producerProperties.getMaxMessageSize());
if (producerProperties.getExtension().getMaxMessageSize() > 0) {
producer.setMaxMessageSize(
producerProperties.getExtension().getMaxMessageSize());
}
try {
@@ -139,7 +157,14 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
toSend.putUserProperty(entry.getKey(), entry.getValue());
}
SendResult sendRes = producer.send(toSend);
SendResult sendRes;
if (producerProperties.getExtension().getTransactional()) {
sendRes = producer.sendMessageInTransaction(toSend,
localTransactionExecuter, headerAccessor.getTransactionalArg());
}
else {
sendRes = producer.send(toSend);
}
if (!sendRes.getSendStatus().equals(SendStatus.SEND_OK)) {
throw new MQClientException("message hasn't been sent", null);
@@ -167,4 +192,14 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
}
public void setLocalTransactionExecuter(
LocalTransactionExecuter localTransactionExecuter) {
this.localTransactionExecuter = localTransactionExecuter;
}
public void setTransactionCheckListener(
TransactionCheckListener transactionCheckListener) {
this.transactionCheckListener = transactionCheckListener;
}
}

72
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQProducerProperties.java
View File

@@ -17,6 +17,8 @@
package org.springframework.cloud.stream.binder.rocketmq.properties;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.LocalTransactionExecuter;
import org.apache.rocketmq.client.producer.TransactionCheckListener;
/**
* @author Timur Valiev
@@ -24,28 +26,62 @@ import org.apache.rocketmq.client.producer.DefaultMQProducer;
*/
public class RocketMQProducerProperties {
private Boolean enabled = true;
private Boolean enabled = true;
/**
* Maximum allowed message size in bytes
* {@link DefaultMQProducer#maxMessageSize}
*/
private Integer maxMessageSize = 0;
/**
* Maximum allowed message size in bytes {@link DefaultMQProducer#maxMessageSize}
*/
private Integer maxMessageSize = 0;
public Boolean getEnabled() {
return enabled;
}
private Boolean transactional = false;
public void setEnabled(Boolean enabled) {
this.enabled = enabled;
}
/**
* full class name of {@link LocalTransactionExecuter}
*/
private String executer;
public Integer getMaxMessageSize() {
return maxMessageSize;
}
/**
* full class name of {@link TransactionCheckListener}
*/
private String transactionCheckListener;
public void setMaxMessageSize(Integer maxMessageSize) {
this.maxMessageSize = maxMessageSize;
}
public Boolean getEnabled() {
return enabled;
}
public void setEnabled(Boolean enabled) {
this.enabled = enabled;
}
public Integer getMaxMessageSize() {
return maxMessageSize;
}
public void setMaxMessageSize(Integer maxMessageSize) {
this.maxMessageSize = maxMessageSize;
}
public Boolean getTransactional() {
return transactional;
}
public void setTransactional(Boolean transactional) {
this.transactional = transactional;
}
public String getExecuter() {
return executer;
}
public void setExecuter(String executer) {
this.executer = executer;
}
public String getTransactionCheckListener() {
return transactionCheckListener;
}
public void setTransactionCheckListener(String transactionCheckListener) {
this.transactionCheckListener = transactionCheckListener;
}
}