其他参考:
http://wangxinchun.iteye.com/blog/2146172
http://wangxinchun.iteye.com/blog/2146120
http://wangxinchun.iteye.com/blog/2145998
http://wangxinchun.iteye.com/blog/2145958
http://wangxinchun.iteye.com/blog/2147335
有这样一种场景,如果Publisher 发布消息是,某一个Subscriber 由于特殊原因,比如断网或在系统死掉,那么希望在这个Subscriber 在重启之后能收到Publisher 在自己挂掉这段时间内的消息。AMQ 提供了这种功能~
做法:
值需要Subscriber 的 MessageConsumer的生成方式做变动:
MessageConsumer consumer = session.createDurableSubscriber(destination, clientId) ;
对,就是指定clientId ,重启后broker根据clientId发送遗漏的消息,并且这个clientId 不能有冲突,最好使用机器的ip+port+businessID 。
请看下面这个例子:
public class Subscriber implements MessageListener {
private static final String BROKER_URL = "tcp://localhost:61616";
private static final Boolean NON_TRANSACTED = false;
private final CountDownLatch countDownLatch;
public Subscriber(CountDownLatch latch) {
countDownLatch = latch;
}
public static void main(String[] args) {
String url = BROKER_URL;
if (args.length > 0) {
url = args[0].trim();
}
System.out.println("\nWaiting to receive messages... Either waiting for END message or press Ctrl+C to exit");
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("admin", "password", url);
Connection connection = null;
final CountDownLatch latch = new CountDownLatch(1);
try {
connection = connectionFactory.createConnection();
String clientId = System.getProperty("clientId")+"";
connection.setClientID(clientId);
connection.start();
Session session = connection.createSession(NON_TRANSACTED, Session.AUTO_ACKNOWLEDGE);
Topic destination = session.createTopic("test-topic");
MessageConsumer consumer = session.createDurableSubscriber(destination, clientId) ;
consumer.setMessageListener(new Subscriber(latch));
latch.await();
consumer.close();
session.close();
} catch (Exception e) {
System.out.println("Caught exception!");
}
finally {
if (connection != null) {
try {
connection.close();
} catch (JMSException e) {
System.out.println("Could not close an open connection...");
}
}
}
}
@Override
public void onMessage(Message message) {
try {
if (message instanceof TextMessage) {
String text = ((TextMessage) message).getText();
if ("END".equalsIgnoreCase(text)) {
System.out.println("Received END message!");
countDownLatch.countDown();
}
else {
System.out.println("Received message:" +text);
}
}
} catch (JMSException e) {
System.out.println("Got a JMS Exception!");
}
}
}
验证:1、启动Subscriber
2、启动Publisher
3、关闭Subscriber
4、使用Publisher 发消息。
5、启动Subscriber 注意:此时Subscriber 收到了消息~
分享到:
相关推荐
activemq-all-5.8.0.jar 下载 activemq-all-5.8.0.jar 下载 activemq-all-5.8.0.jar 下载 activemq-all-5.8.0.jar 下载 activemq-all-5.8.0.jar 下载
activemq-pool-5.8.0-sources.jar
赠送jar包:activemq-core-5.7.0.jar; 赠送原API文档:activemq-core-5.7.0-javadoc.jar; 赠送源代码:activemq-core-5.7.0-sources.jar; 包含翻译后的API文档:activemq-core-5.7.0-javadoc-API文档-中文...
apache-activemq-5.16.0.zip
赠送jar包:activemq-protobuf-1.1.jar; 赠送原API文档:activemq-protobuf-1.1-javadoc.jar; 赠送源代码:activemq-protobuf-1.1-sources.jar; 包含翻译后的API文档:activemq-protobuf-1.1-javadoc-API文档-...
apache-activemq-5.9.0-bin.zip
apache-activemq-5.8.0-bin.zip
apache-activemq-5.15.15二进制包,安装包,apache-activemq-5.15.15-bin.tar.gz
activemq-web-console的默认使用方式是通过在activemq.xml中导入jetty.xml配置一个jetty server来实现的。其实activemq-web-console完全可以和activemq-broker分开来部署。 activemq-web-console包含3个apps, 1.一...
activemq-cpp-library-3.9.5编译的C++库文件,支持vs2015,vs2017,包括了依赖的库文件和头文件,可以在项目中直接使用。
解压缩apache-activemq-5.5.1-bin.zip,然后双击apache-activemq-5.5.1\bin\activemq.bat运行ActiveMQ程序。 包含了apache-activemq-5.5.1-bin.zip以及ActiveMQ一个helloworld的demo启动ActiveMQ以后,登陆:...
activemq-all.5.14jar包,亲测可用!!
activemq-all-5.2.0.JAR包,欢迎下载。编写java中间件的时候会用到。这是activemq实现的jms中间件。希望能帮助到你。
tar -zxvf apache-activemq-5.15.12-bin.tar.gz 2.进入bin目录 cd /apache-activemq-5.15.12/bin 3.运行,没有配置环境变量只能在bin目录下使用命令 ./activemq 4.配置环境变量,配置完环境变量之后...
activemq-protobuf-1.1.jar;activemq-protobuf-1.1.jar
apache-activemq-5.15.9-bin.zip apache-activemq-5.15.9-bin.zip apache-activemq-5.15.9-bin.zip
apache-activemq-5.15.3-bin apache-activemq-5.15.3-bin
标签:activemq-kahadb-store-5.9.1.jar,activemq,kahadb,store,5.9.1,jar包下载,依赖包
自己编译的activemq-cpp的2.X的最后版本,在windows + vs2010下测试通过。压缩包中包含了include\lib\dll文件。 由于尝试了N个3.X的版本都不同过,于是退到2.X版本,通过了,另外在vs2005下编译出来的也没有问题。...
activemq-cpp-library-3.9.5-src.zip