activemq两种实现方式

2018/3/10 18:55:01 人评论 次浏览 分类:java

第一种:点对点

#发布者
public
class Producer { private static final String userName = ActiveMQXAConnectionFactory.DEFAULT_USER; private static final String password = ActiveMQXAConnectionFactory.DEFAULT_PASSWORD; private static final String brokerURL = "tcp://192.168.178.X:61616"; public static void main(String[] args) throws JMSException { //1.创建连接工厂类 ConnectionFactory factory = new ActiveMQXAConnectionFactory(userName, password, brokerURL); //2.创建连接 Connection connection = factory.createConnection(); //3.启动连接 connection.start(); //4.创建会话对象session(事务transacted为true,参数2不生效) //acknowledgeMode: Session session = connection.createSession(true,Session.AUTO_ACKNOWLEDGE); //5.目的地 Queue queue = session.createQueue("mq-test-01"); //7.创建发送者 MessageProducer producer = session.createProducer(queue); for(int i=1;i<=10;i++) { //6.消息对象 TextMessage message = session.createTextMessage(); message.setText("消息"+i); //8.发送消息 producer.send(message); } //9.会话提交 // session.commit(); //10.关闭连接 connection.close(); } }

 

#消费者
public
class Consumer1 { private static final String userName = ActiveMQXAConnectionFactory.DEFAULT_USER; private static final String password = ActiveMQXAConnectionFactory.DEFAULT_PASSWORD; private static final String brokerURL = "tcp://192.168.178.X:61616"; public static void main(String[] args) throws JMSException { //1.创建连接工厂类 ConnectionFactory factory = new ActiveMQXAConnectionFactory(userName, password, brokerURL); //2.创建连接 Connection connection = factory.createConnection(); //3.启动连接 connection.start(); //4.创建会话对象session(事务transacted为true,参数2不生效) //acknowledgeMode: Session session = connection.createSession(true,Session.AUTO_ACKNOWLEDGE); //5.目的地 Queue queue = session.createQueue("mq-test-01"); //6.接收消息对象 MessageConsumer consumer = session.createConsumer(queue); //7.通过监听器接收消息 consumer.setMessageListener(new MessageListener() { public void onMessage(Message message) { TextMessage textMessage = (TextMessage)message; //获取消息 try { String msg = textMessage.getText(); System.out.println(msg); } catch (JMSException e) { } } }); } }

 

 

第二种: 发布者/订阅者

启动顺序:先订阅、再发布

#订阅者
public
class Subscriber1 { private static final String userName = ActiveMQXAConnectionFactory.DEFAULT_USER; private static final String password = ActiveMQXAConnectionFactory.DEFAULT_PASSWORD; private static final String brokerURL = "tcp://192.168.129.10:61616"; public static void main(String[] args) throws JMSException { //1.创建连接工厂类 ConnectionFactory factory = new ActiveMQXAConnectionFactory(userName, password, brokerURL); //2.创建连接 Connection connection = factory.createConnection(); //3.启动连接 connection.start(); //4.创建会话对象session(事务transacted为true,参数2不生效) //acknowledgeMode: Session session = connection.createSession(true,Session.AUTO_ACKNOWLEDGE); //5.目的地 Topic topic = session.createTopic("mq-test-02"); //6.接收消息对象 MessageConsumer consumer = session.createConsumer(topic); //7.通过监听器接收消息 consumer.setMessageListener(new MessageListener() { public void onMessage(Message message) { TextMessage textMessage = (TextMessage)message; //获取消息 try { String msg = textMessage.getText(); System.out.println(msg); } catch (JMSException e) { } } }); } }

 

#发布者
public
class Publisher { private static final String userName = ActiveMQXAConnectionFactory.DEFAULT_USER; private static final String password = ActiveMQXAConnectionFactory.DEFAULT_PASSWORD; private static final String brokerURL = "tcp://192.168.129.10:61616"; public static void main(String[] args) throws JMSException { //1.创建连接工厂类 ConnectionFactory factory = new ActiveMQXAConnectionFactory(userName, password, brokerURL); //2.创建连接 Connection connection = factory.createConnection(); //3.启动连接 connection.start(); //4.创建会话对象session(事务transacted为true,参数2不生效) //acknowledgeMode: Session session = connection.createSession(true,Session.AUTO_ACKNOWLEDGE); //5.目的地 Topic topic = session.createTopic("mq-test-02"); //7.创建发送者 MessageProducer producer = session.createProducer(topic); for(int i=1;i<=10;i++) { //6.消息对象 TextMessage message = session.createTextMessage(); message.setText("消息"+i); //8.发送消息 producer.send(message); //设置自动持久化 producer.setDeliveryMode(DeliveryMode.PERSISTENT); } //9.会话提交 // session.commit(); //10.关闭连接 connection.close(); } }

 

相关知识

  • Servlet 工作原理解析

    Web 技术成为当今主流的互联网 Web 应用技术之一,而 Servlet 是 Java Web 技术的核心基础。因而掌握 Servlet 的工作原理是成为一名合格的 Java Web 技术开发人员的基本要求。本文将带你认识 Java Web 技术是如何基于 Servlet 工作,你将知道:以 Tomcat 为例了解 Servlet 容…

    2017/7/18 11:11:00

共有访客发表了评论 网友评论

验证码: 看不清楚?