<dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
            <version>2021.0.5.0</version>
        </dependency>

spring:
  application:
    # 应用名称
    name: jnpf-ftb
  cloud:
    stream:
      rocketmq:
        binder:
          name-server: ${ROCKETMQ_HOST:192.168.3.22:30094}
        bindings:
          output:
            producer:
              sync: true
              group: jnpf-group1
      bindings:
        permission-output:  #生产
          content-type: text/json
          destination: permission-topic
          group: jnpf-group1
        permission-input: #消费
          content-type: text/json
          destination: permission-topic
          group: jnpf-ftb-consumer

package jnpf.qualifications.consummer;

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;

public interface PositionGradeConsumerSource {
    String OUTPUT = "permission-output";
    String INPUT = "permission-input";
    @Output(OUTPUT)
    MessageChannel output();
    @Input(INPUT)
    SubscribableChannel input();
}

package jnpf.qualifications.consummer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.stereotype.Component;
import org.springframework.messaging.Message;

@Slf4j
@Component
@EnableBinding(PositionGradeConsumerSource.class)
public class PositionGradeConsumer {
    @StreamListener(target = PositionGradeConsumerSource.INPUT,condition = "headers['ROCKET_TAGS'] == 'TAG_GRADE'")
    public void receive(Message<String> message) {
        // 获取消息体
        String payload = message.getPayload();

        // 获取 headers
        var headers = message.getHeaders();

        log.error("接收到消息：{}，Tags={}", payload, tags);
    }
}