Spring Cloud Stream集成RocketMQ（kafka/rabbitMQ通用）

什么是Spring Cloud Stream

Spring Cloud Stream 是 Spring 生态系统中的一个框架，用于简化构建消息驱动微服务的开发和集成。它通过抽象化的方式将消息中间件（如 RabbitMQ、Kafka、RocketMQ 等）的复杂通信逻辑封装成简单的编程模型，使开发者能够专注于业务逻辑，而无需过多关注底层消息系统的实现细节。

详细解释

这里是官网代码中推出的解释：
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 生产或消费数据即可，屏蔽了开发者与底层消息中间件的接触。

总结：
Binder：解决“用什么消息中间件”的问题（如 Kafka vs RabbitMQ）。
Binding：解决“消息从哪里来、到哪里去”的问题（如 Topic 名称、消费者组）。

协作关系：

下图是 Spring Cloud Stream 的架构设计。

+-------------------+         +-------------------+
|    Application    |         |    Application    |
|  (Microservice)   |         |  (Microservice)   |
+-------------------+         +-------------------+|                               ||  Output Binding (Producer)    |  Input Binding (Consumer)↓                               ↓
+--------------------------------------------------+
|               Binder (抽象层)                     |
|   (Kafka/RabbitMQ/RocketMQ 的适配实现)            |
+--------------------------------------------------+|                               |↓                               ↓
+-------------------+         +-------------------+
|   Message Broker  |         |   Message Broker  |
|  (e.g., Kafka)    |         |  (e.g., RabbitMQ) |
+-------------------+         +-------------------+

业务代码 → Binding（定义通道） → Binder（连接中间件） → 消息中间件

根据官网文档，集成了下面这些消息中间件或者流事件平台。这里用rokectMQ举例

使用说明

下载github中的rocketMQ代码

1.首先点开下面github中的RocketMQ示例代码

直接下载全部 直接看examples

下载代码，可以看见很多示例，下面以(orderly 顺序消息)说明

这里他直接在启动类中简单实现了生产数据和消费数据的代码

并且带有说明 见readme

代码说明（orderly顺序消费）

@SpringBootApplication
public class RocketMQOrderlyConsumeApplication {private static final Logger log = LoggerFactory.getLogger(RocketMQOrderlyConsumeApplication.class);@Autowiredprivate StreamBridge streamBridge;/**** tag array.*/public static final String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};public static void main(String[] args) {SpringApplication.run(RocketMQOrderlyConsumeApplication.class, args);}@Beanpublic ApplicationRunner producer() {return args -> {for (int i = 0; i < 100; i++) {String key = "KEY" + i;Map<String, Object> headers = new HashMap<>();headers.put(MessageConst.PROPERTY_KEYS, key);headers.put(MessageConst.PROPERTY_TAGS, tags[i % tags.length]);headers.put(MessageConst.PROPERTY_ORIGIN_MESSAGE_ID, i);Message<SimpleMsg> msg = new GenericMessage(new SimpleMsg("Hello RocketMQ " + i), headers);streamBridge.send("producer-out-0", msg);}};}@Beanpublic Consumer<Message<SimpleMsg>> consumer() {return msg -> {String tagHeaderKey = RocketMQMessageConverterSupport.toRocketHeaderKey(MessageConst.PROPERTY_TAGS).toString();log.info(Thread.currentThread().getName() + " Receive New Messages: " + msg.getPayload().getMsg() + " TAG:" +msg.getHeaders().get(tagHeaderKey).toString());try {Thread.sleep(100);}catch (InterruptedException ignored) {}};}}

配置文件

server:port: 28082
spring:application:name: rocketmq-orderly-consume-examplecloud:stream:function:definition: consumer;rocketmq:binder:name-server: localhost:9876bindings:producer-out-0:producer:group: output_1messageQueueSelector: orderlyMessageQueueSelectorconsumer-in-0:consumer:# tag: {@code tag1||tag2||tag3 }; sql: {@code 'color'='blue' AND 'price'>100 } .subscription: 'TagA || TagC || TagD'push:orderly: truebindings:producer-out-0:destination: orderlyconsumer-in-0:destination: orderlygroup: orderly-consumerlogging:level:org.springframework.context.support: debug

