关于ActiveMQ的简单整理（2）

今天使用java代码实现Activemq的应答模式。即消息发送者发送一个消息给消费者，消费者接收后返回一个成功消息给发送者。

package com.tgb.model;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;

public class MessageReply {

public static void main(String[] args) throws Exception {
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
factory.setTrustAllPackages(true);
Connection connection = factory.createConnection();
connection.start();
// 消息发送使用这个Queue
Queue queue = new ActiveMQQueue("oneQueue");

// 消息回复使用这个Queue
Queue replyQueue = new ActiveMQQueue("replyQueue");
final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

// 创建一个消息，并设置它的JMSReplyTo为replyQueue。
User u=new User();
u.setUsername("aa");
u.setMobile("xx");
Message message = session.createObjectMessage(u);
message.setJMSReplyTo(replyQueue);
MessageProducer producer = session.createProducer(queue);
producer.send(message);
// 消息的接收者
MessageConsumer comsumer = session.createConsumer(queue);
comsumer.setMessageListener(new MessageListener() {
public void onMessage(Message m) {
try {
User user=(User) ((ObjectMessage)m).getObject();
System.out.println("发送成功："+user.getUsername()+"_"+user.getMobile());

// 创建一个新的MessageProducer来发送一个回复消息。
MessageProducer producer = session.createProducer(m.getJMSReplyTo());
producer.send(session.createTextMessage("回复成功"));
} catch (JMSException e1) {
e1.printStackTrace();
}
}
});
// 这个接收者用来接收回复的消息
MessageConsumer comsumer2 = session.createConsumer(replyQueue);
comsumer2.setMessageListener(new MessageListener() {
public void onMessage(Message m) {
try {
System.out.println(((TextMessage) m).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
}
}

运行成功后的结果：

发送成功：aa_xx
回复成功

其中，传输对象消息时必须需要以下代码：

factory.setTrustAllPackages(true);

否则会出现以下异常：

javax.jms.JMSException: Failed to build body from content. Serializable class not available to broker. Reason: java.lang.ClassNotFoundException: Forbidden class com.tgb.model.User! This class is not trusted to be serialized as ObjectMessage payload. Please take a look at http://activemq.apache.org/objectmessage.html for more information on how to configure trusted classes.

具体配置及代码可以参照异常中提供的网址。