序上一篇:
http://wangxinchun.iteye.com/blog/2145958
对于consumer而言,receive消息的时候,可以根据关键字筛选自己感兴趣的消息,当然可以定义不同的Destination 队列来承载这些消息,但是一般情况下,同一个队列可以保存一个主题的消息,减少队列的个数,为此新增了consumer的receive时的过滤功能。
case如下:
消息消费者:
/**
通过改动intended参数,我们分别获取到了50条消息
*/
public class Consumer {
private static final String BROKER_URL = "tcp://localhost:61616";
private static final Boolean NON_TRANSACTED = false;
private static final long TIMEOUT = 20000;
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... will timeout after " + TIMEOUT / 1000 +"s");
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("admin", "password", url);
Connection connection = null;
try {
connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(NON_TRANSACTED, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue("test-queue");
//通过改动intended参数,我们分别获取到了50条消息
MessageConsumer consumer = session.createConsumer(destination, "intended = 'you'");
int i = 0;
while (true) {
Message message = consumer.receive(TIMEOUT);
if (message != null) {
if (message instanceof TextMessage) {
String text = ((TextMessage) message).getText();
System.out.println("Got " + i++ + ". message: " + text);
}
} else {
break;
}
}
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...");
}
}
}
}
}
消息生产者:
/**
* 此producer 对队列test-queue 发送消息,设置intended 属性,值分别为me you
*/
public class Producer {
private static final String BROKER_URL = "tcp://localhost:61616";
private static final Boolean NON_TRANSACTED = false;
private static final int NUM_MESSAGES_TO_SEND = 100;
private static final long DELAY = 100;
public static void main(String[] args) {
String url = BROKER_URL;
if (args.length > 0) {
url = args[0].trim();
}
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("admin", "password", url);
Connection connection = null;
try {
connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(NON_TRANSACTED, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue("test-queue");
MessageProducer producer = session.createProducer(destination);
for (int i = 0; i < NUM_MESSAGES_TO_SEND; i++) {
TextMessage message = session.createTextMessage("Message #" + i);
System.out.println("Sending message #" + i);
if (i % 2 == 0) {
System.out.println("Sending to me");
message.setStringProperty("intended", "me");
} else {
System.out.println("Sending to you");
message.setStringProperty("intended", "you");
}
producer.send(message);
Thread.sleep(DELAY);
}
producer.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...");
}
}
}
}
}
验证:通过Producer 发送了100 条消息。
通过Consumer 设置intended 为me 收到50 条消息,通过Consumer设置intended 为you 收到50条消息,符合预期~
分享到:
相关推荐
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
解压缩apache-activemq-5.5.1-bin.zip,然后双击...包含了apache-activemq-5.5.1-bin.zip以及ActiveMQ一个helloworld的demo启动ActiveMQ以后,登陆:http://localhost:8161/admin/,创建一个Queue,命名为FirstQueue。
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,包括了依赖的库文件和头文件,可以在项目中直接使用。
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-all.5.14jar包,亲测可用!!
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