ActiveMQ集群下的消息回流功能

2024-04-23 07:38

本文主要是介绍ActiveMQ集群下的消息回流功能,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

"丢失"的消息

  如果有broker1和broker2通过networkConnector连接,有一个consumer1连接到broker1,一个consumer2连接到broker2,程序往broker1上面发送30条消息,这时consumer2连接到broker2消费消息,当consumer2消费了15条消息时,broker2挂掉了。 但是还剩下15条消息在broker2上面,这些消息就好像消息了,除非broker2重启了,然后有消费者连接到broker2上来消费消息,遇到这样的情况该怎么办呢?

  从5.6版本起,在destinationPolicy上新增的选择replayWhenNoConsumers,这个选项使得broker2上有需要转发的消息但是没有消费者时,把消息回流到它原来的broker1上,同时需要把enableAudit设置为false,为了防止消息回流后被当做重复消息而不被分发,activemq.xml配置如下:

多个broker配置

<policyEntry queue=">" enableAudit="false">
   <networkBridgeFilterFactory>
     <conditionalNetworkBridgeFilterFactory replayWhenNoConsumers="true"/>
   </networkBridgeFilterFactory>
</policyEntry>
配置 ActiveMQ的静态网络链接
必须写上duplex="true"
<networkConnectors><networkConnector name="local network" duplex="true" uri="static://(tcp://ip:prot,tcp://ip:port)"/>
</networkConnectors>

配置允许双向连接

先运行发送消息程序,然后运行61716端口的程序接收消息,再接收了5条后将程序断掉,再去运行61616端口的程序,也是可以接收消息的,这样就实现了消息的回流。

消息发送程序:

复制代码
 1 import javax.jms.Connection;2 import javax.jms.ConnectionFactory;3 import javax.jms.Destination;4 import javax.jms.MessageProducer;5 import javax.jms.Session;6 import javax.jms.TextMessage;7 8 import org.apache.activemq.ActiveMQConnectionFactory;9 
10 public class JmsSend {
11     public static void main(String[] args) throws Exception {
12         ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.1.81:61616");
13         Connection connection = connectionFactory.createConnection();
14         connection.start();
15         
16         Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
17         Destination destination = session.createQueue("my-queue");
18         
19         MessageProducer producer = session.createProducer(destination);
20         for(int i = 0;i < 10;i++){
21             TextMessage message = session.createTextMessage("message,1212 --->" + i);
22             Thread.sleep(1000);
23             //通过生产者发出消息
24             producer.send(message);
25         }
26         session.commit();
27         session.close();
28         connection.close();
29     }
30 }
复制代码
复制代码

 61616端口接收消息程序:

复制代码
 1 import javax.jms.Connection;2 import javax.jms.ConnectionFactory;3 import javax.jms.Destination;4 import javax.jms.MessageConsumer;5 import javax.jms.Session;6 import javax.jms.TextMessage;7 8 import org.apache.activemq.ActiveMQConnectionFactory;9 
10 public class JmsReceiver {
11     public static void main(String[] args) throws Exception {
12         ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://192.168.1.81:61616");
13         Connection connection =  cf.createConnection();
14         connection.start();
15         
16         Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
17         Destination destination = session.createQueue("my-queue");
18         MessageConsumer consumer = session.createConsumer(destination);
19         int i = 0;
20         while(i < 10){
21             Thread.sleep(3000);
22             i++;
23             TextMessage message = (TextMessage)consumer.receive();
24             session.commit();
25             System.out.println("111接收到的消息是:"+message.getText());
26         }
27         session.close();
28         connection.close();
29     }
30 }
复制代码

 61716端口接收消息程序:

复制代码
 1 import javax.jms.Connection;2 import javax.jms.ConnectionFactory;3 import javax.jms.Destination;4 import javax.jms.MessageConsumer;5 import javax.jms.Session;6 import javax.jms.TextMessage;7 8 import org.apache.activemq.ActiveMQConnectionFactory;9 
10 public class JmsReceiver2 {
11     public static void main(String[] args) throws Exception {
12         ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://192.168.1.81:61716");
13         Connection connection =  cf.createConnection();
14         connection.start();
15         
16         Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
17         Destination destination = session.createQueue("my-queue");
18         MessageConsumer consumer = session.createConsumer(destination);
19         int i = 0;
20         while(i < 10){
21             Thread.sleep(3000);
22             i++;
23             TextMessage message = (TextMessage)consumer.receive();
24             session.commit();
25             System.out.println("222接收到的消息是:"+message.getText());
26         }
27         session.close();
28         connection.close();
29     }
30 }

