此版本仍在开发中，尚未被视为稳定版本。对于最新的稳定版本，请使用 Spring AMQP 3.2.0！spring-doc.cn

使用 RabbitMQ Stream 插件

版本 2.4 引入了对 RabbitMQ Stream Plugin 的 Java 客户端的初始支持。spring-doc.cn

RabbitStreamTemplatespring-doc.cn
StreamListenerContainerspring-doc.cn

将依赖项添加到您的项目中：spring-rabbit-streamspring-doc.cn

maven

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

Gradle

compile 'org.springframework.amqp:spring-rabbit-stream:3.2.1-SNAPSHOT'

您可以像往常一样预置队列，使用 Bean 使用该方法指定队列类型。例如：RabbitAdminQueueBuilder.stream()spring-doc.cn

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

但是，这仅在您还使用非流组件（例如 or ）时才有效，因为在打开 AMQP 连接时，会触发 admin 声明定义的 bean。如果您的应用程序仅使用流组件，或者您希望使用高级流配置功能，则应改为配置：SimpleMessageListenerContainerDirectMessageListenerContainerStreamAdminspring-doc.cn

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

请参阅 RabbitMQ 文档，以了解有关 .StreamCreatorspring-doc.cn

发送消息

它提供了（AMQP）功能的子集。RabbitStreamTemplateRabbitTemplatespring-doc.cn

RabbitStream操作

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;

}

该实现具有以下构造函数和属性：RabbitStreamTemplatespring-doc.cn

RabbitStreamTemplate

public RabbitStreamTemplate(Environment environment, String streamName) {
}

public void setMessageConverter(MessageConverter messageConverter) {
}

public void setStreamConverter(StreamMessageConverter streamConverter) {
}

public synchronized void setProducerCustomizer(ProducerCustomizer producerCustomizer) {
}

在方法中用于将对象转换为 Spring AMQP 。MessageConverterconvertAndSendMessagespring-doc.cn

这用于从 Spring AMQP 转换为 native stream 。StreamMessageConverterMessageMessagespring-doc.cn

您也可以直接发送本机流 s;使用该方法提供对消息生成器的访问。MessagemessageBuilder()Producerspring-doc.cn

这提供了一种在构建 producer 之前对其进行自定义的机制。ProducerCustomizerspring-doc.cn

请参阅 Java 客户端文档，了解如何自定义和。EnvironmentProducerspring-doc.cn

接收消息

异步消息接收由（以及使用时）提供。StreamListenerContainerStreamRabbitListenerContainerFactory@RabbitListenerspring-doc.cn

侦听器容器需要一个流名称以及一个流名称。Environmentspring-doc.cn

您可以使用经典接收 Spring AMQP ，也可以使用新接口接收本机流：MessageMessageListenerMessagespring-doc.cn

public interface StreamMessageListener extends MessageListener {

	void onStreamMessage(Message message, Context context);

}

有关支持的属性的信息，请参阅 Message Listener Container Configuration 。spring-doc.cn

与模板类似，容器具有 property。ConsumerCustomizerspring-doc.cn

请参阅 Java 客户端文档，了解如何自定义和。EnvironmentConsumerspring-doc.cn

使用时，请配置 ;此时，大多数属性（等）将被忽略。仅支持、和。此外，只能包含一个流名称。@RabbitListenerStreamRabbitListenerContainerFactory@RabbitListenerconcurrencyidqueuesautoStartupcontainerFactoryqueuesspring-doc.cn

例子

@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 将属性添加到（及其工厂）中。还提供了一个新的工厂 Bean 来创建无状态重试拦截器，该拦截器在使用原始流消息时具有可选的使用。adviceChainStreamListenerContainerStreamMessageRecovererspring-doc.cn

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

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

超级流

Super Stream 是分区流的抽象概念，通过将多个流队列绑定到具有参数的 exchange 来实现。x-super-stream: truespring-doc.cn

供应

为方便起见，可以通过定义 type 为的单个 bean 来供应 super 流。SuperStreamspring-doc.cn

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

检测到此 bean 并将声明 exchange （）和 3 个队列（分区） - 其中是、、、绑定的路由键等于。RabbitAdminmy.super.streammy.super-stream-nn012nspring-doc.cn

如果您还希望通过 AMQP 向 Exchange 发布信息，则可以提供自定义路由密钥：spring-doc.cn

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

键的数量必须等于分区的数量。spring-doc.cn

使用 SuperStream 进行制作

您必须将 a 添加到：superStreamRoutingFunctionRabbitStreamTemplatespring-doc.cn

@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
    });
    return template;
}

您还可以使用 .RabbitTemplatespring-doc.cn

使用具有单个活跃使用者的 Super Streams

在侦听器容器上调用该方法，在超级流上启用单个活跃的 consumer。superStreamspring-doc.cn

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

此时，当并发数大于 1 时，实际并发数由 ;要实现完全并发，请将环境设置为 1。请参阅配置环境。EnvironmentmaxConsumersByConnection

千分尺观察

从版本 3.0.5 开始，现在支持对和流侦听器容器使用 Micrometer 进行观察。该容器现在还支持 Micrometer 计时器（未启用观察时）。RabbitStreamTemplatespring-doc.cn

在每个组件上设置以启用观察;这将禁用 Micrometer Timers，因为计时器现在将随每个观测一起管理。使用带注释的侦听器时，请在容器工厂上设置。observationEnabledobservationEnabledspring-doc.cn

有关更多信息，请参阅 Micrometer Tracing 。spring-doc.cn

要向计时器/跟踪添加标签，请分别配置自定义或模板或侦听器容器。RabbitStreamTemplateObservationConventionRabbitStreamListenerObservationConventionspring-doc.cn

默认实施会添加 template observations 的标签和 containers 的标签。namelistener.idspring-doc.cn

您可以子类化 or or 提供全新的实现。DefaultRabbitStreamTemplateObservationConventionDefaultStreamRabbitListenerObservationConventionspring-doc.cn

有关更多详细信息，请参阅千分尺观测文档。spring-doc.cn