目标：使用开源JMS应用框架Apache ActiveMQ，实现P2P方式下Request-Response Socket通信
模式：P2P
特征：发送端可以实时发送请求并取得响应，接收端可以实时获取请求并返回响应
库：activemq-all-5.11.1.jar
前提：启动activemq消息服务程序
--------------------------------------------------------------------------------------------
发送端：
package merrick.activemq;
import java.util.UUID;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
public class Sender_p2p {
private static String user = ActiveMQConnection.DEFAULT_USER;
private static String password =ActiveMQConnection.DEFAULT_PASSWORD;
private static String url = "tcp://localhost:61616";
public static final String LONGMSG = "123456789012345678901234567890123456789012345678901234567890";
public static void main(String[] args) {
try {
new Sender_p2p().send(UUID.randomUUID().toString()+LONGMSG);
} catch (Throwable e) {
e.printStackTrace();
}
}
/***
* P2P模式
* Request-Response
* 类似：发送请求同步获取响应
* 此处可应用为TCP SOCKET Client端
*
* */
public void send(String msg) throws Exception {
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user,password,url);
Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
Queue destination = session.createQueue("Request&Response_X");
MessageProducer producer = session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
TextMessage message = session.createTextMessage(msg);
Destination tempDest = session.createTemporaryQueue();
MessageConsumer responseConsumer = session.createConsumer(tempDest);
message.setJMSReplyTo(tempDest);
String correlationId = UUID.randomUUID().toString();
message.setJMSCorrelationID(correlationId); //
responseConsumer.setMessageListener(new MessageListener() {//监听回应
@Override
public void onMessage(Message arg0) {
try {
TextMessage textMessage=(TextMessage)arg0;
System.out.println(textMessage.getText()); //显示接收者的Response
} catch (JMSException e) {
e.printStackTrace();
} finally {//得到回应后关闭会话和连接
try {
session.commit();
session.close();
connection.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}});
producer.send(message); //发送消息
System.out.println("<"+Thread.currentThread().getId() + "> send ok: "+msg);
session.commit();
}
}
--------------------------------------------------------------------------------------------
接收端：
package merrick.activemq;
import java.util.UUID;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
public class Receiver_p2p {
private static String user = ActiveMQConnection.DEFAULT_USER;
private static String password =ActiveMQConnection.DEFAULT_PASSWORD;
private static String url = "tcp://localhost:61616";
public static void main(String[] args) {
try {
Receiver_p2p obj = new Receiver_p2p();
obj.receive();
} catch (Exception e) {
e.printStackTrace();
}
}
/***
* P2P模式
* Request-Response
* 类似：获取即时内容，同步返回响应
* (但Session不能关闭)
* 此处可应用为TCP SOCKET Server端
* */
public void receive() throws Exception{
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user,password,url);
Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
Queue destination = session.createQueue("Request&Response_X");
MessageConsumer consumer = session.createConsumer(destination);
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) { //监听消息
try {
TextMessage textMessage=(TextMessage)message;
System.out.println("[Received]"+textMessage.getText()); //取得消息
Message response = session.createTextMessage(UUID.randomUUID().toString() + " [from receiver] "+ textMessage.getText().length()+", " + textMessage.getText());//及时给予Response
response.setJMSCorrelationID(message.getJMSCorrelationID());
MessageProducer producer = session.createProducer(null);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
producer.send(message.getJMSReplyTo(),response); //发送回应
} catch (JMSException e) {
e.printStackTrace();
} finally {
try {
session.commit();
session.close();
connection.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
});
}
}