这篇关于ActiveMQ集群下的消息回流功能的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/928178

相关文章

Java实现文件图片的预览和下载功能

《Java实现文件图片的预览和下载功能》这篇文章主要为大家详细介绍了如何使用Java实现文件图片的预览和下载功能,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... Java实现文件(图片)的预览和下载 @ApiOperation("访问文件") @GetMapping("

SpringKafka消息发布之KafkaTemplate与事务支持功能

《SpringKafka消息发布之KafkaTemplate与事务支持功能》通过本文介绍的基本用法、序列化选项、事务支持、错误处理和性能优化技术,开发者可以构建高效可靠的Kafka消息发布系统,事务支... 目录引言一、KafkaTemplate基础二、消息序列化三、事务支持机制四、错误处理与重试五、性能优

SpringIntegration消息路由之Router的条件路由与过滤功能

《SpringIntegration消息路由之Router的条件路由与过滤功能》本文详细介绍了Router的基础概念、条件路由实现、基于消息头的路由、动态路由与路由表、消息过滤与选择性路由以及错误处理... 目录引言一、Router基础概念二、条件路由实现三、基于消息头的路由四、动态路由与路由表五、消息过滤

Spring Boot 3.4.3 基于 Spring WebFlux 实现 SSE 功能(代码示例)

《SpringBoot3.4.3基于SpringWebFlux实现SSE功能(代码示例)》SpringBoot3.4.3结合SpringWebFlux实现SSE功能,为实时数据推送提供... 目录1. SSE 简介1.1 什么是 SSE?1.2 SSE 的优点1.3 适用场景2. Spring WebFlu

基于SpringBoot实现文件秒传功能

《基于SpringBoot实现文件秒传功能》在开发Web应用时,文件上传是一个常见需求,然而,当用户需要上传大文件或相同文件多次时,会造成带宽浪费和服务器存储冗余,此时可以使用文件秒传技术通过识别重复... 目录前言文件秒传原理代码实现1. 创建项目基础结构2. 创建上传存储代码3. 创建Result类4.

Python+PyQt5实现多屏幕协同播放功能

《Python+PyQt5实现多屏幕协同播放功能》在现代会议展示、数字广告、展览展示等场景中,多屏幕协同播放已成为刚需,下面我们就来看看如何利用Python和PyQt5开发一套功能强大的跨屏播控系统吧... 目录一、项目概述:突破传统播放限制二、核心技术解析2.1 多屏管理机制2.2 播放引擎设计2.3 专

一文详解SpringBoot响应压缩功能的配置与优化

《一文详解SpringBoot响应压缩功能的配置与优化》SpringBoot的响应压缩功能基于智能协商机制,需同时满足很多条件,本文主要为大家详细介绍了SpringBoot响应压缩功能的配置与优化,需... 目录一、核心工作机制1.1 自动协商触发条件1.2 压缩处理流程二、配置方案详解2.1 基础YAML

使用PyTorch实现手写数字识别功能

《使用PyTorch实现手写数字识别功能》在人工智能的世界里,计算机视觉是最具魅力的领域之一,通过PyTorch这一强大的深度学习框架,我们将在经典的MNIST数据集上,见证一个神经网络从零开始学会识... 目录当计算机学会“看”数字搭建开发环境MNIST数据集解析1. 认识手写数字数据库2. 数据预处理的

Python实战之屏幕录制功能的实现

《Python实战之屏幕录制功能的实现》屏幕录制,即屏幕捕获,是指将计算机屏幕上的活动记录下来,生成视频文件,本文主要为大家介绍了如何使用Python实现这一功能,希望对大家有所帮助... 目录屏幕录制原理图像捕获音频捕获编码压缩输出保存完整的屏幕录制工具高级功能实时预览增加水印多平台支持屏幕录制原理屏幕

Python实现自动化表单填写功能

《Python实现自动化表单填写功能》在Python中,自动化表单填写可以通过多种库和工具实现,本文将详细介绍常用的自动化表单处理工具,并对它们进行横向比较,可根据需求选择合适的工具,感兴趣的小伙伴跟... 目录1. Selenium简介适用场景示例代码优点缺点2. Playwright简介适用场景示例代码