三个java文件： JMSTest.java package com.lizongbo.jmsdemo; import javax.jms.MessageListener;
import javax.naming.Context;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueConnection;
import javax.jms.QueueSession;
import javax.jms.QueueSender;
import javax.jms.QueueReceiver;
import javax.jms.Queue;
import java.io.Serializable;
import java.util.Properties;
import javax.jms.Message;
import javax.jms.JMSException; /**
<p>Title:JMS demo </p> <p>Description: </p> <p>Copyright: Copyright (c) 2005</p> <p>Company: </p> @author lizongbo
@version 1.0
* Use this class to send and receive point-to-point messages.
* To send a text message:
* <code>
* JMSTest jMSTest = new JMSTest();
* jMSTest.sendText("Hello world");
* jMSTest.close(); //Release resources
* </code>
* To receive a message:
* <code>
* JMSTest jMSTest = new JMSTest();
* jMSTest.getQueueReceiver();
* </code>
*/
public class JMSTest
implements MessageListener {
private static Context context;
private boolean transacted = true;
private int acknowledgementMode = javax.jms.Session.AUTO_ACKNOWLEDGE;
private QueueConnectionFactory queueConnectionFactory;
private QueueConnection queueConnection;
private QueueSession queueSession;
private QueueSender queueSender;
private QueueReceiver queueReceiver;
private Queue queue;
private String queueConnectionFactoryName = "testjms";
private String sendQueueName = "com.lizongbo.jms001";
private String recvQueueName = "com.lizongbo.jms001";
public void setTransacted(boolean transacted) {
this.transacted = transacted;
} public void sendObject(Serializable message) throws Exception {
javax.jms.ObjectMessage objectMessage = getQueueSession().
createObjectMessage();
objectMessage.clearBody();
objectMessage.setObject(message);
getQueueSender().send(objectMessage);
if (isTransacted()) {
getQueueSession().commit();
}
} public void sendText(String message) throws Exception {
javax.jms.TextMessage textMessage = getQueueSession().createTextMessage();
textMessage.clearBody();
textMessage.setText(message);
getQueueSender().send(textMessage);
if (isTransacted()) {
getQueueSession().commit();
}
} public QueueReceiver getQueueReceiver() throws Exception {
if (queueReceiver == null) {
queueReceiver = getQueueSession().createReceiver(getRecvQueue());
queueReceiver.setMessageListener(this);
}
return queueReceiver;
} public QueueSender getQueueSender() throws Exception {
if (queueSender == null) {
queueSender = getQueueSession().createSender(getSendQueue());
}
return queueSender;
} public Queue getRecvQueue() throws Exception {
if (queue == null) {
Object obj = getContext().lookup(recvQueueName);
queue = (Queue) obj;
}
return queue;
} public Queue getSendQueue() throws Exception {
if (queue == null) {
Object obj = getContext().lookup(sendQueueName);
queue = (Queue) obj;
}
return queue;
} public QueueSession getQueueSession() throws Exception {
if (queueSession == null) {
queueSession = getQueueConnection().createQueueSession(isTransacted(),
getAcknowledgementMode());
}
return queueSession;
} public QueueConnection getQueueConnection() throws Exception {
if (queueConnection == null) {
queueConnection = getQueueConnectionFactory().createQueueConnection();
queueConnection.start();
}
return queueConnection;
} public QueueConnectionFactory getQueueConnectionFactory() throws Exception {
if (queueConnectionFactory == null) {
Object obj = getContext().lookup(queueConnectionFactoryName);
queueConnectionFactory = (QueueConnectionFactory) obj;
}
return queueConnectionFactory;
} public void setRecvQueueName(String recvQueueName) {
this.recvQueueName = recvQueueName;
} public String getRecvQueueName() {
return recvQueueName;
} public void setSendQueueName(String sendQueueName) {
this.sendQueueName = sendQueueName;
} public String getSendQueueName() {
return sendQueueName;
} public void setQueueConnectionFactoryName(String queueConnectionFactoryName) {
this.queueConnectionFactoryName = queueConnectionFactoryName;
} public String getQueueConnectionFactoryName() {
return queueConnectionFactoryName;
} public void setAcknowledgementMode(int acknowledgementMode) {
this.acknowledgementMode = acknowledgementMode;
} public int getAcknowledgementMode() {
return acknowledgementMode;
} public boolean isTransacted() {
return transacted;
} public static void main(String[] args) throws Exception {
JMSTest jt = new JMSTest();
jt.sendText("test jms Text message!!!--lizongbo " + new java.util.Date().toString());
jt.sendText("测试jms文本消息！！！--lizongbo" + new java.util.Date().toString());
} public Context getInitialContext() throws Exception {
// String url = "t3://*.*:7001";
// String user = null;
// String password = null;
String url = "t3://127.0.0.1:7001";
String user = "weblogic";
String password = "weblogic"; Properties properties;
try {
properties = new Properties();
properties.put(Context.INITIAL_CONTEXT_FACTORY,
"weblogic.jndi.WLInitialContextFactory");
properties.put(Context.PROVIDER_URL, url);
if (user != null) {
properties.put(Context.SECURITY_PRINCIPAL, user);
properties.put(Context.SECURITY_CREDENTIALS,
password == null ? "" : password);
}
return new javax.naming.InitialContext(properties);
}
catch (Exception e) {
System.out.println("Unable to connect to WebLogic server at " + url);
System.out.println("Please make sure that the server is running.");
throw e;
}
} private Context getContext() throws Exception {
if (context == null) {
try {
context = getInitialContext();
}
catch (Exception ex) {
ex.printStackTrace();
throw ex;
}
}
return context;
} public void onMessage(Message message) {
if (message instanceof javax.jms.BytesMessage) {
javax.jms.BytesMessage bytesMessage = (javax.jms.BytesMessage) message;
System.out.println("这是一个BytesMessage，内容是：" + bytesMessage);
/** @todo Process bytesMessage here */
}
else if (message instanceof javax.jms.MapMessage) {
javax.jms.MapMessage mapMessage = (javax.jms.MapMessage) message;
System.out.println("这是一个MapMessage，内容是：" + mapMessage);
/** @todo Process mapMessage here */
}
else if (message instanceof javax.jms.ObjectMessage) {
javax.jms.ObjectMessage objectMessage = (javax.jms.ObjectMessage) message;
System.out.println("这是一个objectMessage，内容是：" + objectMessage);
/** @todo Process objectMessage here */
}
else if (message instanceof javax.jms.StreamMessage) {
javax.jms.StreamMessage streamMessage = (javax.jms.StreamMessage) message;
System.out.println("这是一个StreamMessage，内容是：" + streamMessage);
/** @todo Process streamMessage here */
}
else if (message instanceof javax.jms.TextMessage) {
javax.jms.TextMessage objectMessage = (javax.jms.TextMessage) message;
System.out.println("这是一个TextMessage，内容是：" + objectMessage);
/** @todo Process textMessage here */
}
if (isTransacted()) {
try {
getQueueSession().commit();
}
catch (Exception ex) {
ex.printStackTrace();
}
}
} public void close() throws Exception {
if (queueSender != null) {
queueSender.close();
}
if (queueReceiver != null) {
queueReceiver.close();
}
if (queueSession != null) {
queueSession.close();
}
if (queueConnection != null) {
queueConnection.close();
}
}
}
JMSFetchTest.java package com.lizongbo.jmsdemo;
/**
*
<p>Title:JMS demo </p> <p>Description: </p> <p>Copyright: Copyright (c) 2005</p> <p>Company: </p> @author lizongbo
@version 1.0
*/
public class JMSFetchTest {
public JMSFetchTest() {
super();
} public static void main(String[] args) throws Exception {
JMSTest jt = new JMSTest();
jt.getQueueReceiver();
synchronized (jt) {
jt.wait(10000);
}
System.out.println("jieshu!!!");
//jt.sendObject("test jms message!!!--lizongbo");
//jt.sendObject("测试jms消息！！！--lizongbo");
} }
JMSpublicTest.java package com.lizongbo.jmsdemo; import javax.jms.MessageListener;
import javax.naming.Context;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicConnection;
import javax.jms.TopicSession;
import javax.jms.TopicPublisher;
import javax.jms.TopicSubscriber;
import javax.jms.Topic;
import java.util.Properties;
import java.io.Serializable;
import javax.jms.Message; /**
<p>Title:JMS demo </p> <p>Description: </p> <p>Copyright: Copyright (c) 2005</p> <p>Company: </p> @author lizongbo
@version 1.0
* Use this class to publish and subscribe to messages.
* To send a text message:
* <code>
* JMSpublicTest jMSpublicTest = new JMSpublicTest();
* jMSpublicTest.publishText("Hello world");
* jMSpublicTest.close(); //Release resources
* </code>
* To receive a message:
* <code>
* JMSpublicTest jMSpublicTest = new JMSpublicTest();
* jMSpublicTest.getTopicSubscriber();
* </code>
*/
public class JMSpublicTest
implements MessageListener {
private static Context context;
private boolean transacted = false;
private int acknowledgementMode = javax.jms.Session.AUTO_ACKNOWLEDGE;
private TopicConnectionFactory topicConnectionFactory;
private TopicConnection topicConnection;
private TopicSession topicSession;
private TopicPublisher topicPublisher;
private TopicSubscriber topicSubscriber;
private Topic topic;
private String topicConnectionFactoryName = "testjms";
private String publishTopicName = "com.lizongbo.jms002";
private String subscribeTopicName = "com.lizongbo.jms002";
private String clientId = "";
private String durableName = "";
private boolean durable = false;
public void setTransacted(boolean transacted) {
this.transacted = transacted;
} public TopicSubscriber getTopicSubscriber() throws Exception {
if (topicSubscriber == null) {
if (isDurable()) {
topicSubscriber = getTopicSession(true).createDurableSubscriber(
getSubscribeTopic(), getDurableName());
}
else {
topicSubscriber = getTopicSession(true).createSubscriber(
getSubscribeTopic());
}
topicSubscriber.setMessageListener(this);
getTopicConnection(true).start();
}
return topicSubscriber;
} public TopicPublisher getTopicPublisher() throws Exception {
if (topicPublisher == null) {
topicPublisher = getTopicSession(false).createPublisher(getPublishTopic());
}
return topicPublisher;
} public Topic getPublishTopic() throws Exception {
if (topic == null) {
Object obj = getContext().lookup(publishTopicName);
topic = (Topic) obj;
}
return topic;
} public TopicSession getTopicSession(boolean consumer) throws Exception {
if (topicSession == null) {
topicSession = getTopicConnection(consumer).createTopicSession(
isTransacted(), getAcknowledgementMode());
}
return topicSession;
} public TopicConnection getTopicConnection(boolean consumer) throws Exception {
if (topicConnection == null) {
topicConnection = getTopicConnectionFactory().createTopicConnection();
if (isDurable() && consumer) {
topicConnection.setClientID(clientId);
}
topicConnection.start();
}
return topicConnection;
}
public TopicConnectionFactory getTopicConnectionFactory() throws Exception {
if (topicConnectionFactory == null) {
Object obj = getContext().lookup(topicConnectionFactoryName);
topicConnectionFactory = (TopicConnectionFactory) obj;
}
return topicConnectionFactory;
} public void setDurable(boolean durable) {
this.durable = durable;
} public boolean isDurable() {
return durable;
} public void setDurableName(String durableName) {
this.durableName = durableName;
} public String getDurableName() {
return durableName;
} public void setClientId(String clientId) {
this.clientId = clientId;
} public String getClientId() {
return clientId;
} public void setSubscribeTopicName(String subscribeTopicName) {
this.subscribeTopicName = subscribeTopicName;
} public String getSubscribeTopicName() {
return subscribeTopicName;
} public void setPublishTopicName(String publishTopicName) {
this.publishTopicName = publishTopicName;
} public String getPublishTopicName() {
return publishTopicName;
} public void setTopicConnectionFactoryName(String topicConnectionFactoryName) {
this.topicConnectionFactoryName = topicConnectionFactoryName;
} public String getTopicConnectionFactoryName() {
return topicConnectionFactoryName;
} public void setAcknowledgementMode(int acknowledgementMode) {
this.acknowledgementMode = acknowledgementMode;
} public int getAcknowledgementMode() {
return acknowledgementMode;
} public boolean isTransacted() {
return transacted;
} public static void main(String[] args) throws InterruptedException, Exception { JMSpublicTest jtc = new JMSpublicTest();
jtc.setDurable(true);
jtc.setClientId("ip192.168.9.226");
jtc.setDurableName("test226aa");
jtc.getTopicSubscriber();
//开始发布一个消息
JMSpublicTest jtp = new JMSpublicTest();
jtp.setDurable(true);
jtp.setClientId("ip192.168.9249");
jtp.setDurableName("test226aa");
jtp.publishText("发布一个消息！！！");
jtp.close();
//在这里保持监听来获取消息
synchronized (jtc) {
jtc.wait(10000);
}
jtc.close();
System.out.println(jtp.getClientId());
System.out.println("jieshu!!!"); } public Context getInitialContext() throws Exception {
// String url = "t3://*.*:7001";
// String user = null;
// String password = null;
String url = "t3://127.0.0.1:7001";
String user = "weblogic";
String password = "weblogic"; Properties properties;
try {
properties = new Properties();
properties.put(Context.INITIAL_CONTEXT_FACTORY,
"weblogic.jndi.WLInitialContextFactory");
properties.put(Context.PROVIDER_URL, url);
if (user != null) {
properties.put(Context.SECURITY_PRINCIPAL, user);
properties.put(Context.SECURITY_CREDENTIALS,
password == null ? "" : password);
}
return new javax.naming.InitialContext(properties);
}
catch (Exception e) {
System.out.println("Unable to connect to WebLogic server at " + url);
System.out.println("Please make sure that the server is running.");
throw e;
}
} private Context getContext() throws Exception {
if (context == null) {
try {
context = getInitialContext();
}
catch (Exception ex) {
ex.printStackTrace();
throw ex;
}
}
return context;
} public Topic getSubscribeTopic() throws Exception {
if (topic == null) {
Object obj = getContext().lookup(subscribeTopicName);
topic = (Topic) obj;
}
return topic;
} public void publishText(String message) throws Exception {
javax.jms.TextMessage textMessage = getTopicSession(false).
createTextMessage();
textMessage.clearBody();
textMessage.setText(message);
getTopicPublisher().publish(textMessage);
if (isTransacted()) {
getTopicSession(false).commit();
}
} public void publishObject(Serializable message) throws Exception {
javax.jms.ObjectMessage objectMessage = getTopicSession(false).
createObjectMessage();
objectMessage.clearBody();
objectMessage.setObject(message);
getTopicPublisher().publish(objectMessage);
if (isTransacted()) {
getTopicSession(false).commit();
}
} public void onMessage(Message message) {
if (message instanceof javax.jms.BytesMessage) {
javax.jms.BytesMessage bytesMessage = (javax.jms.BytesMessage) message;
System.out.println("这是一个BytesMessage，内容是：" + bytesMessage);
/** @todo Process bytesMessage here */
}
else if (message instanceof javax.jms.MapMessage) {
javax.jms.MapMessage mapMessage = (javax.jms.MapMessage) message;
System.out.println("这是一个MapMessage，内容是：" + mapMessage);
/** @todo Process mapMessage here */
}
else if (message instanceof javax.jms.ObjectMessage) {
javax.jms.ObjectMessage objectMessage = (javax.jms.ObjectMessage) message;
System.out.println("这是一个objectMessage，内容是：" + objectMessage);
/** @todo Process objectMessage here */
}
else if (message instanceof javax.jms.StreamMessage) {
javax.jms.StreamMessage streamMessage = (javax.jms.StreamMessage) message;
System.out.println("这是一个StreamMessage，内容是：" + streamMessage);
/** @todo Process streamMessage here */
}
else if (message instanceof javax.jms.TextMessage) {
javax.jms.TextMessage objectMessage = (javax.jms.TextMessage) message;
System.out.println("这是一个TextMessage，内容是：" + objectMessage);
/** @todo Process textMessage here */
}
if (isTransacted()) {
try {
getTopicSession(false).commit();
}
catch (Exception ex) {
ex.printStackTrace();
}
}
} public void close() throws Exception {
if (topicPublisher != null) {
topicPublisher.close();
}
if (topicSubscriber != null) {
topicSubscriber.close();
}
if (topicSession != null) {
topicSession.close();
}
if (topicConnection != null) {
topicConnection.close();
}
}
}
|
本类阅读TOP10