操纵018：Stream Queue

操纵018：Stream Queue

一、启用插件

   说明：只有启用了Stream插件，才华使用流式队列的完整功能
  在集群每个节点中依次执行如下操纵：

1. # 启用Stream插件
2. rabbitmq-plugins enable rabbitmq_stream
4. # 重启rabbit应用
5. rabbitmqctl stop_app
6. rabbitmqctl start_app
8. # 查看插件状态
9. rabbitmq-plugins list

复制代码

二、负载平衡

在文件/etc/haproxy/haproxy.cfg末了追加：

1. frontend rabbitmq_stream_frontend
2. bind 192.168.200.100:33333
3. mode tcp
4. default_backend rabbitmq_stream_backend
6. backend rabbitmq_stream_backend
7. mode tcp
8. balance roundrobin
9. server rabbitmq1 192.168.200.100:5552 check
10. server rabbitmq2 192.168.200.150:5552 check
11. server rabbitmq3 192.168.200.200:5552 check

复制代码

三、Java代码

1、引入依赖

Stream 专属 Java 客户端官方网址：https://github.com/rabbitmq/rabbitmq-stream-java-client

Stream 专属 Java 客户端官方文档网址：https://rabbitmq.github.io/rabbitmq-stream-java-client/stable/htmlsingle/

1. <dependencies>
2.     <dependency>
3.         <groupId>com.rabbitmq</groupId>
4.         <artifactId>stream-client</artifactId>
5.         <version>0.15.0</version>
6.     </dependency>
8.     <dependency>
9.         <groupId>org.slf4j</groupId>
10.         <artifactId>slf4j-api</artifactId>
11.         <version>1.7.30</version>
12.     </dependency>
13.     <dependency>
14.         <groupId>ch.qos.logback</groupId>
15.         <artifactId>logback-classic</artifactId>
16.         <version>1.2.3</version>
17.     </dependency>
18. </dependencies>

复制代码

2、创建Stream

   说明：不需要创建交换机
  ①代码方式创建

1. Environment environment = Environment.builder()
2.         .host("192.168.200.100")
3.         .port(33333)
4.         .username("atguigu")
5.         .password("123456")
6.         .build();
8. environment.streamCreator().stream("stream.atguigu.test2").create();
10. environment.close();

复制代码

②ManagementUI创建

3、生产者端步调

①内部机制说明

[1]官方文档

   Internally, the Environment will query the broker to find out about the topology of the stream and will create or re-use a connection to publish to the leader node of the stream.
  翻译：
   在内部，Environment将查询broker以了解流的拓扑结构，并将创建或重用毗连以发布到流的 leader 节点。
  [2]剖析

• 在 Environment 中封装的毗连信息仅负责毗连到 broker
  
• Producer 在构建对象时会访问 broker 拉取集群中 Leader 的毗连信息
  
• 未来实际访问的是集群中的 Leader 节点
  
• Leader 的毗连信息格式是：节点名称:端口号
  

[3]配置

为了让本机的应用步调知道 Leader 节点名称对应的 IP 地址，我们需要在本地配置 hosts 文件，创建从节点名称到 IP 地址的映射关系

②示例代码

1. Environment environment = Environment.builder()
2.         .host("192.168.200.100")
3.         .port(33333)
4.         .username("atguigu")
5.         .password("123456")
6.         .build();
8. Producer producer = environment.producerBuilder()
9.         .stream("stream.atguigu.test")
10.         .build();
12. byte[] messagePayload = "hello rabbit stream".getBytes(StandardCharsets.UTF_8);
14. CountDownLatch countDownLatch = new CountDownLatch(1);
16. producer.send(
17.         producer.messageBuilder().addData(messagePayload).build(),
18.         confirmationStatus -> {
19.             if (confirmationStatus.isConfirmed()) {
20.                 System.out.println("[生产者端]the message made it to the broker");
21.             } else {
22.                 System.out.println("[生产者端]the message did not make it to the broker");
23.             }
25.             countDownLatch.countDown();
26.         });
28. countDownLatch.await();
30. producer.close();
32. environment.close();

复制代码

4、消耗端步调

1. Environment environment = Environment.builder()
2.         .host("192.168.200.100")
3.         .port(33333)
4.         .username("atguigu")
5.         .password("123456")
6.         .build();
8. environment.consumerBuilder()
9.         .stream("stream.atguigu.test")
10.         .name("stream.atguigu.test.consumer")
11.         .autoTrackingStrategy()
12.         .builder()
13.         .messageHandler((offset, message) -> {
14.             byte[] bodyAsBinary = message.getBodyAsBinary();
15.             String messageContent = new String(bodyAsBinary);
16.             System.out.println("[消费者端]messageContent = " + messageContent + " Offset=" + offset.offset());
17.         })
18.         .build();

复制代码

四、指定偏移量消耗

1、偏移量

2、官方文档说明

   The offset is the place in the stream where the consumer starts consuming from. The possible values for the offset parameter are the following:
  

• OffsetSpecification.first(): starting from the first available offset. If the stream has not been truncated, this means the beginning of the stream (offset 0).
  
• OffsetSpecification.last(): starting from the end of the stream and returning the last chunk of messages immediately (if the stream is not empty).
  
• OffsetSpecification.next(): starting from the next offset to be written. Contrary to OffsetSpecification.last(), consuming with OffsetSpecification.next() will not return anything if no-one is publishing to the stream. The broker will start sending messages to the consumer when messages are published to the stream.
  
• OffsetSpecification.offset(offset): starting from the specified offset. 0 means consuming from the beginning of the stream (first messages). The client can also specify any number, for example the offset where it left off in a previous incarnation of the application.
  
• OffsetSpecification.timestamp(timestamp): starting from the messages stored after the specified timestamp. Note consumers can receive messages published a bit before the specified timestamp. Application code can filter out those messages if necessary.
  

  3、指定Offset消耗

1. Environment environment = Environment.builder()
2.         .host("192.168.200.100")
3.         .port(33333)
4.         .username("atguigu")
5.         .password("123456")
6.         .build();
8. CountDownLatch countDownLatch = new CountDownLatch(1);
10. Consumer consumer = environment.consumerBuilder()
11.         .stream("stream.atguigu.test")
12.         .offset(OffsetSpecification.first())
13.         .messageHandler((offset, message) -> {
14.             byte[] bodyAsBinary = message.getBodyAsBinary();
15.             String messageContent = new String(bodyAsBinary);
16.             System.out.println("[消费者端]messageContent = " + messageContent);
17.             countDownLatch.countDown();
18.         })
19.         .build();
21. countDownLatch.await();
23. consumer.close();

复制代码

4、对比

• autoTrackingStrategy 方式：始终监听Stream中的新消息（狗狗看家，忠于职守）
  
• 指定偏移量方式：针对指定偏移量的消息消耗之后就制止（狗狗叼飞盘，叼回来就完）
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。