ActiveMQ发布订阅模式
Topic主题发布和订阅消息
前面讲的案例都是点对点的消息，即一个生产者发送的一条消息只能被一个消费者消费，然后就移除了。
在这里插入图片描述
而topic模式一条消息可以被多个消费者订阅，关系如下:
在这里插入图片描述
定义生产者
package com.dpb.topic;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;
import org.apache.activemq.ActiveMQConnectionFactory;
/**
* ActiveMQ中的生产者(Producer)
* @author dengp
*
*/
public class MyProducer {
public void sendhello2ActiveMq(String messageText) {
TopicSession session = null;
TopicConnection conn = null;
try {
TopicConnectionFactory factory = new ActiveMQConnectionFactory("admin", "admin", "tcp://192.168.88.121:61616");
conn = factory.createTopicConnection();
conn.start();
session = conn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
// 创建消息队列
Topic topic = session.createTopic("test-topic");
// 创建消息发送者
TopicPublisher publisher = session.createPublisher(topic);
// 设置持久化模式 NON_PERSISTENT不开启 PERSISTENT 开启 默认是开启
publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
MapMessage mapMessage = session.createMapMessage();
mapMessage.setString("name", "波波烤鸭");
mapMessage.setString("address", "深圳");
publisher.send(mapMessage);
// 提交会话
session.commit();
} catch (Exception e) {
e.printStackTrace();
System.out.println("访问ActiveMQ服务发生错误!!");
} finally {
try {
// 回收会话资源
if (null != session)
session.close();
} catch (JMSException e) {
e.printStackTrace();
}
try {
// 回收链接资源
if (null != conn)
conn.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}定义消费者
package com.dpb.topic;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;
import org.apache.activemq.ActiveMQConnectionFactory;
/**
* ActiveMQ中的消费者(Consumer)
* @author dengp
*
*/
public class MyConsumer {
public void reciveHelloFormActiveMq() {
TopicSession session = null;
TopicConnection conn = null;
try {
TopicConnectionFactory factory = new ActiveMQConnectionFactory("admin", "admin", "tcp://192.168.88.121:61616");
conn = factory.createTopicConnection();
conn.start();
session = conn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
// 创建消息队列
Topic topic = session.createTopic("test-topic");
// 创建消息接受者
MessageConsumer consumer = session.createConsumer(topic);
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message msg) {
if (msg != null) {
MapMessage map = (MapMessage) msg;
try {
System.out.println(map.getString("name") + "接收#" + map.getString("address"));
} catch (JMSException e) {
e.printStackTrace();
}
}
}
});
// 休眠100s再关闭
Thread.sleep(1000 * 100);
// 提交会话
session.commit();
} catch (Exception e) {
e.printStackTrace();
System.out.println("访问ActiveMQ服务发生错误!!");
} finally {
try {
// 回收会话资源
if (null != session)
session.close();
} catch (JMSException e) {
e.printStackTrace();
}
try {
// 回收链接资源
if (null != conn)
conn.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}测试
先启动消费者，可以开启多个
public static void main(String[] args) {
MyConsumer con = new MyConsumer();
con.reciveHelloFormActiveMq();
}
在这里插入图片描述
在这里插入图片描述
启动生产者
public static void main(String[] args) {
MyProducer pro = new MyProducer();
pro.sendhello2ActiveMq("你好啊...topic");
}
在这里插入图片描述
在这里插入图片描述
好了本文介绍到此，下篇介绍ActiveMQ和Spring的整合
本文参与 腾讯云自媒体分享计划，分享自作者个人站点/博客。
原始发表：2019年03月17日，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读