开发笔记:ActiveMQ

package com.zn.p2p;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* 点对点提供者
*/
public class P2P_Provider {
public static void main(String[] args) throws JMSException {
//步骤一：创建连接工厂
ActiveMQConnectionFactory activeMQCOnnectionFactory=new ActiveMQConnectionFactory();
//步骤二：创建连接
Connection cOnnection= activeMQConnectionFactory.createConnection();
//步骤三：启动连接
connection.start();
//步骤四：获取会话工厂
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//步骤五：创建队列
Queue quque = session.createQueue("MyQueue");
//创建消息生产者
MessageProducer producer = session.createProducer(quque);
//消息持久化 参数1为不做持久化 2为持久化
producer.setDeliveryMode(2);
//模拟消息
TextMessage textMessage = session.createTextMessage("Hello ActiveMQ!");
//发送消息
producer.send(textMessage);
System.out.println("生产者生成消息完毕！");
//回收资源
session.close();
connection.close();
}
}

package com.zn.p2p;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* P2P消费者
*/
public class P2P_Consumer {
public static void main(String[] args) throws JMSException {
//步骤一：创建连接工厂
ActiveMQConnectionFactory activeMQCOnnectionFactory=new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
//步骤二：创建连接
Connection cOnnection= activeMQConnectionFactory.createConnection();
//步骤三：启动连接
connection.start();
//步骤四：获取会话工厂
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//步骤五：创建队列
Queue quque = session.createQueue("MyQueue");
//创建消费者
MessageConsumer cOnsumer= session.createConsumer(quque);
//循环获取消息
while (true){
TextMessage message = (TextMessage)consumer.receive();
if (message!=null){
System.out.println("消费者获取信息："+message.getText());
}else {
break;
}
}
//回收资源
session.close();
connection.close();
}
}

package com.zn.pubsub;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* 发布订阅消费者
*/
public class PubSub_Consumer {
public static void main(String[] args) throws JMSException {
//步骤一：创建连接工厂
ActiveMQConnectionFactory activeMQCOnnectionFactory=new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
//步骤二：创建连接
Connection cOnnection= activeMQConnectionFactory.createConnection();
//步骤三：启动连接
connection.start();
//步骤四：获取会话工厂
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//步骤五：创建主题
Topic topic = session.createTopic("MyQueue");
//创建消费者
MessageConsumer cOnsumer= session.createConsumer(topic);
//循环获取消息
while (true){
TextMessage message = (TextMessage)consumer.receive();
if (message!=null){
System.out.println("消费者获取信息："+message.getText());
}else {
break;
}
}
//回收资源
session.close();
connection.close();
}
}

package com.zn.pubsub;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* 发布订阅提供者
*/
public class PubSub_Provider {
public static void main(String[] args) throws JMSException {
//步骤一：创建连接工厂
ActiveMQConnectionFactory activeMQCOnnectionFactory=new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER, ActiveMQConnectionFactory.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616");
//步骤二：创建连接
Connection cOnnection= activeMQConnectionFactory.createConnection();
//步骤三：启动连接
connection.start();
//步骤四：获取会话工厂
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//步骤五：创建主题
Topic topic = session.createTopic("MyQueue");
//创建消息生产者
MessageProducer producer = session.createProducer(null);
//消息持久化 参数1为不做持久化 2为持久化
producer.setDeliveryMode(2);
//模拟消息
TextMessage textMessage = session.createTextMessage("Hello ActiveMQ--PubSub!");
//发送消息
producer.send(topic,textMessage);
System.out.println("生产者生成消息完毕！");
//回收资源
session.close();
connection.close();
}
}