博客

服务端向queue 发送一条消息， 并等待客户端进行处理，然后接收客户端返回的消息

服务端在向客户端发送消息的时候设置了JMSReplyTo  ,msg.setJMSReplyTo(responseQueue);

然后客户端取得responseQueue  ，并将返回的消息发送到responseQueue，以便服务端进行接收

但是responseQueue  队列里的消息有很多，需要区分哪一条是服务器端想要接收和，于是需要进行消息过滤

MessageConsumer (QueueReceiver, TopicSubscriber 的你类) 消息处理类有两种处理方式 ，同步或异步

receive()  同步

setMessageListener(MessageListener listener)  异步

server

msg.setJMSReplyTo(responseQueue);

String filter = “JMSCorrelationID='” + msg.getJMSMessageID() + “‘”;

session.createReceiver(responseQueue, filter);

client

responseMsg.setJMSCorrelationID(msg.getJMSMessageID());

Queue responseQueue = (Queue) msgObj.getJMSReplyTo();

session.createSender(responseQueue).send(responseMsg);

jndi.properties

java.naming.factory.initial=org.apache.activemq.jndi.ActiveMQInitialContextFactory</p> <p> connectionFactoryNames = connectionFactory,queueConnectionFactory,topicConnectionFactry,connF</p> <p>queue.requestQueue=jms.queue1<br /> queue.responseQueue=jms.queue2

