一、消息选择器
消息选择器:过滤消息属性与设置条件相等的消息进行消费。语义与sql一致。
private final String selector_1 = "sex='w'";
this.consumer = session.createConsumer(destination, selector_1);
二、消息异步接收
消息异步接收 :当消息到达时,ActiveMQ主动通知消费端,可以注册一个MessageListener类实现onMessage方法,监听MQ送达消息
示例:
public class Producer {
// 建立connectionFactory工厂对象
private ActiveMQConnectionFactory connectionFactory;
// 连接对象
private Connection connection;
// session对象
private Session session;
// 生产者
private MessageProducer producer;
public Producer() {
this.connectionFactory = new ActiveMQConnectionFactory();
try {
this.connection = connectionFactory.createConnection("fu", "fu");
this.connection.start();
//参一:开启事务,参二,手工签收
this.session = connection.createSession(Boolean.TRUE, Session.CLIENT_ACKNOWLEDGE);
this.producer = session.createProducer(null);
} catch (Exception e) {
e.printStackTrace();
}
}
public void send() throws Exception {
Destination destination = this.session.createQueue("first");
MapMessage map = this.session.createMapMessage();
//设置消息
map.setString("name", "zs");
map.setString("age", "40");
//设置消息属性
map.setStringProperty("sex", "m");
MapMessage map1 = this.session.createMapMessage();
map1.setString("name", "ls");
map1.setString("age", "20");
map1.setStringProperty("sex", "w");
MapMessage map2 = this.session.createMapMessage();
map2.setString("name", "ww");
map2.setString("age", "35");
map2.setStringProperty("sex", "w");
//参一目标,参二数据,参三非持久化,参四做优先及,参五失效时间
this.producer.send(destination, map, DeliveryMode.PERSISTENT, 2, 1000*10);
this.producer.send(destination, map1, DeliveryMode.PERSISTENT, 2, 1000*10);
this.producer.send(destination, map2, DeliveryMode.PERSISTENT, 9, 1000*10);
//提交事务
this.session.commit();
//关闭连接
this.connection.close();
}
public static void main(String[] args) throws Exception {
Producer p = new Producer();
p.send();
}
}
public class Comsumer {
// private final String selector_0 = "age>30";
// 消息过滤的不是消息本身,而是过滤消息附带的某些属性
private final String selector_1 = "sex='w'";
// 建立connectionFactory工厂对象
private ActiveMQConnectionFactory connectionFactory;
// 连接对象
private Connection connection;
// session对象
private Session session;
// 生产者
private MessageConsumer consumer;
// 目标地址
private Destination destination;
public Comsumer() {
this.connectionFactory = new ActiveMQConnectionFactory();
try {
this.connection = connectionFactory.createConnection("fu", "fu");
this.connection.start();
this.session = connection.createSession(Boolean.TRUE, Session.CLIENT_ACKNOWLEDGE);
this.destination = this.session.createQueue("first");
// 消息过滤的不是消息本身,而是过滤消息附带的某些属性
this.consumer = session.createConsumer(destination, selector_1);
} catch (Exception e) {
e.printStackTrace();
}
}
public void recever() throws Exception {
// 消息异步接收:当消息到达时,ActiveMQ主动通知消费端,可以注册一个MessageListener类实现onMessage方法,监听MQ送达消息
this.consumer.setMessageListener(new Listener());
}
class Listener implements MessageListener {
public void onMessage(Message message) {
try {
if (message instanceof TextMessage) {
} else if (message instanceof MapMessage) {
MapMessage m = (MapMessage) message;
System.out.println(m.toString());
System.out.println(m.getString("name"));
System.out.println(m.getString("age"));
// 手工签收消息
m.acknowledge();
}
} catch (JMSException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) throws Exception {
Comsumer c = new Comsumer();
c.recever();
}
}
分享到:
相关推荐
springboot集成activemq实现消息接收demo
ActiveMQ与Spring整合之异步发送邮件。
ActiveMq整合SpringMVC实现批量邮件进行异步发送,下载后可在Eclipse中直接使用(含有ActiveMq安装文件已配置好解压即可用)希望对大家有所帮助
activeMQ的测试工具,用于发送和接收activeMQ消息,jar包形式的,安装完jdk之后用java -jar xxx.jar命令运行
整合Spring + ActiveMQ 的朋友可以下载看一下 简单易懂
SpringBoot+ActiveMq+MQTT实现消息的发送和接收 后台消费者、生产者、消息发送接口、发送消息业务类等相关配置
Android基于MQTT协议利用ActiveMQ发送消息给Android端接收,属于长连接那种,类似Socket通信
springboot整合ActiveMQ源码,适合范围消息队列入门小伙伴,对ActiveMQ消息队列不太了解,不知道如何发送消息,接收消息可以围观。
activemq 通过ajax发送接收消息简单例子
详细描述了ActiveMQ消息过期-时间设置和自动清除解决方案。
ActiveMQ允许客户端使用多种协议接收消息,其中WebSocket协议的URL使用ws://开头,默认端口号是61614
ActiveMQ 消息队列
本代码关于activemq-cpp的核心代码参考的chenxun2009的博客园,其他部分包括:从配置文件中读取消息通道,过滤条件等信息。
activemq消息中间件-视频教程activemq消息中间件-视频教程activemq消息中间件-视频教程activemq消息中间件-视频教程
实现了ActiveMQ的初步封装,比较适合新手入门学习,简单明了
ActiveMQ队列消息过期时间设置和自动清除解决方案.docx
ActiveMQ消息队列主题订阅Spring整合
activeMQ的发送消息后接收者返回信息
SpringBoot快速玩转ActiveMQ消息队列,jdk8下的简要版介绍。