spring cloud stream3.0 rocketmq替换rabbit全过程_rocketmq替换rabbitmq

1、依赖

implementation("org.springframework.cloud:spring-cloud-stream-binder-rabbit")
1

更改为：

implementation("org.springframework.cloud:spring-cloud-stream-binder-rocketmq:0.9.0.RELEASE")
1

或者maven:

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-rocketmq</artifactId>
    <version>0.9.0.RELEASE</version>
</dependency>
1
2
3
4
5

可以去选择自己所需要的版本：https://mvnrepository.com/artifact/org.springframework.cloud/spring-cloud-stream-binder-rocketmq

2、服务配置

spring:
  rabbitmq:
    host: rabbitmqhost.com
1
2
3

更换为：

spring:
  cloud:
    stream:
      rocketmq:
        binder:
          name-server: rocketmq1.com:9876;rocketmq2.com:9876
1
2
3
4
5
6

3、topic配置

spring:
  cloud:
	stream:
      defaultBinder: rabbit
      binders:
        rabbit:
          type: rabbit
      bindings:
        test-in-0:
          destination: test
          group: test.group
      rabbit:
        bindings:
          test-in-0:
            consumer:
              autoBindDlq: true
              deadLetterQueueName: system.global-dlq.system
    function:
      definition: test
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

更换为：

spring:
  cloud:
	stream:
      defaultBinder: rocketmq
      binders:
        rocketmq:
          type: rocketmq
      bindings:
        test-in-0:
          destination: test
          # rocketmq 不能使用. 只能使用英文和横线、下划线
          group: test_group
          # 可以设置spring.cloud.stream 自带的重试为1，表示禁止重试
          # 我们使用rocketmq的重试策略就好
          consumer:
            max-attempts: 1
    function:
      definition: test
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

1、rocketmq的topic只支持 名命为：[%|a-zA-Z0-9_-]+$，大小写英文、数字、下划线、英文横线。不支持其他包括 “.”；
2、一个消费者订阅多个Topic不能使用同一个group，因此这里命名为Topic name + group
3、spring.cloud.stream.bindings.(channelName).consumer.max-attempts=1 表示禁用spring-cloud-stream框架重试，因此项目中使用mq本身重试，默认为16次，也可以设置spring.cloud.stream.rocketmq.bindings.(channelName).consumer.delayLevelWhenNextConsume = -1禁用mq端重试
4、其他配置可参考官方：https://github.com/alibaba/spring-cloud-alibaba/wiki/RocketMQ-en

5、单元测试修改

由于一个消费者订阅多个Topic不能使用同一个group，多个单元测试一起运行时会报错：

Caused by: org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'org.springframework.rocketmq.spring.starter.internalRocketMQTransAnnotationProcessor' defined in class path resource [org/springframework/cloud/stream/binder/rocketmq/config/RocketMQComponent4BinderAutoConfiguration.class]: Unsatisfied dependency expressed through method 'transactionAnnotationProcessor' parameter 0; nested exception is org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'transactionHandlerRegistry' defined in class path resource [org/springframework/cloud/stream/binder/rocketmq/config/RocketMQComponent4BinderAutoConfiguration.class]: Unsatisfied dependency expressed through method 'transactionHandlerRegistry' parameter 0; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'rocketMQTemplate' defined in class path resource [org/springframework/cloud/stream/binder/rocketmq/config/RocketMQComponent4BinderAutoConfiguration.class]: Invocation of init method failed; nested exception is org.apache.rocketmq.client.exception.MQClientException: The producer group[rocketmq_binder_default_group_name] has been created before, specify another name please.
1

需要添加注解@DirtiesContext 来解决：

@DirtiesContext ：标记为污染ApplicationContext环境，运行完毕后将被移除，新的springbootTest会重新构建环境
有关此注解的可以参考：https://docs.spring.io/spring-framework/docs/5.0.8.RELEASE/spring-framework-reference/testing.html#dirtiescontext