package org.jixiuf.chat;<br /> import javax.jms.JMSException;<br /> import javax.jms.MapMessage;<br /> import javax.jms.Queue;<br /> import javax.jms.QueueConnection;<br /> import javax.jms.QueueReceiver;<br /> import javax.jms.QueueSender;<br /> import javax.jms.QueueSession;<br /> import javax.jms.Session;<br /> import javax.jms.TextMessage;<br /> import org.jixiuf.util.JNDIUtil;<br /> public class ChatServer {<br /> QueueConnection conn;<br /> Queue requestQueue;<br /> Queue responseQueue;<br /> QueueSession session;<br /> int age = 10;<br /> public ChatServer() {<br /> conn = JNDIUtil.getQueueConnection();<br /> requestQueue = JNDIUtil.lookupQueue(“requestQueue”);<br /> responseQueue = JNDIUtil.lookupQueue(“responseQueue”);<br /> try {<br /> session = conn.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);<br /> conn.start();//一般情况下，如果只是发送消息，而不接收不需要start() ;<br /> //但是这个发送消息的服务端需要处理客户端反馈回来的消息，所以start();<br /> } catch (JMSException e) {<br /> e.printStackTrace();<br /> }<br /> }<br /> public void request() {<br /> try {<br /> MapMessage msg = session.createMapMessage();<br /> msg.setInt(“age”, (int)(this.age *Math.random()*10));<br /> msg.setJMSReplyTo(responseQueue);<br /> // 注意这里，设置replyto 何处，如此，客户端便可将消息发送到responseQueue，<br /> // 然后下面可以从responseQueue 接收客户端返回的消息<br /> QueueSender sender = session.createSender(requestQueue);<br /> sender.send(msg);</p> <p> // 处理客户端反馈回来的消息<br /> String filter = “JMSCorrelationID='” + msg.getJMSMessageID() + “‘”;<br /> QueueReceiver receiver = session.createReceiver(responseQueue,<br /> filter);<br /> TextMessage resMsg = (TextMessage) receiver.receive(20 * 1000);<br /> if (resMsg == null) {<br /> System.out.print(“客户端超时，无法收到客户端返回的消息；”);<br /> } else {<br /> System.out.println(“客户端返回的信息是：” + resMsg.getText());<br /> }<br /> } catch (JMSException e) {<br /> e.printStackTrace();<br /> }<br /> }<br /> public void exit() {<br /> try {<br /> conn.close();<br /> } catch (JMSException e) {<br /> e.printStackTrace();<br /> }<br /> System.exit(0);<br /> }<br /> public static void main(String[] args) {<br /> ChatServer server = new ChatServer();<br /> server.request();<br /> server.exit();<br /> }<br /> }<br />
package org.jixiuf.util;<br /> import javax.jms.JMSException;<br /> import javax.jms.Queue;<br /> import javax.jms.QueueConnection;<br /> import javax.jms.QueueConnectionFactory;<br /> import javax.jms.Session;<br /> import javax.naming.InitialContext;<br /> import javax.naming.NamingException;<br /> public class JNDIUtil {<br /> static InitialContext ctx = null;<br /> static QueueConnection conn1 = null;<br /> static {<br /> try {<br /> ctx = new InitialContext();<br /> } catch (NamingException e) {<br /> e.printStackTrace();<br /> }<br /> }<br /> public static QueueConnectionFactory lookupQueueConnectionFactory() {<br /> QueueConnectionFactory connf = null;<br /> try {<br /> connf = (QueueConnectionFactory) ctx.lookup(“connF”);<br /> } catch (NamingException e) {<br /> e.printStackTrace();<br /> }<br /> return connf;<br /> }<br /> // 此类维护两个Connection ，一个供server 一个供client ，其实这两个connection 没什么区别，<br /> public static QueueConnection getQueueConnection() {<br /> if (conn1 == null) {<br /> try {<br /> conn1 = lookupQueueConnectionFactory().createQueueConnection();<br /> } catch (JMSException e) {<br /> e.printStackTrace();<br /> }<br /> }<br /> return conn1;<br /> }<br /> public static QueueConnection crateQueueConnection() {<br /> try {<br /> return lookupQueueConnectionFactory().createQueueConnection();<br /> } catch (JMSException e) {<br /> e.printStackTrace();<br /> }<br /> return null;<br /> }<br /> public static Queue lookupQueue(String queueName) {<br /> try {<br /> return (Queue) ctx.lookup(queueName);<br /> } catch (NamingException e) {<br /> e.printStackTrace();<br /> }<br /> return null;<br /> }<br /> public static void closeSession(Session session) {<br /> try {<br /> session.close();<br /> session = null;<br /> } catch (JMSException e) {<br /> e.printStackTrace();<br /> }<br /> }<br /> } 
package org.jixiuf.chat;<br /> import javax.jms.JMSException;<br /> import javax.jms.MapMessage;<br /> import javax.jms.Message;<br /> import javax.jms.MessageListener;<br /> import javax.jms.Queue;<br /> import javax.jms.QueueConnection;<br /> import javax.jms.QueueReceiver;<br /> import javax.jms.QueueSession;<br /> import javax.jms.Session;<br /> import javax.jms.TextMessage;<br /> import org.jixiuf.util.JNDIUtil;<br /> public class ChatClient implements MessageListener {<br /> QueueConnection conn;<br /> Queue requestQueue;<br /> QueueSession session;<br /> public ChatClient() {<br /> conn = JNDIUtil.getQueueConnection();<br /> requestQueue = JNDIUtil.lookupQueue(“requestQueue”);<br /> try {<br /> session = conn.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);<br /> conn.start();<br /> QueueReceiver receiver = session.createReceiver(requestQueue);<br /> receiver.setMessageListener(this);<br /> } catch (JMSException e) {<br /> e.printStackTrace();<br /> }<br /> }<br /> @Override<br /> public void onMessage(Message msgObj) {<br /> System.out.println(msgObj.getClass());<br /> if (!(msgObj instanceof MapMessage)) {<br /> return;<br /> }<br /> MapMessage msg = (MapMessage) msgObj;<br /> try {<br /> int requestAge = msg.getInt(“age”);<br /> System.out.println(requestAge);<br /> TextMessage responseMsg = session.createTextMessage();<br /> responseMsg.setJMSCorrelationID(msg.getJMSMessageID());<br /> // 这里将responseMsg 的JMSCorrelationID 设成msg.getJMSMessageID ,<br /> // 因此，服务器端只需要处理responseMsg 中JMSCorrelationID==msg.getJMSMessageID<br /> // 的消息<br /> if (requestAge > 50) {<br /> responseMsg.setText(“年龄大于50 ，不批准修改”);<br /> } else {<br /> responseMsg.setText(“请求通过”);<br /> }<br /> Queue responseQueue = (Queue) msgObj.getJMSReplyTo();<br /> session.createSender(responseQueue).send(responseMsg); // 将responseMsg<br /> // 发送到responseQueue队列中，以便服务器端进行处理<br /> } catch (JMSException e) {<br /> e.printStackTrace();<br /> }<br /> }<br /> public void exit() {<br /> try {<br /> conn.close();<br /> } catch (JMSException e) {<br /> e.printStackTrace();<br /> }<br /> System.exit(0);<br /> }<br /> public static void main(String[] args) throws JMSException {<br /> ChatClient client = new ChatClient();</p> <p> }<br /> }<br />

过滤

filter

server

String uuid = UUID.randomUUID().toString();

msg.setStringProperty(“uuid”, uuid);

String filter = “JMSCorrelationID='” + uuid + “‘”;

QueueReceiver receiver = session.createReceiver(responseQueue, filter);

client

responseMsg.setJMSCorrelationID(msg.getStringProperty(“uuid”));