From 8f24bd979b97c2abf8fd3458eea164df24d7931f Mon Sep 17 00:00:00 2001 From: fangjian0423 Date: Wed, 12 Dec 2018 16:09:14 +0800 Subject: [PATCH] rocketmq binder support transactional message and update example, docs for 1.x branch --- .../src/main/asciidoc-zh/rocketmq.adoc | 251 +++++++++++++++++- .../rocketmq/RocketMQBinderConstants.java | 38 +-- .../RocketMQMessageChannelBinder.java | 65 ++++- .../RocketMQMessageHeaderAccessor.java | 154 ++++++----- .../integration/RocketMQMessageHandler.java | 49 +++- .../RocketMQProducerProperties.java | 72 +++-- 6 files changed, 514 insertions(+), 115 deletions(-) diff --git a/spring-cloud-alibaba-docs/src/main/asciidoc-zh/rocketmq.adoc b/spring-cloud-alibaba-docs/src/main/asciidoc-zh/rocketmq.adoc index 238f0b4b..3724fd41 100644 --- a/spring-cloud-alibaba-docs/src/main/asciidoc-zh/rocketmq.adoc +++ b/spring-cloud-alibaba-docs/src/main/asciidoc-zh/rocketmq.adoc @@ -1 +1,250 @@ -== Spring Cloud Alibaba Rocket Binder +== Spring Cloud Alibaba RocketMQ Binder + +### RocketMQ 介绍 + +https://rocketmq.apache.org[RocketMQ] 是一款开源的分布式消息系统，基于高可用分布式集群技术，提供低延时的、高可靠的消息发布与订阅服务。同时，广泛应用于多个领域，包括异步通信解耦、企业解决方案、金融支付、电信、电子商务、快递物流、广告营销、社交、即时通信、移动应用、手游、视频、物联网、车联网等。 + +具有以下特点： + +* 能够保证严格的消息顺序 + +* 提供丰富的消息拉取模式 + +* 高效的订阅者水平扩展能力 + +* 实时的消息订阅机制 + +* 亿级消息堆积能力 + +### RocketMQ 基本使用 + +* 下载 RocketMQ + +下载 https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.3.2/rocketmq-all-4.3.2-bin-release.zip[RocketMQ最新的二进制文件]，并解压 + +解压后的目录结构如下： + +``` +apache-rocketmq +├── LICENSE +├── NOTICE +├── README.md +├── benchmark +├── bin +├── conf +└── lib +``` + +* 启动 NameServer + +```bash +nohup sh bin/mqnamesrv & +tail -f ~/logs/rocketmqlogs/namesrv.log +``` + +* 启动 Broker + +```bash +nohup sh bin/mqbroker -n localhost:9876 & +tail -f ~/logs/rocketmqlogs/broker.log +``` + +* 发送、接收消息 + +发送消息： + +```bash +sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer +``` + +发送成功后显示：`SendResult [sendStatus=SEND_OK, msgId= ...` + +接收消息： + +```bash +sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer +``` + +接收成功后显示：`ConsumeMessageThread_%d Receive New Messages: [MessageExt...` + +* 关闭 Server + +```bash +sh bin/mqshutdown broker +sh bin/mqshutdown namesrv +``` + +### Spring Cloud Stream 介绍 + +Spring Cloud Stream 是一个用于构建基于消息的微服务应用框架。它基于 SpringBoot 来创建具有生产级别的单机 Spring 应用，并且使用 `Spring Integration` 与 Broker 进行连接。 + +Spring Cloud Stream 提供了消息中间件配置的统一抽象，推出了 publish-subscribe、consumer groups、partition 这些统一的概念。 + +Spring Cloud Stream 内部有两个概念：Binder 和 Binding。 + +* Binder: 跟外部消息中间件集成的组件，用来创建 Binding，各消息中间件都有自己的 Binder 实现。 + +比如 `Kafka` 的实现 `KafkaMessageChannelBinder`，`RabbitMQ` 的实现 `RabbitMessageChannelBinder` 以及 `RocketMQ` 的实现 `RocketMQMessageChannelBinder`。 + +* Binding: 包括 Input Binding 和 Output Binding。 + +Binding 在消息中间件与应用程序提供的 Provider 和 Consumer 之间提供了一个桥梁，实现了开发者只需使用应用程序的 Provider 或 Consumer 生产或消费数据即可，屏蔽了开发者与底层消息中间件的接触。 + +.Spring Cloud Stream +image::https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/images/SCSt-overview.png[] + +使用 Spring Cloud Stream 完成一段简单的消息发送和消息接收代码： + +```java +MessageChannel messageChannel = new DirectChannel(); + +// 消息订阅 +((SubscribableChannel) messageChannel).subscribe(new MessageHandler() { + @Override + public void handleMessage(Message message) throws MessagingException { + System.out.println("receive msg: " + message.getPayload()); + } +}); + +// 消息发送 +messageChannel.send(MessageBuilder.withPayload("simple msg").build()); +``` + +这段代码所有的消息类都是 `spring-messaging` 模块里提供的。屏蔽具体消息中间件的底层实现，如果想用更换消息中间件，在配置文件里配置相关消息中间件信息以及修改 binder 依赖即可。 + +**Spring Cloud Stream 底层也是基于这段代码去做了各种抽象。** + +### Spring Cloud Alibaba RocketMQ Binder 实现原理 + +.RocketMQ Binder处理流程 +image::https://cdn.nlark.com/lark/0/2018/png/64647/1543560843558-24525bf4-1d0e-4e10-be5f-bdde7127f6e6.png[] + + +RocketMQ Binder 的核心主要就是这3个类：`RocketMQMessageChannelBinder`，`RocketMQInboundChannelAdapter` 和 `RocketMQMessageHandler`。 + +`RocketMQMessageChannelBinder` 是个标准的 Binder 实现，其内部构建 `RocketMQInboundChannelAdapter` 和 `RocketMQMessageHandler`。 + +`RocketMQMessageHandler` 用于 RocketMQ `Producer` 的启动以及消息的发送，其内部会根据 `spring-messaging` 模块内 `org.springframework.messaging.Message` 消息类，去创建 RocketMQ 的消息类 `org.apache.rocketmq.common.message.Message`。 + +在构造 `org.apache.rocketmq.common.message.Message` 的过程中会根据 `org.springframework.messaging.Message` 的 Header 构造成 `RocketMQMessageHeaderAccessor`。然后再根据 `RocketMQMessageHeaderAccessor` 中的一些属性，比如 tags、keys、flag等属性设置到 RocketMQ 的消息类 `org.apache.rocketmq.common.message.Message` 中。 + +`RocketMQInboundChannelAdapter` 用于 RocketMQ `Consumer` 的启动以及消息的接收。其内部还支持 https://github.com/spring-projects/spring-retry[spring-retry] 的使用。 + +在消费消息的时候可以从 Header 中获取 `Acknowledgement` 并进行一些设置。 + +比如使用 `MessageListenerConcurrently` 进行异步消费的时候，可以设置延迟消费： + +```java +@StreamListener("input") +public void receive(Message message) { + RocketMQMessageHeaderAccessor headerAccessor = new RocketMQMessageHeaderAccessor(message); + Acknowledgement acknowledgement = headerAccessor.getAcknowledgement(message); + acknowledgement.setConsumeConcurrentlyStatus(ConsumeConcurrentlyStatus.RECONSUME_LATER); + acknowledgement.setConsumeConcurrentlyDelayLevel(1); +} +``` + +比如使用 `MessageListenerOrderly` 进行顺序消费的时候，可以设置延迟消费： + +```java +@StreamListener("input") +public void receive(Message message) { + RocketMQMessageHeaderAccessor headerAccessor = new RocketMQMessageHeaderAccessor(message); + Acknowledgement acknowledgement = headerAccessor.getAcknowledgement(message); + acknowledgement.setConsumeOrderlyStatus(ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT); + acknowledgement.setConsumeOrderlySuspendCurrentQueueTimeMill(5000); +} +``` + +Provider端支持的配置： + +:frame: topbot +[width="60%",options="header"] +|==== +^|配置项 ^|含义 ^| 默认值 +|`spring.cloud.stream.rocketmq.bindings.your-output-binding.producer.enabled`|是否启用producer|true +|`spring.cloud.stream.rocketmq.bindings.your-output-binding.producer.max-message-size`|消息发送的最大字节数|0(大于0才会生效，RocketMQ 默认值为4M = 1024 * 1024 * 4) +|`spring.cloud.stream.rocketmq.bindings.your-output-binding.producer.transactional`|是否启用 `TransactionMQProducer` 发送事务消息|false +|`spring.cloud.stream.rocketmq.bindings.your-output-binding.producer.executer`|事务消息对应的 `org.apache.rocketmq.client.producer.LocalTransactionExecuter` 接口实现类 全类名。 比如 `org.test.MyExecuter`| +|`spring.cloud.stream.rocketmq.bindings.your-output-binding.producer.transaction-check-listener`|事务消息对应的 `org.apache.rocketmq.client.producer.TransactionCheckListener` 接口实现类 全类名。 比如 `org.test.MyTransactionCheckListener`| +|==== + +Consumer端支持的配置： + +:frame: topbot +[width="60%",options="header"] +|==== +^|配置项 ^|含义| 默认值 +|`spring.cloud.stream.rocketmq.bindings.your-input-binding.consumer.enabled`|是否启用consumer|true +|`spring.cloud.stream.rocketmq.bindings.your-input-binding.consumer.tags`|Consumer订阅只有包括这些tags的topic消息。多个标签之间使用 "\|\|" 分割(不填表示不进行tags的过滤，订阅所有消息)| +|`spring.cloud.stream.rocketmq.bindings.your-input-binding.consumer.sql`|Consumer订阅满足sql要求的topic消息(如果同时配置了tags内容，sql的优先级更高)| +|`spring.cloud.stream.rocketmq.bindings.your-input-binding.consumer.broadcasting`|Consumer是否是广播模式|false +|`spring.cloud.stream.rocketmq.bindings.your-input-binding.consumer.orderly`|顺序消费 or 异步消费|false +|==== + +### Endpoint支持 + +在使用Endpoint特性之前需要在 Maven 中添加 `spring-boot-starter-actuator` 依赖，并在配置中允许 Endpoints 的访问。 + +* Spring Boot 1.x 中添加配置 `management.security.enabled=false`。暴露的 endpoint 路径为 `/rocketmq_binder` +* Spring Boot 2.x 中添加配置 `management.endpoints.web.exposure.include=*`。暴露的 endpoint 路径为 `/actuator/rocketmq-binder` + +Endpoint 会统计消息最后一次发送的数据，消息发送成功或失败的次数，消息消费成功或失败的次数等数据。 + +```json +{ + "runtime": { + "lastSend.timestamp": 1542786623915 + }, + "metrics": { + "scs-rocketmq.consumer.test-topic.totalConsumed": { + "count": 11 + }, + "scs-rocketmq.consumer.test-topic.totalConsumedFailures": { + "count": 0 + }, + "scs-rocketmq.producer.test-topic.totalSentFailures": { + "count": 0 + }, + "scs-rocketmq.consumer.test-topic.consumedPerSecond": { + "count": 11, + "fifteenMinuteRate": 0.012163847780107841, + "fiveMinuteRate": 0.03614605351360527, + "meanRate": 0.3493213353657594, + "oneMinuteRate": 0.17099243039490175 + }, + "scs-rocketmq.producer.test-topic.totalSent": { + "count": 5 + }, + "scs-rocketmq.producer.test-topic.sentPerSecond": { + "count": 5, + "fifteenMinuteRate": 0.005540151995103271, + "fiveMinuteRate": 0.01652854617838251, + "meanRate": 0.10697493212602836, + "oneMinuteRate": 0.07995558537067671 + }, + "scs-rocketmq.producer.test-topic.sentFailuresPerSecond": { + "count": 0, + "fifteenMinuteRate": 0.0, + "fiveMinuteRate": 0.0, + "meanRate": 0.0, + "oneMinuteRate": 0.0 + }, + "scs-rocketmq.consumer.test-topic.consumedFailuresPerSecond": { + "count": 0, + "fifteenMinuteRate": 0.0, + "fiveMinuteRate": 0.0, + "meanRate": 0.0, + "oneMinuteRate": 0.0 + } + } +} +``` + +注意：要想查看统计数据需要在pom里加上 https://mvnrepository.com/artifact/io.dropwizard.metrics/metrics-core[metrics-core依赖]。如若不加，endpoint 将会显示 warning 信息而不会显示统计信息： + +```json +{ + "warning": "please add metrics-core dependency, we use it for metrics" +} +``` \ No newline at end of file diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQBinderConstants.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQBinderConstants.java index 9c8b5221..a54d2c1c 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQBinderConstants.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQBinderConstants.java @@ -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"; + } + } } diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java index 03009b43..4982ffa1 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java @@ -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 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 getClassConfiguration(String destName, String className, + Class 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; + } + } diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageHeaderAccessor.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageHeaderAccessor.java index fa480681..23671b60 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageHeaderAccessor.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageHeaderAccessor.java @@ -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 Jim */ 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 getUserProperties() { - Map result = new HashMap<>(); - for (Map.Entry 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 getUserProperties() { + Map result = new HashMap<>(); + for (Map.Entry 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; + } } diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQMessageHandler.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQMessageHandler.java index 0ab7386c..331df01b 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQMessageHandler.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQMessageHandler.java @@ -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 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 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; + } + } \ No newline at end of file diff --git a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQProducerProperties.java b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQProducerProperties.java index 8c252834..1a05ad50 100644 --- a/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQProducerProperties.java +++ b/spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQProducerProperties.java @@ -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; + } }