publish/subscribe
特点:A发送的消息可以被所有监听A的对象的接收,就好比学校的广播,所有的学生都可以收听校园广播信息。
消息生产者:
package com.zhiwei.advanced.mq.activemq.sp;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
/**
* 发布/订阅消息模型 测试顺序:先订阅才能收到消息
*/
public class JMSProducer {
private final static String user = ActiveMQConnection.DEFAULT_USER; // 默认用户名
private final static String password = ActiveMQConnection.DEFAULT_PASSWORD; // 默认密码
private final static String brokeURL = ActiveMQConnection.DEFAULT_BROKER_URL; // 链接地址
public static void main(String[] args) throws Exception {
// 连接工厂
ConnectionFactory factory = new ActiveMQConnectionFactory(JMSProducer.user, JMSProducer.password,JMSProducer.brokeURL);
// 创建连接
Connection connection = factory.createConnection();
// 启动连接
connection.start();
// 接受或发送消息的线程
Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createTopic("FirstTopic"); // 创建消息主题:Destination子类:Queue/Topic
MessageProducer messageProducer = session.createProducer(destination); // 创建消息生产者
// 发送文本消息
for (int i = 0; i < 10; i++) {
TextMessage message = session.createTextMessage("JMS Provider发送消息:" + i);
System.out.println("JMS Provider发送消息:" + i);
messageProducer.send(message);
}
// 启用事务时session需要提交
session.commit();
session.close();
connection.close();
}
}
消息消费者1:
package com.zhiwei.advanced.mq.activemq.sp;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
/**
* 发布/订阅消息模型
*
* 特別注意:发布订阅消息模型必须先客户端监听,然后主题发送消息
*/
public class JMSConsumer01{
private final static String user = ActiveMQConnection.DEFAULT_USER; // 默认用户名
private final static String password www.yigouyule2.cn/= ActiveMQConnection.DEFAULT_PASSWORD; // 默认密码
private final static String brokeURL = ActiveMQConnection.DEFAULT_BROKER_URL; // 链接地址
public static void main(String[] args) throws Exception {
ConnectionFactory factory www.zhenghongyule.cn= new ActiveMQConnectionFactory(JMSConsumer01.user, JMSConsumer01.password,JMSConsumer01.brokeURL); // 链接工厂
Connection connection = factory.createConnection(); // 连接
connection.start(); // 启动连接
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);// 接受或发送消息的线程:消费不需事务
Destination destination =www. jiahuayulpt.com session.createTopic("FirstTopic"); // 创建连接消息主题:Destination子类:Queue/Topic
MessageConsumer messageConsumer = session.createConsumer(destination); // 创建消息生产者
messageConsumer.setMessageListener(new JMSListener()); //注册消息监听 :阻塞监听
}
}
消息消费者2:
package com.zhiwei.advanced.mq.activemq.sp;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.www.zhongyiyuL.cn Destination;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import org.apache.www.myzx157.com activemq.ActiveMQConnection;
import org.apache.www.mytxyl1.com activemq.ActiveMQConnectionFactory;
/**
* 发布/订阅消息模型
*/
public class JMSConsumer02{
private final static String user = ActiveMQConnection.DEFAULT_USER; // 默认用户名
private final static String password = ActiveMQConnection.DEFAULT_PASSWORD; // 默认密码
private final static String brokeURL = ActiveMQConnection.DEFAULT_BROKER_URL; // 链接地址
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ActiveMQConnectionFactory(JMSConsumer02.user, JMSConsumer02.password,JMSConsumer02.brokeURL); // 链接工厂
Connection connection = factory.createConnection(); // 连接
connection.start(); // 启动连接
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);// 接受或发送消息的线程:消费不需事务
Destination destination = session.createTopic("FirstTopic"); // 创建连接消息主题:Destination子类:Queue/Topic
MessageConsumer messageConsumer = session.createConsumer(destination); // 创建消息生产者
messageConsumer.setMessageListener(new JMSListener()); //注册消息监听 :阻塞监听
}
}
消息队列监听器:
package com.zhiwei.advanced.mq.activemq.sp;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
public class JMSListener implements MessageListener{
@Override
public void onMessage(Message message) {
if(message instanceof TextMessage){
try {
System.out.println(((TextMessage) message).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
【ActiveMQ】- 发布/订阅模式
版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/li123128/article/details/87909062
猜你喜欢
转载自blog.csdn.net/li123128/article/details/87909062
今日推荐
周排行