首先看activeMQ通讯中的点对点模式——队列模式。特点：
1）每一个消息只能有一个消费者，一对一的关系
2）生产者和消费者没有时间上的相关性
3）消息被消费后队列中不会再存储，不会再次被消费掉。
-
简易图如下：
activemq.png - 具体实现看如下代码：
生产者：
package com.jjclub.activeMQ_01;
import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
//生产者01
public class Producer01 {
//服务器地址
private static String url = "tcp://localhost:61616";
//队列名称
private static String queueName="queue1";
public static void main(String[] args) {
//创建activemq连接工场
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(url);
Connection connection = null;
Session session =null;
MessageProducer messageProducer =null;
try {
//创建连接connection
connection = activeMQConnectionFactory.createConnection();
//启动连接
connection.start();
//创建连接session;第一个参数为事务，第二个参数为签发机制。暂时选择默认，后续说明；
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//创建消息目的地队列（队列名称）
Queue queue = session.createQueue(queueName);
//创建生产者
messageProducer = session.createProducer(queue);
//创建消息体（文本消息内容）
TextMessage textMessage = session.createTextMessage("hello");
//发送消息到队列中
messageProducer.send(textMessage);
} catch (JMSException e) {
e.printStackTrace();
}finally {
try {
//关闭消息
messageProducer.close();
session.close();
connection.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
消费者：
1）receive方式
package com.jjclub.activeMQ_01;
import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
//消费者
public class Consumer01 {
//服务器地址
private static String url = "tcp://localhost:61616";
//队列名称
private static String queueName="queue1";
public static void main(String[] args) {
//创建activemq连接工场
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(url);
Connection connection = null;
Session session =null;
MessageConsumer messageConsumer =null;
try {
//创建连接connection
connection = activeMQConnectionFactory.createConnection();
//启动连接
connection.start();
//创建连接session;第一个参数为事务，第二个参数为签发机制。暂时选择默认，后续说明；
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//创建消息目的地队列（队列名称）
Queue queue = session.createQueue(queueName);
//创建消费者
messageConsumer = session.createConsumer(queue);
while(true) {
// messageConsumer.receive();此方法会一直等待消息，不会中止进程
// messageConsumer.receive(4000L);等待4s后，若无消息，则中止进程，不再等待
Message message = messageConsumer.receive();
if(message!=null) {
TextMessage textMessage = (TextMessage) message;
System.out.println("消费的消息是"+textMessage);
}else {
break;
}
}
} catch (JMSException e) {
e.printStackTrace();
}finally {
try {
//关闭消息
messageConsumer.close();
session.close();
connection.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
2）监听方式
package com.jjclub.activeMQ_01;
import java.io.IOException;
import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
//消费者
public class Consumer01_1 {
//服务器地址
private static String url = "tcp://localhost:61616";
//队列名称
private static String queueName="queue1";
public static void main(String[] args) {
//创建activemq连接工场
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(url);
Connection connection = null;
Session session =null;
MessageConsumer messageConsumer =null;
try {
//创建连接connection
connection = activeMQConnectionFactory.createConnection();
//启动连接
connection.start();
//创建连接session;第一个参数为事务，第二个参数为签发机制。暂时选择默认，后续说明；
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//创建消息目的地队列（队列名称）
Queue queue = session.createQueue(queueName);
//创建消费者
messageConsumer = session.createConsumer(queue);
//此方式在进程不停止情况下，会一直监听消息，一旦有消息就执行消费
messageConsumer.setMessageListener(new MessageListener() {
public void onMessage(Message message) {
if(message!=null) {
TextMessage textMessage = (TextMessage)message;
System.out.println("消费消息为"+textMessage);
}
}
});
//保证控制台不灭，进程不停止。
System.in.read();
} catch (JMSException e) {
e.printStackTrace();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}finally {
try {
//关闭消息
messageConsumer.close();
session.close();
connection.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
