概要
- ActiveMQ 的 request-response通信方式
- ActiveMQ 的事务
- ActiveMQ消息消费的同步和异步
- ActiveMQ的零散知识点
ActiveMQ 的 request-response通信方式
在前面的两种模式中都是一方负责发送消息而另外一方负责处理。而我们实际中的很多应用相当于一种一应一答的过程，需要双方都能给对方发送消息。于是请求-应答的这种通信方式也很重要。它也应用的很普遍。
首先发送端代码
public class ReSend {
public static void main(String[] args) throws JMSException {
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination adminQueue = session.createQueue("shli.1");
//创建生产者发送消息
MessageProducer producer = session.createProducer(adminQueue);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
//创建一个临时的通道，并且创建消费者
Destination tempDest = session.createTemporaryQueue();
MessageConsumer responseConsumer = session.createConsumer(tempDest);
//设置临时消息的一个监听器
responseConsumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message arg0) {
try {
System.out.println("收到回复内容："+((TextMessage)arg0).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
//模拟消息
TextMessage txtMessage = session.createTextMessage();
txtMessage.setText("我是ReSend");
//设置消息头部信息 使得回复的消息可以定位到刚刚创建的临时通道
txtMessage.setJMSReplyTo(tempDest);
//这里在消息头部设置一个标示，好像仅仅是用于业务上的判断，并不影响消息的应答
String correlationId = UUID.randomUUID().toString();
txtMessage.setJMSCorrelationID(correlationId);
producer.send(txtMessage);
}
}
跟之前P2P不一样的是发送端这里有创建一个临时通道以及临时通道对应的消费者，并且在消息的头部添加了JMSReplyTo和MSCorrelationID，用于接收端进行消息的回复。
然后接收端的代码
public class ReReceive {
public static void main(String[] args) throws JMSException {
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
Connection connection = connectionFactory.createConnection();
connection.start();
final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination adminQueue = session.createQueue("shli.1");
//创建回复用的生产者
final MessageProducer replyProducer = session.createProducer(null);
replyProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
//创建消费者对消息进行消费
MessageConsumer consumer = session.createConsumer(adminQueue);
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
try {
TextMessage response = session.createTextMessage();
if (message instanceof TextMessage) {
TextMessage txtMsg = (TextMessage) message;
String messageText = txtMsg.getText();
System.out.println("收到了来自ReSend的消息------"+ messageText);
response.setText("回复ReSend，From ReReceive");
}
response.setJMSCorrelationID(message.getJMSCorrelationID());
replyProducer.send(message.getJMSReplyTo(), response);
} catch (JMSException e) {
e.printStackTrace();
}
}
});
}
}
相比以之前的P2P接收端这里多了一个生产者
replyProducer，然后在收到消息之后，根据发送端发送过来的消息的头部信息，返回应答信息~
ActiveMQ 的本地事务
Session，用于发送和接受消息，而且是单线程的，支持事务的。比如之前的代码
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
第一个参数为true的时候表示开启事务，为false的时候表示关闭事务。
在一个JMS客户端，可以使用本地事务来组合消息的发送和接收。JMS Session接口提供了commit和rollback方法。事务提交意味着生产的所有消息被发送，消费的所有消息被确认；事务回滚意味着生产的所有消 息被销毁，消费的所有消息被恢复并重新提交，除非它们已经过期。
事务性的会话总是牵涉到事务处理中，commit或rollback方法一旦被调用，一个事务就结束了，而另一个事务被开始。关闭事务性会话将回滚其中的事务。
需要注意的是，如果使用请求/回复机制，即发送一个消息，同时希望在同一个事务中等待接收该消息的回复，那么程序将被挂起，因为直到事务提交，发送操作才会真正执行。
需要注意的还有一个，消息的生产和消费不能包含在同一个事务中。
createSession的第二个参数
在事务性会话中，当一个事务被提交的时候，确认自动发生。在非事务性会话中，消息何时被确认取决于创建会话时的应答模式（acknowledgement mode）。该参数有以下三个可选值：
• Session.AUTO_ACKNOWLEDGE。当客户成功的从receive方法返回的时候，或者从MessageListener.onMessage方法成功返回的时候，会话自动确认客户收到的消息。
• Session.CLIENT_ACKNOWLEDGE。 客户通过消息的acknowledge方法确认消息。需要注意的是，在这种模式中，确认是在会话层上进行：确认一个被消费的消息将自动确认所有已被会话消 费的消息。例如，如果一个消息消费者消费了10个消息，然后确认第5个消息，那么所有10个消息都被确认。
JMS消息只有在被确认之后，才认为已经被成功地消费了。消息的成功消费通常包含三个阶段：客户接收消息、客户处理消息和消息被确认。
• Session.DUPS_ACKNOWLEDGE。 该选择只是会话迟钝第确认消息的提交。如果JMS provider失败，那么可能会导致一些重复的消息。如果是重复的消息，那么JMS provider必须把消息头的JMSRedelivered字段设置为true
ActiveMQ消息消费的同步和异步
同步接收消息：接收者主动接收消息，若消息队列中没有消息则阻塞等待，当然也有其它方法可以规定等待时间或是不等待。
Message message = messageConsumer1.receive();
这种方式线程会一直阻塞，直到接收到消息，和Socket服务器端接收客户端请求的方式差不多
异步接收消息：当消息队列有消息时会调用接收者的onMessage方法，接收者不用阻塞等待，可执行其它业务
messageConsumer1.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message arg0) {
MapMessage map = (MapMessage)arg0;
try {
String shli = map.getString("name");
double price = map.getDouble("price");
boolean up = map.getBoolean("up");
System.out.println(shli + "-1-1-1-" + price +"-1--1-1--"+up+"---Thread---" +Thread.currentThread().getName());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
这种方式不会出现线程阻塞，当消息到达时，会在其他线程中执行onMessage()中的逻辑
ActiveMQ的零散功能点
消息过期 和 优先级
可以设置消息在一定时间后过期，默认是永不过期。设置方法 后两个send方法的最后一个参数
可以使用消息优先级来指示JMS provider首先提交紧急的消息。优先级分10个级别，从0（最低）到9（最高）。如果不指定优先级，默认级别是4。需要注意的是，JMS provider并不一定保证按照优先级的顺序提交消息。设置方法 后两个send方法的最后第三个参数
JMS Selectors
消息对象有消息属性，用于消息选择器
使用如下：
-
生产者端需要设置消息属性，一定要注意的是setXxxProperty(filed,value)
message.setIntProperty("pa", 6);
-
给出条件，其实本质上就是SQL92语法
String condition = "pa > 7";
-
创建消费者的时候，指定条件即可
MessageConsumer messageConsumer1 = session.createConsumer(destination,condition);
持久性
• PERSISTENT。指示JMS provider持久保存消息，以保证消息不会因为JMS provider的失败而丢失。
• NON_PERSISTENT。不要求JMS provider持久保存消息
Producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
持久化订阅
在第一篇的代码中我们发现，在订阅模式中，如果消费者不在线，在上线之后是没有办法重新收到生产者发送的消息的。那是因为我们没有将消费者设置为持久化订阅。
客户可以通过会话上的createDurableSubscriber方法来创建一个持久订阅，该方法的第一个参数必须是一个topic。第二个参数是订阅的名称。
这样设置之后，消费者上线后就可以收到之前生产者发送的（离线）消息。
但是这种没有持久性的消息只是被缓存在内存中，如果ActiveMQ被关闭并重新打开的话，消费者同样接收不到(离线)消息。
如果在发送端加上一句代码，这样就算是ActiveMQ重启也不会出现消息丢失的现象了
Producer.setDeliveryMode(DeliveryMode.PERSISTENT);
JMS provider会存储发布到持久订阅对应的topic上的消息。如果最初创建持久订阅的客户或者任何其它客户使用相同的连接工厂和连接的客户ID、相同 的主题和相同的订阅名再次调用会话上的createDurableSubscriber方法，那么该持久订阅就会被激活。JMS provider会象客户发送客户处于非激活状态时所发布的消息。
持久订阅在某个时刻只能有一个激活的订阅者。持久订阅在创建之后会一直保留，直到应用程序调用会话上的unsubscribe方法。
最后
下一遍将记录ActiveMQ和其他框架的整合
网友评论