配置文件解释

主要配置

绑定服务器

rocketmq:binder:name-server: localhost:9876 //RocketMQ 的 NameServer 地址为 localhost:9876，用于获取 Broker 路由信息。意思是这里只有一个broker，并不是集群配置

生产者消费者配置

bindings:producer-out-0:producer:group: output_1 //生产者组:生产者组名为 output_1，用于事务消息或消息查询。messageQueueSelector: orderlyMessageQueueSelector //队列选择器:使用自定义的 orderlyMessageQueueSelector 选择消息队列，确保相同业务标识的消息发往同一队列，实现顺序性。

consumer-in-0:consumer:subscription: 'TagA || TagC || TagD'//订阅过滤:使用 Tag 过滤，订阅包含 TagA、TagC 或 TagD 的消息（逻辑或）。push:orderly: true//顺序消费:push.orderly: true 启用顺序消费模式，按队列顺序单线程处理消息。

生产者消费者绑定组和目的地

bindings:producer-out-0:destination: orderly //生产者目的地:生产者发送至 Topic 为 orderly。consumer-in-0:destination: orderlygroup: orderly-consumer //消费者组:消费者组名为 orderly-consumer，相同组内消费者分摊消费队列，不同组独立消费。

producer

这里逻辑很简单，就循环发送了100条数据，顺序发送给不同的tags,组装成了Message对象，然后通过streamBridge发送到producer-out-0的通道

