跳到主要内容
版本：4.1.1

使用 RabbitMQ Stream Plugin

Doubao 中英对照 Using the RabbitMQ Stream Plugin

2.4 版本引入了对用于 RabbitMQ Stream Plugin 的 RabbitMQ Stream Plugin Java Client 的初步支持。

  • RabbitStreamTemplate
  • StreamListenerContainer

向你的项目添加 spring-rabbit-stream 依赖：

<dependency>
  <groupId>org.springframework.amqp</groupId>
  <artifactId>spring-rabbit-stream</artifactId>
  <version>4.1.1</version>
</dependency>

你可以像平常那样配置队列：使用 RabbitAdmin bean，并使用 QueueBuilder.stream() 方法指定队列类型。例如：

@Bean
Queue stream() {
    return QueueBuilder.durable("stream.queue1")
            .stream()
            .build();
}

不过，这只有在你同时使用非 stream 组件（例如 SimpleMessageListenerContainer 或 DirectMessageListenerContainer）时才有效，因为 admin 是在打开 AMQP 连接时被触发去声明已定义的 bean。如果你的应用只使用 stream 组件，或者你希望使用高级 stream 配置特性，则应改为配置 StreamAdmin：

@Bean
StreamAdmin streamAdmin(Environment env) {
    return new StreamAdmin(env, sc -> {
        sc.stream("stream.queue1").maxAge(Duration.ofHours(2)).create();
        sc.stream("stream.queue2").create();
    });
}

关于 StreamCreator 的更多信息，参见 RabbitMQ 文档。

发送消息​

RabbitStreamTemplate 提供了 RabbitTemplate（AMQP）功能的一个子集。

public interface RabbitStreamOperations extends AutoCloseable {

	CompletableFuture<Boolean> send(Message message);

	CompletableFuture<Boolean> convertAndSend(Object message);

	CompletableFuture<Boolean> convertAndSend(Object message, @Nullable MessagePostProcessor mpp);

	CompletableFuture<Boolean> send(com.rabbitmq.stream.Message message);

	MessageBuilder messageBuilder();

	MessageConverter messageConverter();

	StreamMessageConverter streamMessageConverter();

	@Override
	void close() throws AmqpException;

}

RabbitStreamTemplate 实现具有以下构造器和属性：

public RabbitStreamTemplate(Environment environment, String streamName) {
}

public void setMessageConverter(MessageConverter messageConverter) {
}

public void setStreamConverter(StreamMessageConverter streamConverter) {
}

public void setProducerCustomizer(ProducerCustomizer producerCustomizer) {
}

MessageConverter 在 convertAndSend 方法中用于把对象转换为 Spring AMQP Message。

StreamMessageConverter 用于把 Spring AMQP Message 转换为原生 stream Message。

你也可以直接发送原生 stream Message；messageBuilder() 方法提供对 Producer 的消息 builder 的访问。

ProducerCustomizer 提供了一种在 producer 构建之前对其进行自定义的机制。

关于如何自定义 Environment 和 Producer，参见 Java Client 文档。

接收消息​

异步消息接收由 StreamListenerContainer 提供（使用 @RabbitListener 时则由 StreamRabbitListenerContainerFactory 提供）。

listener 容器需要一个 Environment 以及单个 stream 名称。

你既可以用经典的 MessageListener 接收 Spring AMQP Message，也可以用一个新接口接收原生 stream Message：

public interface StreamMessageListener extends MessageListener {

	void onStreamMessage(Message message, Context context);

}

关于支持的属性，参见消息 Listener 容器配置。

与 template 类似，容器有一个 ConsumerCustomizer 属性。

关于如何自定义 Environment 和 Consumer，参见 Java Client 文档。

使用 @RabbitListener 时，配置一个 StreamRabbitListenerContainerFactory；此时，大多数 @RabbitListener 属性（concurrency 等）会被忽略。只支持 id、queues、autoStartup 和 containerFactory。此外，queues 只能包含一个 stream 名称。

示例​

@Bean
RabbitStreamTemplate streamTemplate(Environment env) {
    RabbitStreamTemplate template = new RabbitStreamTemplate(env, "test.stream.queue1");
    template.setProducerCustomizer((name, builder) -> builder.name("test"));
    return template;
}

@Bean
RabbitListenerContainerFactory<StreamListenerContainer> rabbitListenerContainerFactory(Environment env) {
    return new StreamRabbitListenerContainerFactory(env);
}

@RabbitListener(queues = "test.stream.queue1")
void listen(String in) {
    ...
}

@Bean
RabbitListenerContainerFactory<StreamListenerContainer> nativeFactory(Environment env) {
    StreamRabbitListenerContainerFactory factory = new StreamRabbitListenerContainerFactory(env);
    factory.setNativeListener(true);
    factory.setConsumerCustomizer((id, builder) -> {
        builder.name("myConsumer")
                .offset(OffsetSpecification.first())
                .manualTrackingStrategy();
    });
    return factory;
}

