RabbitMQ Stream插件使用案例代码

更新时间：2024年04月17日 11:49:49 作者：Doker技术人的品牌

这篇文章主要介绍了RabbitMQ Stream插件使用案例代码,2.4版为RabbitMQ流插件引入了对RabbitMQStream插件Java客户端的初始支持,需要的朋友可以参考下

然而，这仅在您还使用non-stream 组件（如SimpleMessageListenerContainer或DirectMessageListeneerContainer）时才有效，因为在打开AMQP连接时会触发管理员来声明定义的bean。如果您的应用程序仅使用流组件，或者您希望使用高级流配置功能，则应改为配置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();
    });
}

一、Sending Messages

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 synchronized void setProducerCustomizer(ProducerCustomizer producerCustomizer) {
}

MessageConverter在convertAndSend方法中用于将对象转换为Spring AMQP消息。

StreamMessageConverter用于将Spring AMQP消息转换为本机流消息。

您也可以直接发送本机流消息；使用messageBuilder（）方法提供对生产者的消息生成器的访问。

ProducerCustomizer提供了一种机制，用于在生成生产者之前对其进行自定义。

二、Receiving Messages

异步消息接收由StreamListenerContainer（以及使用@RabbitListener时的StreamRabbitListerContainerFactory）提供。

侦听器容器需要一个Environment以及一个流名称。

您可以使用经典的MessageListener接收Spring AMQP消息，也可以使用新接口接收本地流消息：

public interface StreamMessageListener extends MessageListener {
	void onStreamMessage(Message message, Context context);
}

有关支持的属性的信息，请参阅消息侦听器容器配置。

与模板类似，容器具有ConsumerCustomizer属性。

有关自定义环境和使用者的信息，请参阅Java客户端文档。

使用@RabbitListener时，配置StreamRabbitListerContainerFactory；此时，大多数@RabbitListener属性（并发等）将被忽略。仅支持id、队列、autoStartup和containerFactory。此外，队列只能包含一个流名称。

三、Examples

@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版将adviceChain属性添加到StreamListenerContainer（及其工厂）。还提供了一个新的工厂bean来创建一个无状态重试拦截器，该拦截器带有一个可选的StreamMessageRecoverer，用于在使用原始流消息时使用。

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

四、Super Streams

超级流是分区流的抽象概念，通过将多个流队列绑定到具有参数x-Super-Stream:true的交换来实现。

1、调配

为了方便起见，可以通过定义类型为SuperStream的单个bean来提供超级流。

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

RabbitAdmin检测到这个bean，并将声明交换（my.super.stream）和3个队列（分区）-my.super-stream-n，其中n是0，1，2，绑定的路由密钥等于n。

如果您还希望通过AMQP向exchange 发布，您可以提供自定义路由密钥：

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

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

2、向超级流生产消息

你必须向 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
    });
    return template;
}

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

到此这篇关于RabbitMQ Stream插件使用详解的文章就介绍到这了,更多相关RabbitMQ Stream插件内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

如何解决java压缩文件乱码问题
在本篇文章中我们给大家分享的是一篇关于java压缩文件乱码问题的解决办法，有需要的朋友们可以学习下。
2019-12-12
java生成json数据示例
这篇文章主要介绍了java生成json数据示例,需要的朋友可以参考下
2014-04-04
10本Java架构师必读书籍
这篇文章主要介绍了10本Java架构师必读书籍，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2017-08-08
SpringBoot深入浅出分析初始化器
这篇文章主要介绍了SpringBoot初始化器的分析，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2022-07-07
Java 匿名对象与匿名内部类的使用
很多小伙伴对匿名对象和匿名内部类的写法有点陌生,本文主要介绍了Java 匿名对象与匿名内部类的使用,具有一定的参考价值,感兴趣的可以了解一下
2024-01-01
Java的Hibernate框架中集合类数据结构的映射编写教程
Hibernate可以将Java中几个内置的集合结构映射为数据库使用的关系模型,下面我们就来看一下Java的Hibernate框架中集合类数据结构的映射编写教程:
2016-07-07
JVM垃圾回收原理解析
这篇文章主要介绍了JVM垃圾回收原理解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2019-08-08
Java判断主机是否能ping通代码实例
这篇文章主要介绍了Java判断主机是否能ping通代码实例,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2020-01-01
Kotlin 标准函数和静态方法示例详解
这篇文章主要为大家介绍了Kotlin 标准函数和静态方法示例详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪
2022-10-10
史上最全的java随机数生成算法分享
这篇文章主要介绍了史上最全的java随机数生成算法,我分享一个最全的随机数的生成算法,最代码的找回密码的随机数就是用的这个方法
2014-01-01