Spring Cloud Stream集成RocketMQ（kafka/rabbitMQ通用）_spring cloud stream rocketmq

什么是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顺序消费）

@SpringBootApplicationpublic 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: 28082spring: application: name: rocketmq-orderly-consume-example cloud: stream: function: definition: consumer; rocketmq: binder: name-server: localhost:9876 bindings: producer-out-0: producer:  group: output_1  messageQueueSelector: orderlyMessageQueueSelector consumer-in-0: consumer:  # tag: {@code tag1||tag2||tag3 }; sql: {@code \'color\'=\'blue\' AND \'price\'>100 } .  subscription: \'TagA || TagC || TagD\'  push: orderly: true bindings: producer-out-0: destination: orderly consumer-in-0: destination: orderly group: 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: orderly group: orderly-consumer //消费者组:消费者组名为 orderly-consumer，相同组内消费者分摊消费队列，不同组独立消费。

producer

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

@Beanpublic 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 和 headers Message<SimpleMsg> msg = new GenericMessage<>(new SimpleMsg(\"Hello RocketMQ \" + i), headers); // 发送消息到名为 \"producer-out-0\" 的输出通道 streamBridge.send(\"producer-out-0\", msg); } };}

selector（供生产者使用）

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

@Componentpublic 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:TagAThread-0 Receive: Hello RocketMQ 4 TAG:TagEThread-0 Receive: Hello RocketMQ 5 TAG:TagAThread-0 Receive: Hello RocketMQ 9 TAG:TagE...（后续i=10,14,15...）

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

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

Thread-3 Receive: Hello RocketMQ 3 TAG:TagDThread-3 Receive: Hello RocketMQ 8 TAG:TagDThread-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. 将订单数据假如延迟队列

 @Override public 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@Slf4jpublic class RocketMqProducer { @Resource private 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-0spring.cloud.stream.bindings.paymentProducer-out-0.binder=rocketmqspring.cloud.stream.bindings.paymentProducer-out-0.content-type=application/jsonspring.cloud.stream.bindings.paymentProducer-out-0.destination=payment-topicspring.cloud.stream.paymentTransactionProducer=paymentTransactionProducer-out-0spring.cloud.stream.bindings.paymentTransactionProducer-out-0.binder=rocketmqspring.cloud.stream.bindings.paymentTransactionProducer-out-0.content-type=application/jsonspring.cloud.stream.bindings.paymentTransactionProducer-out-0.destination=payment-topicspring.cloud.stream.rocketmq.bindings.paymentTransactionProducer-out-0.producer.producerType=Transspring.cloud.stream.rocketmq.bindings.paymentTransactionProducer-out-0.producer.transactionListener=RocketMqTransactionListener# 定义 通道 为 paymentConsumer 的 消费者，tags 定义只接受 payment和all 消息spring.cloud.stream.bindings.paymentConsumer-in-0.binder=rocketmqspring.cloud.stream.bindings.paymentConsumer-in-0.content-type=application/jsonspring.cloud.stream.bindings.paymentConsumer-in-0.destination=payment-topicspring.cloud.stream.bindings.paymentConsumer-in-0.group=payment-customer-groupspring.cloud.stream.rocketmq.bindings.paymentConsumer-in-0.consumer.group=payment-customer-groupspring.cloud.stream.rocketmq.bindings.paymentConsumer-in-0.consumer.subscription=payment||allspring.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@Configurationpublic class AsyncExecuteConsumer { @Value(\"${payment.asyncMsg.maxRetryCount}\") private Integer maxRetryCount; @Resource private AsyncRetryInfoMapper asyncRetryInfoMapper; @Resource private TransactionTemplate transactionTemplate; @Bean public 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个类型 入账，支付，提现

@Getterpublic 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; }}

@Slf4jpublic 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。 @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 */ @PostConstruct protected abstract void registerService(); /** * 验证业务上的事务是否提交 * * @param dto 参数 */ protected abstract Boolean bizVerify(T dto); /** * 执行核心业务 * * @param dto 参数 */ protected abstract void asyncExcute(T dto);}

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

@Slf4j@Servicepublic class PaymentBillsDelayConsumer extends AsyncExecuteService<String> { @Lazy @Resource private PaymentBillsService paymentBillsService; @Lazy @Resource private BillsPlanService billsPlanService; @Lazy @Resource private PaymentThirdBillsService paymentThirdBillsService; @Lazy @Resource private PaymentRequestService paymentRequestService; @Lazy @Resource private RedisUtil redisUtil; @Lazy @Resource private AllinPayService allinPayService; @Lazy @DubboReference private InspectPayScheduleRpcServiceI inspectPayScheduleRpcService; @Override protected void registerService() { SERVICES.put(AsyncExecuteTypeEnums.DELAY_CHECK_PAYMENT_RESULT, this); } @Override protected Boolean bizVerify(String paymentBillId) { PaymentBills paymentBills = paymentBillsService.getById(Long.valueOf(paymentBillId)); if (Objects.isNull(paymentBills)) { log.error(\"支付订单id：{} 不存在\", paymentBillId); return false; } return true; } @Override protected void asyncExcute(String paymentBillId) { log.info(\"支付订单id：{} 开始执行订单超时处理\", paymentBillId); log.info(\"当前秒数：{}\", LocalDateTime.now().getSecond()); PaymentBills paymentBills = paymentBillsService.getById(Long.valueOf(paymentBillId)); // 删除redis缓存的key Long 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.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，也可以下载案例进行参考