OpenJMS特性及使用

1、OpenJMS 是一个开源的 Java Message Service API 1.0.2 规范的实现 , 它包含有以下特 性:*. 它既支持点到点(point-to-point (PTP 模型和发布 /订阅(Pub/Sub模型。 *. 支持同步与异步消息发送*. JDBC持久性管理使用数据库表来存储消息*. 可视化管理界面。*. Applet支持。*. 能够与 Jakarta Tomcat这样的 Servlet 容器结合。*. 支持 RMI, TCP, HTTP 与 SSL 协议。*. 客户端验证*. 提供可靠消息传输、事务和消息过滤介绍服务器搭建,在 openJms , 解压后可以直接使用

2、, 在 openjms-0.7.7-beta-1bin 里, 有 openJms 的运行脚本, 执行 startup 启动, 弹出一个新的窗口, 服务就运行在新窗口内, shutdown 为停止命令:服务运行后, 就可以开始使用 JMS 服务了, 至此服务搭建完毕, 简单得不能再简 单了。下 面 是 消 息 发 送 和 接 收 的 开 发 , 开 发 中 需 要 的 jar 包 在 openjms-0.7.7-beta-1lib里可以找到:把上面的类包加入到项目中,下面是消息发送服务的代码:/* author qianli 2011-12-26* For openJms 演示 , Messag

3、e 发送*/public class QueueSend public static void main(String args try /取得 JNDI 上下文和连接Hashtable properties = new Hashtable(;properties.put(Context.INITIAL_CONTEXT_FACTORY,"org.exolab.jms.jndi.InitialContextFactory"/openJms默认的端口是 1099properties.put(Context.PROVIDER_URL,"rmi:/localhost:10

4、99/"Context context = new InitialContext(properties;/获得 JMS 信息连接队列工厂QueueConnectionFactory queueConnectionFactory =(QueueConnectionFactory context.lookup("JmsQueueConnectionFactory"/获得 JMS 信息连接队列QueueConnection queueConnection =queueConnectionFactory.createQueueConnection(;/产生队列 Sessi

5、on ,设置事务为 false ,自动应答消息接收 QueueSession queueSession =queueConnection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE;/获得默认内建在 JMS 里的队列之一:queue1Queue queue = (Queue context.lookup("queue1"/产生 JMS 队列发送器QueueSender queueSender =queueSession.createSender(queue;/发送数据到 JMSUserBean ub = new Use

6、rBean(;ub.setUsername("weida"ub.setPassword("111"ObjectMessage message = queueSession.createObjectMessage(ub;queueSender.send(message;System.out.println("信息写入 JMS 服务器队列 "/以下做清除工作,代码略/ . . catch (Exception e e.printStackTrace(;/* author qianli 2011-12-26* For openJms 演示

7、, 同步接收 Message*/public class QueueReceiveSynchronous public static void main(String args try /取得 JNDI 上下文和连接Hashtable properties = new Hashtable(;properties.put(Context.INITIAL_CONTEXT_FACTORY,"org.exolab.jms.jndi.InitialContextFactory"properties.put(Context.PROVIDER_URL,"rmi:/localho

8、st:1099/"Context context = new InitialContext(properties;/获得 JMS 信息连接队列工厂QueueConnectionFactory queueConnectionFactory =(QueueConnectionFactory context.lookup("JmsQueueConnectionFactory"/获得 JMS 信息连接队列QueueConnection queueConnection =queueConnectionFactory.createQueueConnection(;/启动接收队

9、列线程queueConnection.start(;/产生队列 Session ,设置事务为 false ,自动应答消息接收 QueueSession queueSession =queueConnection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE;/获得默认内建在 JMS 里的队列之一:queue1Queue queue = (Queue context.lookup("queue1"/产生 JMS 队列接收器QueueReceiver queueReceiver =queueSession.createRec

10、eiver(queue;/通过同步的方法接收消息Message message = queueReceiver.receive(;UserBean ub = null;if (message instanceof ObjectMessageub = (UserBean(ObjectMessagemessage.getObject(; System.out.println("username="+ub.getUsername(; System.out.println("password="+ub.getPassword(; /以下做清除工作,代码略/ .

11、. catch (Exception e e.printStackTrace(;/* author qianli 2011-12-26* For openJms 演示 , 非同步接收 Message*/public class QueueReceiveAsynchronous implements MessageListener private UserBean ub = null;QueueReceiveAsynchronous(try /取得 JNDI 上下文和连接Hashtable properties = new Hashtable(;properties.put(Context.IN

12、ITIAL_CONTEXT_FACTORY,"org.exolab.jms.jndi.InitialContextFactory" properties.put(Context.PROVIDER_URL,"rmi:/localhost:1099/"Context context = new InitialContext(properties; /获得 JMS 信息连接队列工厂QueueConnectionFactory queueConnectionFactory =(QueueConnectionFactory context.lookup("

13、;JmsQueueConnectionFactory"/获得 JMS 信息连接队列QueueConnection queueConnection =queueConnectionFactory.createQueueConnection(; /产生队列 Session ,设置事务为 false ,自动应答消息接收 QueueSession queueSession =queueConnection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE;/获得默认内建在 JMS 里的队列之一:queue1Queue queue = (Que

14、ue context.lookup("queue1"/产生 JMS 队列接收器QueueReceiver queueReceiver =queueSession.createReceiver(queue;queueReceiver.setMessageListener(this;/启动接收队列线程queueConnection.start(; catch (Exception e e.printStackTrace(;public static void main(String args System.out.println("非同步消息的接收:"try

15、 QueueReceiveAsynchronous listener =new QueueReceiveAsynchronous(;Thread.sleep(1000; catch (InterruptedException e e.printStackTrace(;public void onMessage(Message message if (message instanceof ObjectMessagetry ub = (UserBean(ObjectMessagemessage.getObject(; catch (JMSException e e.printStackTrace(

16、;if(ub != nullSystem.out.println("username="+ub.getUsername(; System.out.println("password="+ub.getPassword(; ub = null;/* author qianli 2011-12-26* openJms 订阅 /发布模式 发布*/public class TopicSender private TopicPublisher m_topicPublisher;private TopicSession m_topicSession;private i

17、nt binit = 0;public TopicSender(String strtopic try Hashtable properties = new Hashtable(;properties.put(Context.INITIAL_CONTEXT_FACTORY,"org.exolab.jms.jndi.InitialContextFactory"properties.put(Context.PROVIDER_URL, "rmi:/localhost:1099/" /*properties.put(".https.keyStore&q

18、uot;,"E:worksnote_namesrcclient.keystore"properties.put(".https.keyStorePassword","openjms" properties.put(".https.trustStore", "C:jsdk2jrelibsecuritycacerts" */ Context context = new InitialContext(properties; / 获得 JMS Topic 连接队列工厂 TopicConnectionFa

19、ctory factory = (TopicConnectionFactory context .lookup("JmsTopicConnectionFactory" / 创建一个 Topic 连接，并启动 TopicConnection topicConnection factory.createTopicConnection(; topicConnection.start(; / 创建一个 Topic 会话，并设置自动应答 m_topicSession = topicConnection.createTopicSession(false, Session.AUTO_AC

20、KNOWLEDGE; / lookup 得到 topic1 Topic topic = (Topic context.lookup(strtopic; / 用 Topic 会话生成 Topic 发布器 m_topicPublisher = m_topicSession.createPublisher(topic; binit = 1; catch (NamingException e e.printStackTrace(; catch (JMSException e e.printStackTrace(; public void Send(String strmsg if (binit = 0

21、 System.out.println("请先初始化 jms" else try TextMessage m_topicSession.createTextMessage(strmsg; m_topicPublisher.publish(message; catch (JMSException e e.printStackTrace(; = message = public static void main(String args TopicSender send = new TopicSender("topic1" send.Send("he

22、llo" /* * author qianli 2011-12-26 * openJms 非持久订阅同步接收演示 */ public class TopicSubscribeSynchronous public static void main(String args try System.out.println("定购消息接收启动：" /取得 JNDI 上下文和连接 Hashtable properties = new Hashtable(; properties.put(Context.INITIAL_CONTEXT_FACTORY, "org.ex

23、olab.jms.jndi.InitialContextFactory" properties.put(Context.PROVIDER_URL, "rmi:/localhost:1099/" Context context = new InitialContext(properties; /获得 Topic 工厂和 Connection TopicConnectionFactory factory = (TopicConnectionFactory context.lookup( "JmsTopicConnectionFactory" Top

24、icConnection topicConnection factory.createTopicConnection(; topicConnection.start(; /创建 Topic 的会话，用于接收信息 TopicSession topicSession = topicConnection.createTopicSession( false, Session.AUTO_ACKNOWLEDGE; /lookup topic1 Topic topic = (Topic context.lookup("topic1" /创建 Topic subscriber TopicS

25、ubscriber topicSubscriber = topicSession.createSubscriber(topic; /收满 10 条订阅消息则退出 for (int i=0; i<10; i+ /同步消息接收，使用 receive 方法，堵塞等待，直到接收消 息 = TextMessage message = (TextMessage topicSubscriber.receive(; System.out.println(" 接 收 订 阅 消 息 "+i+": " + message.getText(; /资源清除，代码略 . .

26、 System.out.println("订阅接收结束." catch (NamingException e e.printStackTrace(; catch (JMSException e e.printStackTrace(; /* * author qianli 2011-12-26 * openJms 非持久订阅异步接收演示 */ public class TopicSubscribeAsynchronous implements MessageListener private TopicConnection topicConnection; private To

27、picSession topicSession; private Topic topic; private TopicSubscriber topicSubscriber; TopicSubscribeAsynchronous( try /取得 JNDI 上下文和连接 Hashtable properties = new Hashtable(; properties.put( Context.INITIAL_CONTEXT_FACTORY, "org.exolab.jms.jndi.InitialContextFactory" properties.put(Context.PROVIDER_URL, "rmi:/localhost:1099/" Context context = new InitialContext(properties; /取得 Topic 的连接工厂和连接 TopicConnectionFactory topicConnectionFactory = (TopicConnectionFactory context.lookup( "JmsTopicConnectionFactory"