MessageConsumer是一个由Session创建的对象，用来从Destination接收消息。
1MessageConsumer createConsumer(Destination destination);
2MessageConsumer createConsumer(Destination destination,String messageSelector);
3MessageConsumer createConsumer(Destination destination,String messageSelector,boolean noLocal);
4TopicSubscriber createDurableSubscriber(Topic topic,String name);
5TopicSubscriber createDurableSubscriber(Topic topic,String name,String messageSelector,boolean noLocal);
其中messageSelector为消息选择器，noLocal标志默认是false,当设置为true时限制消费者只能接受和自己相同的连接（connection）所发布的消息，此标志只适用于主题，不适用队列，name标志订阅主题所对应的订阅名称，持久订阅时需要设置此参数。
public final String SELECTOR="JMS_TYPE='MY_TAG1'";该选择器检查了传入消息的JMS_TYPE属性，并确定了这个属性的值是否等于MY_TAG1.如果相等，则消息被消费掉，如果不相等，name消息会被忽略。
消息的同步接收和异步接收：
消息的同步接收是指客户端主动去接收消息，客户端可以采取MessageConsumer的receive方法去接收下一个消息。
Message receive()
Message receive(long timeout)
Message receiveNoWait();
消息的异步接收是指消息到达的时候，ActiveMQ主动通知客户端，可以通过注册一个实现MessageListener接口的对象到Messageconsumer。MessageListener只有一个必须实现的方法----onMessage,它只接收一个参数，即Message.在为每个发送到Destination的消息实现onMessage时，将调用该方法。
代码实现：
package bhz.mq.p2p;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MapMessage;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
public class Producer {
//单例模式
//1：连接工厂
private ConnectionFactory connectionFactory;
//2:连接对象
private Connection connection;
//3:Session对象
private Session session;
//4:生产者
private MessageProducer messageProducer;
public Producer(){
try {
this.connectionFactory = new ActiveMQConnectionFactory("bhz","bhz",
"tcp://localhost:61616");
this.connection = this.connectionFactory.createConnection();
this.connection.start();
this.session = this.connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
this.messageProducer = this.session.createProducer(null);
} catch (Exception e) {
e.printStackTrace();
}
}
public Session getSession(){
return this.session;
}
public void send1(){
try {
Destination destination = this.session.createQueue("first");
MapMessage msg1 = this.session.createMapMessage();
msg1.setString("name", "张三");
msg1.setString("age", "23");
msg1.setStringProperty("color", "blue");
msg1.setIntProperty("sal", 2200);
MapMessage msg2 = this.session.createMapMessage();
msg2.setString("name", "李四");
msg2.setString("age", "26");
msg2.setStringProperty("color", "red");
msg2.setIntProperty("sal", 1300);
MapMessage msg3 = this.session.createMapMessage();
msg3.setString("name", "王五");
msg3.setString("age", "28");
msg3.setStringProperty("color", "green");
msg3.setIntProperty("sal", 1500);
MapMessage msg4 = this.session.createMapMessage();
msg4.setString("name", "赵六");
msg4.setString("age", "30");
msg4.setStringProperty("color", "blue");
msg4.setIntProperty("sal", 1800);
this.messageProducer.send(destination, msg1, DeliveryMode.NON_PERSISTENT, 2,1000*60*10L);
this.messageProducer.send(destination, msg2, DeliveryMode.NON_PERSISTENT, 3,1000*60*10L);
this.messageProducer.send(destination, msg3, DeliveryMode.NON_PERSISTENT, 6,1000*60*10L);
this.messageProducer.send(destination, msg4, DeliveryMode.NON_PERSISTENT, 9,1000*60*10L);
} catch (Exception e) {
e.printStackTrace();
}
}
public void send2(){
try {
Destination destination = this.session.createQueue("first");
TextMessage msg = this.session.createTextMessage("我是一个字符串内容");
this.messageProducer.send(destination,msg,DeliveryMode.NON_PERSISTENT,9,1000*60*10L);
} catch (Exception e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
Producer p = new Producer();
p.send1();
}
}
package bhz.mq.p2p;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
public class Consumer {
public final String SELECTOR_0="age > 25";
public final String SELECTOR_1="color = 'blue'";
public final String SELECTOR_2="color = 'blue' AND sal > 2000";
public final String SELECTOR_3="receiver = 'A'";
//1：连接工厂
private ConnectionFactory connectionFactory;
//2:连接对象
private Connection connection;
//3:Session对象
private Session session;
//4:消费者
private MessageConsumer messageConsumer;
//5：目标地址
private Destination destination;
public Consumer(){
try {
this.connectionFactory = new ActiveMQConnectionFactory("bhz","bhz",
"tcp://localhost:61616");
this.connection = this.connectionFactory.createConnection();
this.connection.start();
this.session = this.connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
this.destination=this.session.createQueue("first");
//创建消费者的时候发生了变化
this.messageConsumer = this.session.createConsumer(this.destination, SELECTOR_2);
} catch (Exception e) {
e.printStackTrace();
}
}
public void receiver(){
try {
this.messageConsumer.setMessageListener(new Listener());
} catch (Exception e) {
e.printStackTrace();
}
}
class Listener implements MessageListener{
@Override
public void onMessage(Message message) {
try {
if (message instanceof TextMessage) {
}
if (message instanceof MapMessage) {
MapMessage ret = (MapMessage)message;
System.out.println(ret.toString());
System.out.println(ret.getString("name"));
System.out.println(ret.getString("age"));
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) {
Consumer c = new Consumer();
c.receiver();
}
}
