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

相关文章

Android实现两台手机屏幕共享和远程控制功能

《Android实现两台手机屏幕共享和远程控制功能》在远程协助、在线教学、技术支持等多种场景下,实时获得另一部移动设备的屏幕画面,并对其进行操作,具有极高的应用价值,本项目旨在实现两台Android手... 目录一、项目概述二、相关知识2.1 MediaProjection API2.2 Socket 网络

Redis消息队列实现异步秒杀功能

《Redis消息队列实现异步秒杀功能》在高并发场景下,为了提高秒杀业务的性能,可将部分工作交给Redis处理,并通过异步方式执行,Redis提供了多种数据结构来实现消息队列,总结三种,本文详细介绍Re... 目录1 Redis消息队列1.1 List 结构1.2 Pub/Sub 模式1.3 Stream 结

MySQL索引的优化之LIKE模糊查询功能实现

《MySQL索引的优化之LIKE模糊查询功能实现》:本文主要介绍MySQL索引的优化之LIKE模糊查询功能实现,本文通过示例代码给大家介绍的非常详细,感兴趣的朋友一起看看吧... 目录一、前缀匹配优化二、后缀匹配优化三、中间匹配优化四、覆盖索引优化五、减少查询范围六、避免通配符开头七、使用外部搜索引擎八、分

Android实现悬浮按钮功能

《Android实现悬浮按钮功能》在很多场景中,我们希望在应用或系统任意界面上都能看到一个小的“悬浮按钮”(FloatingButton),用来快速启动工具、展示未读信息或快捷操作,所以本文给大家介绍... 目录一、项目概述二、相关技术知识三、实现思路四、整合代码4.1 Java 代码(MainActivi

SpringBoot集成Milvus实现数据增删改查功能

《SpringBoot集成Milvus实现数据增删改查功能》milvus支持的语言比较多,支持python,Java,Go,node等开发语言,本文主要介绍如何使用Java语言,采用springboo... 目录1、Milvus基本概念2、添加maven依赖3、配置yml文件4、创建MilvusClient

使用Python开发一个带EPUB转换功能的Markdown编辑器

《使用Python开发一个带EPUB转换功能的Markdown编辑器》Markdown因其简单易用和强大的格式支持,成为了写作者、开发者及内容创作者的首选格式,本文将通过Python开发一个Markd... 目录应用概览代码结构与核心组件1. 初始化与布局 (__init__)2. 工具栏 (setup_t

SpringBoot实现微信小程序支付功能

《SpringBoot实现微信小程序支付功能》小程序支付功能已成为众多应用的核心需求之一,本文主要介绍了SpringBoot实现微信小程序支付功能,文中通过示例代码介绍的非常详细,对大家的学习或者工作... 目录一、引言二、准备工作(一)微信支付商户平台配置(二)Spring Boot项目搭建(三)配置文件

在Android平台上实现消息推送功能

《在Android平台上实现消息推送功能》随着移动互联网应用的飞速发展,消息推送已成为移动应用中不可或缺的功能,在Android平台上,实现消息推送涉及到服务端的消息发送、客户端的消息接收、通知渠道(... 目录一、项目概述二、相关知识介绍2.1 消息推送的基本原理2.2 Firebase Cloud Me

Spring Boot项目中结合MyBatis实现MySQL的自动主从切换功能

《SpringBoot项目中结合MyBatis实现MySQL的自动主从切换功能》:本文主要介绍SpringBoot项目中结合MyBatis实现MySQL的自动主从切换功能,本文分步骤给大家介绍的... 目录原理解析1. mysql主从复制(Master-Slave Replication)2. 读写分离3.

Redis分片集群的实现

《Redis分片集群的实现》Redis分片集群是一种将Redis数据库分散到多个节点上的方式,以提供更高的性能和可伸缩性,本文主要介绍了Redis分片集群的实现,具有一定的参考价值,感兴趣的可以了解一... 目录1. Redis Cluster的核心概念哈希槽(Hash Slots)主从复制与故障转移2.