@Bean
public ApplicationRunner producer() {return args -> {for (int i = 0; i < 100; i++) {String key = "KEY" + i;// 设置 RocketMQ 消息头Map<String, Object> headers = new HashMap<>();headers.put(MessageConst.PROPERTY_KEYS, key);       // 消息的唯一标识（RocketMQ 的 KEY）headers.put(MessageConst.PROPERTY_TAGS, tags[i % tags.length]); // 消息的 Tag（按 tags 数组循环分配）headers.put(MessageConst.PROPERTY_ORIGIN_MESSAGE_ID, i); // 自定义原始消息ID（可选）// 创建消息对象：包含 payload 和 headersMessage<SimpleMsg> msg = new GenericMessage<>(new SimpleMsg("Hello RocketMQ " + i), headers);// 发送消息到名为 "producer-out-0" 的输出通道streamBridge.send("producer-out-0", msg);}};
}

selector（供生产者使用）

OrderlyMessageQueueSelector 的作用是 供生产者使用的，用于在发送顺序消息时选择特定的消息队列（MessageQueue），确保同一业务逻辑的消息被发送到同一个队列中，从而保证消费者能够按顺序消费。

@Component
public class OrderlyMessageQueueSelector implements MessageQueueSelector {private static final Logger log = LoggerFactory.getLogger(OrderlyMessageQueueSelector.class);/*** to select a fixed queue by id.* @param mqs all message queues of this topic.//当前主题（Topic）下的所有队列* @param msg mq message.//这是即将被消费的消息对象。它包含了消息的内容、属性和一些元数据。* @param arg mq arguments.//这个参数是消费者传入的自定义参数，通常用来携带一些额外的信息。* @return message queue selected.*/@Overridepublic MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {Integer id = (Integer) ((MessageHeaders) arg).get(MessageConst.PROPERTY_ORIGIN_MESSAGE_ID);int index = id % RocketMQOrderlyConsumeApplication.tags.length % mqs.size(); //id%5%队列长度return mqs.get(index);}
}

consumer

1.每个队列由独立线程顺序消费。

2.同一队列中的消息按发送顺序处理，不同队列的消息可能并行处理。

	@Beanpublic Consumer<Message<SimpleMsg>> consumer() {return msg -> {String tagHeaderKey = RocketMQMessageConverterSupport.toRocketHeaderKey(MessageConst.PROPERTY_TAGS).toString();log.info(Thread.currentThread().getName() + " Receive New Messages: " + msg.getPayload().getMsg() + " TAG:" +msg.getHeaders().get(tagHeaderKey).toString());try {Thread.sleep(100);}catch (InterruptedException ignored) {}};}

因为前面设计了selector，所以这里的消费结构应该是
假如这里的队列是默认的4

Thread-0 Receive: Hello RocketMQ 0 TAG:TagA
Thread-0 Receive: Hello RocketMQ 4 TAG:TagE
Thread-0 Receive: Hello RocketMQ 5 TAG:TagA
Thread-0 Receive: Hello RocketMQ 9 TAG:TagE
...（后续i=10,14,15...）

Thread-1 Receive: Hello RocketMQ 1 TAG:TagB
Thread-1 Receive: Hello RocketMQ 6 TAG:TagB
Thread-1 Receive: Hello RocketMQ 11 TAG:TagB
...（后续i=16,21...）

Thread-2 Receive: Hello RocketMQ 2 TAG:TagC
Thread-2 Receive: Hello RocketMQ 7 TAG:TagC
Thread-2 Receive: Hello RocketMQ 12 TAG:TagC
...（后续i=17,22...）

Thread-3 Receive: Hello RocketMQ 3 TAG:TagD
Thread-3 Receive: Hello RocketMQ 8 TAG:TagD
Thread-3 Receive: Hello RocketMQ 13 TAG:TagD
...（后续i=18,23...）

同一个队列中消息是顺序的，这里的thread0中有A,E两个标签，如果要避免这种情况，应该把队列设置为Tags.size的长度
他这里的设计应该就是为了尽可能的将不同标签分布在不同的队列，最终形成同一队列对应同一标签，然后实现顺序消费

实际开发案例（支付订单）说明

有了上面的案例下面我理解起来就方便很多了
注意：下面代码并不完整，只是一个大致逻辑说明

这里以支付订单案例说明

下面是代码前置，就是一个创建订单的流程，有兴趣的可以了解下，不然可以直接跳过看生产者消费者配置
一般咱们支付之前都会先生成订单，参数除了正常的支付单号，支付时间这些基本的东西外有一个支付倒计时这个功能，这个支付倒计时一般是咱们后台给配置的：这里我举个例，比如说后台模板中配置了1.消费下单：15分钟、2.通联支付：30分钟等等，这里我们会根据支付单号查询数据库对应的支付倒计时，这里超时咱们就可以用rockeMQ中延时队列来进行处理
下面代码可以不看，就是一个创建支付单的流程

1. 检测订单是否存在
2. 获取收益台模板（就是上面说的获取倒计时等数据这样一个东西）
3. 先存数据库（防止前端多次下单）后支付的时候再调用第三方接口（也可以是直接对接银行）
4. 存完设置redis缓存信息，防止多次创建订单
5. 将订单数据假如延迟队列

    @Overridepublic BillsPlan save(PaymentBillsDTO paymentBillsDTO) {String key = redisUtil.get("order:" + paymentBillsDTO.getBusinessOrderNo());if (key != null) {log.info("支付订单已创建，请前往收银台支付!");throw ExFactory.bizException(PaymentError.PAYMENT_BILL_COLLECTING);}// 检查订单是否存在PaymentBills byId = this.getOne(Wrappers.<PaymentBills>lambdaQuery().eq(PaymentBills::getBusinessOrderNo, paymentBillsDTO.getBusinessOrderNo()).eq(PaymentBills::getPaymentBillStatus, AgentCollectStatusEnum.COLLECT_SUCCESS.getStatus()).last("limit 1"));if (byId != null) {throw ExFactory.bizException(PaymentError.PAYMENT_BILL_FINISHED);}// 获取收银台PaymentTransactionType xiaofeixiadan = paymentTransactionTypeService.getOne(Wrappers.<PaymentTransactionType>lambdaQuery().eq(PaymentTransactionType::getTypeCode, "xiaofeixiadan"));if (Objects.isNull(xiaofeixiadan)) {throw ExFactory.bizException(PaymentError.PAYMENT_TRANSACTION_TYPE_NOT_EXIST);}Integer typeId = xiaofeixiadan.getId();CashierTemplate cashierTemplate = cashierTemplateService.getOne(Wrappers.<CashierTemplate>lambdaQuery().eq(CashierTemplate::getTransactionTypeId, typeId));if (Objects.isNull(cashierTemplate)) {throw ExFactory.bizException(PaymentError.CASHIER_TEMPLATE_NOT_EXIST);}Integer delayLevel;try {delayLevel = RocketMqDelayLevelEnum.getLevelByMinutes(cashierTemplate.getPaymentCountdown());} catch (Exception e) {throw ExFactory.bizException(PaymentError.CASHIER_TEMPLATE_BAD_TIMEOUT_PARAM);}// 保存数据到payment_bills表PaymentBills paymentBills = PaymentBillsConverter.INSTANCE.from(paymentBillsDTO);paymentBills.setPaymentBillType(TransactionTypeEnum.PAY.getCode());paymentBills.setPaymentBillStatus("1");this.save(paymentBills);// 保存数据到payment_bills_plan表,TODO 根据活动判断是否需要生成多条支付计划，目前只生成一条BillsPlan billsPlan = new BillsPlan();BeanUtil.copyProperties(paymentBillsDTO, billsPlan);billsPlan.setPaymentBillId(paymentBills.getPaymentBillId());billsPlan.setPricingSource("银行卡支付");billsPlan.setPaymentBillStatus("1");billsPlan.setPaymentBillType(TransactionTypeEnum.PAY.getCode());billsPlanService.save(billsPlan);// 记录第三方支付单PaymentThirdBills paymentThirdBills = new PaymentThirdBills();BeanUtil.copyProperties(billsPlan, paymentThirdBills);paymentThirdBills.setChannel("1");paymentThirdBillsService.save(paymentThirdBills);// redis设置订单失效时间redisUtil.set("order:" + paymentBills.getBusinessOrderNo(), String.valueOf(paymentBills.getPaymentBillId()), 30, TimeUnit.MINUTES);redisUtil.expire("order:" + paymentBills.getBusinessOrderNo(), 30, TimeUnit.MINUTES);// 发送延迟消息用于处理超时订单MessageDTO messageDTO = new MessageDTO();messageDTO.setDataJson(String.valueOf(paymentBills.getPaymentBillId()));messageDTO.setTag("payment");messageDTO.setType(AsyncExecuteTypeEnums.DELAY_CHECK_PAYMENT_RESULT.getType());producer.sendDelayMessage(messageDTO, delayLevel);inspectPayScheduleRpcServiceI.updateInspectPayScheduleStatus(paymentBills.getBusinessOrderNo(), 1);return billsPlan;}

producer

这里跟之前的案例没什么区别，都是streamBridge来发送消息，只不过这里是发送延时（延迟）消费，rocketMQ会根据设置的等级来设置延时时间

@RefreshScope
@Service
@Slf4j
public class RocketMqProducer {@Resourceprivate StreamBridge streamBridge;@Value("${spring.cloud.stream.paymentProducer}") // 在nacos中读取配置private String messageProducer;public <T> void sendMqMessage(MessageDTO dto) {streamBridge.send(messageProducer,MessageBuilder.withPayload(dto).setHeader(MessageConst.PROPERTY_TAGS, dto.getTag()).setHeader(MessageConst.PROPERTY_KEYS, dto.getType()).build());}public void sendDelayMessage(MessageDTO dto, Integer delayLevel) {// 创建消息头，设置延迟级别Map<String, Object> headers = new HashMap<>();headers.put(MessageConst.PROPERTY_DELAY_TIME_LEVEL, String.valueOf(delayLevel));headers.put(MessageConst.PROPERTY_TAGS,dto.getTag());headers.put(MessageConst.PROPERTY_KEYS,dto.getType());// 创建消息Message<MessageDTO> message = MessageBuilder.withPayload(dto).copyHeaders(headers).build();// 使用StreamBridge发送消息boolean sent = streamBridge.send(messageProducer, message);if (sent) {System.out.println("延迟消息发送成功");log.info("当前秒数：{}", LocalDateTime.now().getSecond());} else {System.out.println("延迟消息发送失败");}}}

nacos中的静态配置

# 配置 rocketmq 的 nameserver 地址
spring.cloud.stream.rocketmq.binder.name-server=******
spring.cloud.stream.rocketmq.producer.send-type=ASYNC
# 定义 通道 为 paymentProducer 的 生产者，paymentTransactionProducer为有事务的生产者
spring.cloud.stream.paymentProducer=paymentProducer-out-0
spring.cloud.stream.bindings.paymentProducer-out-0.binder=rocketmq
spring.cloud.stream.bindings.paymentProducer-out-0.content-type=application/json
spring.cloud.stream.bindings.paymentProducer-out-0.destination=payment-topic
spring.cloud.stream.paymentTransactionProducer=paymentTransactionProducer-out-0
spring.cloud.stream.bindings.paymentTransactionProducer-out-0.binder=rocketmq
spring.cloud.stream.bindings.paymentTransactionProducer-out-0.content-type=application/json
spring.cloud.stream.bindings.paymentTransactionProducer-out-0.destination=payment-topic
spring.cloud.stream.rocketmq.bindings.paymentTransactionProducer-out-0.producer.producerType=Trans
spring.cloud.stream.rocketmq.bindings.paymentTransactionProducer-out-0.producer.transactionListener=RocketMqTransactionListener# 定义 通道 为 paymentConsumer 的 消费者，tags 定义只接受 payment和all 消息
spring.cloud.stream.bindings.paymentConsumer-in-0.binder=rocketmq
spring.cloud.stream.bindings.paymentConsumer-in-0.content-type=application/json
spring.cloud.stream.bindings.paymentConsumer-in-0.destination=payment-topic
spring.cloud.stream.bindings.paymentConsumer-in-0.group=payment-customer-group
spring.cloud.stream.rocketmq.bindings.paymentConsumer-in-0.consumer.group=payment-customer-group
spring.cloud.stream.rocketmq.bindings.paymentConsumer-in-0.consumer.subscription=payment||all
spring.cloud.stream.rocketmq.bindings.paymentConsumer-in-0.consumer.messageModel=CLUSTERING

这里和案例中的都差不多

consumer

首先定义paymentConsumer的bean对象来接收名为paymentConsumer的topic，编程式事务根据不同的类型来执行不同的方法
着重看注释的地方

1.service.execute(dto);
2.getData(dto);//获取数据参数类型
3. asyncExcute(data);//根据参数类型进行重载

@Slf4j
@Configuration
public class AsyncExecuteConsumer {@Value("${payment.asyncMsg.maxRetryCount}")private Integer maxRetryCount;@Resourceprivate AsyncRetryInfoMapper asyncRetryInfoMapper;@Resourceprivate TransactionTemplate transactionTemplate;@Beanpublic Consumer<MessageDTO> paymentConsumer() {return message -> {log.info("paymentConsumer接到消息：{}", message);handleMessage(message);};}public void handleMessage(MessageDTO dto) {log.info("异步执行流程 接收MQ Content：{}", JSON.toJSONString(dto));if (StringUtils.isEmpty(dto.getType())){log.error("MQ消息type类型为空");return;}AsyncExecuteTypeEnums byType = AsyncExecuteTypeEnums.getByType(dto.getType());if(Objects.isNull(byType)){log.error("MQ消息type类型错误:{}", dto.getType());return;}AsyncExecuteService service = AsyncExecuteService.getService(byType); //这里通过类型获取对应的执行service对象if (Objects.nonNull(service)) {try {// 使用编程式事务确保事务正确传播transactionTemplate.execute(status -> {try {service.execute(dto);return true;} catch (Exception e) {status.setRollbackOnly();throw e;}});} catch (Exception e) {log.error("{}异步任务执行异常:{}",byType.getDesc(),e);// 记录异常信息if(checkRetryCount(dto.getCurrRetryCount())){dto.setErrorMsg(e.getCause().getMessage());dto.setCurrRetryCount(dto.getCurrRetryCount()+1);// 重试log.info("{}异步任务第{}次重试",byType.getDesc(), dto.getCurrRetryCount());handleMessage(dto);`在这里插入代码片`}else{//记录异常到数据库log.info("{}异步任务达到最大重试次数，入库",byType.getDesc());asyncRetryInfoMapper.insert(MsgRetryConverter.INSTANCE.toRetry(dto));}}}}/*** 检查是否可重试* @param currentRetryCount* @return*/public Boolean checkRetryCount(Integer currentRetryCount) {currentRetryCount++;return currentRetryCount <= maxRetryCount;}}

这里通过枚举定义了3个类型 入账，支付，提现

@Getter
public enum AsyncExecuteTypeEnums {/*** 入账*/ACCOUNTING("accouting", "入账"),DELAY_CHECK_PAYMENT_RESULT("delayCheckPaymentResult", "延迟检测支付结果"),DELAY_CHECK_WITHDRAW_RESULT("delayCheckWithdrawResult", "延迟检测提现结果"),;private final String type;private final String desc;AsyncExecuteTypeEnums(String type, String desc) {this.type = type;this.desc = desc;}public static AsyncExecuteTypeEnums getByType(String type) {for (AsyncExecuteTypeEnums asyncExecuteTypeEnums : AsyncExecuteTypeEnums.values()) {if (asyncExecuteTypeEnums.getType().equals(type)) {return asyncExecuteTypeEnums;}}return null;}
}

@Slf4j
public abstract class AsyncExecuteService<T> {/*** Service仓库*/protected static Map<AsyncExecuteTypeEnums, AsyncExecuteService> SERVICES = new HashMap<>();/*** 获取Service** @param type 类型* @return Service*/public static AsyncExecuteService getService(AsyncExecuteTypeEnums type) {return SERVICES.get(type);}/*** 执行流程** @param dto 参数*/@Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRES_NEW)public void execute(MessageDTO dto) {try {T data = getData(dto); //获取对应的类型，以便根据业务执行不同的代码if(bizVerify(data)){asyncExcute(data);}} catch (Exception e) {log.error("执行异步任务时发生异常:{}", e);throw e;}}/*** 获取数据** @param dto 参数* @return 结果*/protected T getData(MessageDTO dto) {try {ParameterizedType parameterizedType = (ParameterizedType) this.getClass().getGenericSuperclass(); //this.getClass().getGenericSuperclass() 获取当前类的泛型父类类型，即 AsyncExecuteService<T>。@SuppressWarnings("unchecked")Class<T> clazz = (Class<T>) parameterizedType.getActualTypeArguments()[0];//parameterizedType.getActualTypeArguments()[0] 获取泛型参数 T 的实际类型。if(clazz.isInstance(String.class)) {return (T) dto.getDataJson();}return JSON.parseObject(dto.getDataJson(), clazz);//使用 JSON.parseObject(dto.getDataJson(), clazz) 将 dataJson 字符串转换为指定的类型 T。}catch (Exception e) {log.error("异步执行流程 转换数据异常 数据：{}", dto, e);throw new RuntimeException("数据转换异常", e);}}/*** 初始化Factory*/@PostConstructprotected abstract void registerService();/*** 验证业务上的事务是否提交** @param dto 参数*/protected abstract Boolean bizVerify(T dto);/*** 执行核心业务** @param dto 参数*/protected abstract void asyncExcute(T dto);}

这里继承了AsyncExecuteService这个抽象类用于实现具体执行体

@Slf4j
@Service
public class PaymentBillsDelayConsumer extends AsyncExecuteService<String> {@Lazy@Resourceprivate PaymentBillsService paymentBillsService;@Lazy@Resourceprivate BillsPlanService billsPlanService;@Lazy@Resourceprivate PaymentThirdBillsService paymentThirdBillsService;@Lazy@Resourceprivate PaymentRequestService paymentRequestService;@Lazy@Resourceprivate RedisUtil redisUtil;@Lazy@Resourceprivate AllinPayService allinPayService;@Lazy@DubboReferenceprivate InspectPayScheduleRpcServiceI inspectPayScheduleRpcService;@Overrideprotected void registerService() {SERVICES.put(AsyncExecuteTypeEnums.DELAY_CHECK_PAYMENT_RESULT, this);}@Overrideprotected Boolean bizVerify(String paymentBillId) {PaymentBills paymentBills = paymentBillsService.getById(Long.valueOf(paymentBillId));if (Objects.isNull(paymentBills)) {log.error("支付订单id：{} 不存在", paymentBillId);return false;}return true;}@Overrideprotected void asyncExcute(String paymentBillId) {log.info("支付订单id：{} 开始执行订单超时处理", paymentBillId);log.info("当前秒数：{}", LocalDateTime.now().getSecond());PaymentBills paymentBills = paymentBillsService.getById(Long.valueOf(paymentBillId));// 删除redis缓存的keyLong businessOrderNo = paymentBills.getBusinessOrderNo();if(redisUtil.hasKey("order:"+businessOrderNo)) {redisUtil.delete("order:"+businessOrderNo);}if (!paymentBills.getPaymentBillStatus().equals(AgentCollectStatusEnum.COLLECTING.getStatus())) {log.info("支付订单id：{} 已支付或已进行超时处理", paymentBillId);return;}// 将支付中的订单/计划单/支付单/支付请求的状态修改为超时// 订单paymentBills.setPaymentBillStatus(AgentCollectStatusEnum.COLLECT_TIMEOUT.getStatus());paymentBillsService.updateById(paymentBills);// 计划单List<BillsPlan> billsPlans = billsPlanService.list(Wrappers.<BillsPlan>lambdaQuery().eq(BillsPlan::getPaymentBillId, paymentBillId).eq(BillsPlan::getPaymentBillStatus, AgentCollectStatusEnum.COLLECTING.getStatus()));if(CollectionUtils.isNotEmpty(billsPlans)) {billsPlans.forEach(billsPlan -> {billsPlan.setPaymentBillStatus(AgentCollectStatusEnum.COLLECT_TIMEOUT.getStatus());});billsPlanService.updateBatchById(billsPlans);paymentRequestService.update(Wrappers.<PaymentRequest>lambdaUpdate().in(PaymentRequest::getPaymentPlanId, billsPlans.stream().map(BillsPlan::getPaymentPlanId).toList()).eq(PaymentRequest::getPaymentStatus, AgentCollectStatusEnum.COLLECTING.getStatus()).set(PaymentRequest::getPaymentStatus, AgentCollectStatusEnum.COLLECT_TIMEOUT.getStatus()));}// 支付单paymentThirdBillsService.update(Wrappers.<PaymentThirdBills>lambdaUpdate().eq(PaymentThirdBills::getPaymentBillId, paymentBillId).eq(PaymentThirdBills::getPaymentBillStatus, AgentCollectStatusEnum.COLLECTING.getStatus()).set(PaymentThirdBills::getPaymentBillStatus, AgentCollectStatusEnum.COLLECT_TIMEOUT.getStatus()));// 将B3的支付状态修改为支付超时// 支付请求单
//        // 调用第三方接口将支付中的请求单关闭
//        paymentRequestService.list(Wrappers.<PaymentRequest>lambdaQuery()
//                .in(PaymentRequest::getPaymentPlanId, billsPlans.stream().map(BillsPlan::getPaymentPlanId).toList())
//                .eq(PaymentRequest::getPaymentStatus, AgentCollectStatusEnum.COLLECTING.getStatus()))
//                .forEach(paymentRequest -> {
//                    try {
//                        JSONObject object = allinPayService.closeOrder(paymentRequest.getPaymentRequestId() + "");
//                        log.info("支付请求id：{} 关闭结果：{}", paymentRequest.getPaymentRequestId(), object);
//                    }catch (Exception e){
//                        log.error("支付请求id：{} 关闭失败", paymentRequest.getPaymentRequestId());
//                    }
//                });// 修改B3付款状态为待支付inspectPayScheduleRpcService.updateInspectPayScheduleStatus(paymentBills.getBusinessOrderNo(), 0);}
}

总结

以上就是spring cloud stream 集成rocketmq的全部，像使用事务消息，获取其他可以继续看看文档，写得还是比较好理解，同理的如果想集成kafka,rabbitMQ，也可以下载案例进行参考