Spring cloud stream 3.1 rocketmq踩坑记录

Posted 2021-09-13 MY1024

spring cloud stream 新绑定方式

新版spring cloud stream文档

新版提倡用函数式进行发送和消费信息

定义返回类型为Supplier, Function or Consumer的bean提供消息发送和消费的bean

看看绑定名称命名规则

• input - <functionName> + -in- + <index>
• output - <functionName> + -out- + <index>

在配置文件中指定spring.cloud.function.definition的名称后会把这个bean绑定到对应的消费者和提供者上.

如下定义 会把bean绑定在消费者consumerEvent-in-0或者提供者consumerEvent-out-0上

多个bean可以用 ; 进行分割

spring:
  cloud:
    function:
      definition: consumerEvent

指定这个消费者的topic和group

spring:
  cloud:
    stream:
      bindings:
        consumerEvent-in-0:
          destination: DEMO
          group: demo-group

注册消费者的bean

// 第一种方式(官方推荐)
@Bean
public Function<Flux<Message<String>>, Mono<Void>> consumerEvent() {
    return flux -> flux.map(message -> {
        System.out.println(message.getPayload());
        return message;
    }).then();
}

// 第二种方式
// 注意使用Flux 要调用 subscribe 不然这个方法不会被消费
@Bean
public Consumer<Flux<Message<String>>> consumerEvent() {
    return flux -> flux.map(message -> {
        System.out.println(message.getPayload());
        return message;
    }).subscribe();
}
// 或
@Bean
public Consumer<Message<String>> consumerEvent() {
    return message -> System.out.println(message.getPayload());
}

示例

提供者

@Configuration
public class EventSender {
    @Bean
    public Demo demo() {
        return new Demo();
    }

    static class Demo implements CommandLineRunner {
        @Autowired
        StreamBridge streamBridge;

        @Override
        public void run(String... args) throws Exception {
            final Message<T> message = MessageBuilder.withPayload("Body")
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .build();

            // 第一个配置的是目的地
            // 如果在yaml中有配置会发送到yaml中目的地
            streamBridge.send("DEMO", message);
        }
    }
}

配置rocketmq和stream的配置

spring:
  application:
    name: demo
  cloud:
    stream:
      rocketmq:
        binder:
          name-server: 127.0.0.1:9876
          group: demo
      bindings:
        consumerEvent-in-0:
          destination: DEMO
          content-type: application/json
          group: demo-group
      function:
        definition: consumerEvent

注册一个消费者

@Configuration
public class EventReceptor {
    @Bean
    public Function<Flux<Message<String>>, Mono<Void>> consumerEvent() {
        return flux -> flux.map(message -> {
            System.out.println(message.getPayload());
            return message;
        }).then();
    }
}

依赖

spring cloud 2020 默认不使用bootstrap启动 要加这个依赖spring-cloud-starter-bootstrap

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-dependencies</artifactId>
    <version>2020.0.2</version>
    <type>pom</type>
    <scope>import</scope>
</dependency>
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-alibaba-dependencies</artifactId>
    <version>2.2.5-RocketMQ-RC1</version>
    <type>pom</type>
    <scope>import</scope>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-bootstrap</artifactId>
    <version>3.0.2</version>
</dependency>

Tag过滤

在新版的时候过滤tag一直失效, 后面看源码发现新版的sql和tag结合到subscription的属性中

this.pushConsumer.subscribe(this.topic, RocketMQUtils.getMessageSelector(((RocketMQConsumerProperties)this.extendedConsumerProperties.getExtension()).getSubscription()));

public static MessageSelector getMessageSelector(String expression) {
    return StringUtils.hasText(expression) && expression.startsWith("sql:") ? MessageSelector.bySql(expression.replaceFirst("sql:", "")) : MessageSelector.byTag(expression);
}

如果消费者要过滤某个tag需要这么写

// 新版 (现在的写法)
rocketmq:
  bindings:
    createUserAccountEvent-in-0:
      consumer:
        subscription: DEMO-TAG

// 旧版 (以前的写法)
rocketmq:
  bindings:
    createUserAccountEvent-in-0:
      consumer:
        tag: DEMO-TAG