@RabbitListener(id = "test", queues = "test.stream.queue2", containerFactory = "nativeFactory")
void nativeMsg(Message in, Context context) {
    ...
    context.storeOffset();
}

@Bean
Queue stream() {
    return QueueBuilder.durable("test.stream.queue1")
            .stream()
            .build();
}

@Bean
Queue stream() {
    return QueueBuilder.durable("test.stream.queue2")
            .stream()
            .build();
}

2.4.5 版本为 StreamListenerContainer（及其工厂）添加了 adviceChain 属性。还提供了一个新的工厂 bean，用于创建带可选 StreamMessageRecoverer 的无状态重试拦截器，以便在消费原始 stream 消息时使用。

@Bean
public StreamRetryOperationsInterceptorFactoryBean sfb(RetryTemplate retryTemplate) {
    StreamRetryOperationsInterceptorFactoryBean rfb =
            new StreamRetryOperationsInterceptorFactoryBean();
    rfb.setRetryOperations(retryTemplate);
    rfb.setStreamMessageRecoverer((msg, context, throwable) -> {
        ...
    });
    return rfb;
}
important

此容器不支持有状态重试。

Super Streams（超级流）​

Super Stream 是分区 stream 的抽象概念，通过把多个 stream 队列绑定到一个带有参数 x-super-stream: true 的 exchange 来实现。

配置供应​

为方便起见，可以通过定义单个 SuperStream 类型的 bean 来供应一个 super stream。

@Bean
SuperStream superStream() {
    return new SuperStream("my.super.stream", 3);
}

RabbitAdmin 会检测到此 bean，并声明该 exchange（my.super.stream）和 3 个队列（分区）——my.super-stream-n，其中 n 为 0、1、2，并以等于 n 的 routing key 绑定。

如果你还希望通过 AMQP 向该 exchange 发布消息，可以提供自定义 routing key：

@Bean
SuperStream superStream() {
    return new SuperStream("my.super.stream", 3, (q, i) -> IntStream.range(0, i)
					.mapToObj(j -> "rk-" + j)
					.collect(Collectors.toList()));
}

key 的数量必须等于分区的数量。

向 SuperStream 生产消息​

你必须向 RabbitStreamTemplate 添加一个 superStreamRoutingFunction：

@Bean
RabbitStreamTemplate streamTemplate(Environment env) {
    RabbitStreamTemplate template = new RabbitStreamTemplate(env, "stream.queue1");
    template.setSuperStreamRouting(message -> {
        // some logic to return a String for the client's hashing algorithm, e.g.
        return message.getApplicationProperties().get("ROUTING_KEY").toString();
    });
    return template;
}

以下是使用 RabbitStreamTemplate 发送消息的示例：

//Send a message with the routing property for the hashing algorithm
rabbitStreamTemplate.send(
    rabbitStreamTemplate.messageBuilder()
        .addData(converter.convert(domainObject))
        .applicationProperties()
        .entry("ROUTING_KEY", domainObject.getId())
        .messageBuilder()
        .build());

你也可以使用 RabbitTemplate 通过 AMQP 发布消息。

用单 active consumer 消费 Super Streams​

在 listener 容器上调用 superStream 方法，即可在 super stream 上启用单 active consumer。

@Bean
StreamListenerContainer container(Environment env, String name) {
    StreamListenerContainer container = new StreamListenerContainer(env);
    container.superStream("ss.sac", "myConsumer", 3); // concurrency = 3
    container.setupMessageListener(msg -> {
        ...
    });
    container.setConsumerCustomizer((id, builder) -> builder.offset(OffsetSpecification.last()));
    return container;
}
important

此时，当并发大于 1 时，实际并发还会受 Environment 的进一步控制；要达到完全并发，请把环境的 maxConsumersByConnection 设为 1。参见配置 Environment。

Micrometer 观测​

从 3.0.5 版本开始，RabbitStreamTemplate 和 stream listener 容器现在支持使用 Micrometer 进行观测。容器现在还支持 Micrometer timer（在未启用观测时）。

在每个组件上设置 observationEnabled 以启用观测；这会禁用 Micrometer Timer，因为这些 timer 现在会随每次观测一起管理。使用注解 listener 时，在容器工厂上设置 observationEnabled。

更多信息参见 Micrometer Tracing。

要为 timer/trace 添加标签，请分别为 template 或 listener 容器配置自定义的 RabbitStreamTemplateObservationConvention 或 RabbitStreamListenerObservationConvention。

默认实现为 template 观测添加 name 标签，为容器添加 listener.id 标签。

你既可以子类化 DefaultRabbitStreamTemplateObservationConvention 或 DefaultStreamRabbitListenerObservationConvention，也可以提供全新的实现。

更多细节参见 Micrometer 观测文档。