天空中飘着小雨，实在是适合写代码的时节。
1
package ch02.chat; import java.io.Serializable; import javax.jms.Connection;
import javax.jms.ConnectionFactory;
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.ObjectMessage;
import javax.jms.Session;
/*本工具封装了*/ import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber; import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory; public class JMSTopic {
TopicConnectionFactory connectionFactory;
// Connection ：JMS 客户端到JMS Provider 的连接
TopicConnection connection = null;
//用来发布的会话 //TopicSession proSession = null;
//2一个订阅会话
//TopicSession conSession = null;
TopicSession session=null; //主题发布者
MessageProducer producer=null;
//主题
MessageConsumer consumer=null; // Destination ：消息的目的地;消息发送给谁.
Destination destination;
// MessageProducer：消息发送者 //默认构造函数,默认的连接activemq，可以写多个构造函数
public JMSTopic()
{
connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD,
"tcp://localhost:61616");
try {
connection= connectionFactory.createTopicConnection();
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
try {
connection.start();
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} }
public JMSTopic(String user,String name)
{
connectionFactory = new ActiveMQConnectionFactory(
user,
name,
"tcp://localhost:61616");
try {
connection= connectionFactory.createTopicConnection();
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
try {
connection.start();
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} } //设计session类型
public void setSession() throws JMSException
{
session= connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); }
//设置为原子类型
public void setAtomicSession() throws JMSException
{
session= connection.createTopicSession(true, Session.AUTO_ACKNOWLEDGE); } //此处先固定消息为String类型
public void writeMessage(String t,String message,int priority )
{
try { producer=session.createProducer(session.createTopic(t)); //使用message构造TextMessage
TextMessage text=session.createTextMessage();
text.setJMSPriority(priority);
text.setText(message);
producer.send(text); } catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
//创建发布会话应该是可以配置的，此处先固定 } public void writeMessage(String t,Object o)
{
try { producer=session.createProducer(session.createTopic(t)); //使用message构造TextMessage
ObjectMessage text=session.createObjectMessage();
text.setObject((Serializable) o);
producer.send(text); } catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
//创建发布会话应该是可以配置的，此处先固定 } //使用某个Message监听器来监听某个Topic
public void receiveMsg(String c,MessageListener ml)
{
try { Topic t=session.createTopic(c);
consumer=session.createConsumer(t);
//设置过来的监视器
consumer.setMessageListener(ml); } catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} }
public Message receiveMsg(String c)
{
try { Topic t=session.createTopic(c);
consumer=session.createConsumer(t);
//设置过来的监视器
Message message=consumer.receive();
return message; } catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return null;
}
//同步接收信息 public void commit() throws JMSException
{
session.commit();
}
public void rollback() throws JMSException
{
session.rollback();
}
public void close() throws JMSException
{
if(connection!=null) connection.close();
if(session!=null) session.close();
if(producer!=null) session.close();
if(consumer!=null) consumer.close(); } }
2.如何使用呢？
a.做个有关事务的实
package ch02.chat; import java.util.Scanner; import javax.jms.JMSException;
import javax.jms.Session; public class ClientTest {
public static void main(String args[]) throws JMSException
{
//第一个例子，建立一个原子的session做个实验看看，这个一个不会发送任何信息到服务器，
JMSTopic jt=new JMSTopic();
//jt.setSession();
try{ jt.setAtomicSession(); jt.writeMessage("que1", "hansongjiang",4);
int x=10/0; //会抛出异常，实现回滚，所以que1中不会添加任何信息
jt.writeMessage("que1","hansong",4);
jt.commit();
}
catch(Exception e)
{
jt.rollback(); }
finally
{ jt.close();
} //如果
/* jt=new JMSTopic();
try
{
jt.setSession();
jt.writeMessage("que1", "hansongjiang",4);
//int x=10/0;
jt.writeMessage("que1","zhangsan",4); }
catch(Exception e)
{
System.out.println("异常我抓住了"); } */ } }
入队的信息为的信息为个数为0
b.非事务的执行后呢？我们使用getAtomicSession获得的session执行后入topic个数为1
package ch02.chat; import java.util.Scanner; import javax.jms.JMSException;
import javax.jms.Session; public class ClientTest {
public static void main(String args[]) throws JMSException
{ JMSTopic jt=new JMSTopic();
try
{
jt.setSession();
jt.writeMessage("que1", "hansongjiang",4);
//int x=10/0;
jt.writeMessage("que1","zhangsan",4); }
catch(Exception e)
{
System.out.println("异常我抓住了"); } */ } }