JMS-详解

1.MQ和J2EEAPI的关系JDBC：JMS客户端可以使用JDBC接口，可以将JDBC和JMS包含在一个事务里。这种包含可以在EJB里，也可以直接调用JTA(JavaTransactionAPI)接口实现。JavaBeans：JavaBeans可以用JMSSession发送或接收消息。EJB：EJB2.0规范中定义了新的Message-DrivenBeans组件模型，它可以作为消息接受者异步接收消息。JTA(JavaTransactionAPI)：JMS客户端可以用JTA启动事务。JMSProvider可以选择是否支持分布式事务。JTS(JavaTransactionService)：JMS可以和JTS一起组成一个分布式事务，如将发送接收消息和更新数据库包含在一个事务里。JNDI：JMS客户端通过JNDI调用JMS中的对象。AMQ还可以作为Apusic应用服务器J2EEConnectorArchitecture资源提供者，可以通过资源适配器访问AMQ。同时，AMQ也可以通过提供资源适配器作为Weblogic,Websphere的消息提供者。2.JMS编程模型图6.1.JMS编程模型ConnectionFactoryConnectionFactory是在jms.xml文件事先定义好的，用来创建Connection的工厂Contextctx=newInitialContext();ConnectionFactoryconnectionFactory=(ConnectionFactory)ctx.lookup(JNDI_NAME);如果应用程序与服务器不在同一个虚拟机时Hashtableenv=newHashtable();env.put(Context.INITIAL_CONTEXT_FACTORY,com.apusic.naming.jndi.CNContextFactory);env.put(Context.PROVIDER_URL,iiop://hostname:4888);//取决于地址和端口newInitialContext(env);ConnectionFactoryconnectionFactory=(ConnectionFactory)ctx.lookup(JNDI_NAME);DestinationsDestinations是指消息发送客户端的消息目标和消息接收客户端的消息来源，它也是预先在jms.xml定义好的。对于PTP类型，Destinations对应的类型为Queue，对于Pub/Sub类型，Destinations对应的类型为Topic。获取方式：DestinationmyDest=(Destination)ctx.lookup(testTopic);QueuemyQueue=(Queue)ctx.lookup(testQueue);或者：//通过下面这种方式创建出来的Queue，代表一个远程AMQ服务器上的队列，@符号之前//是队列的名称，@符号之后是服务器路由名。发送消息到该队列时，会将消息路由到@符号之后//指定的服务器，并将消息放进@符号之前指定的队列。QueuemessageQueue=queueSession.createQueue(testQueue@ServerRouterName);Connections为客户端和消息服务器建立的一个连接，它可以创建Session，并且可以创建多个Session(稍后介绍)。获取方法：Connectionconnection=connectionFactory.createConnection();当使用完连接的时候，你需要关闭连接，否则该连接会一直保持。关闭连接的时候它同时会关闭该连接产生的Session。connection.close();连接创建好后，一般先调用start()方法。SessionsSessions是单线程的，用来创建消息发送者和消息接受者的上下文。Sessions同时还可以执行MessageListeners获取方法：Sessionsession=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);Sessionsession=connection.createSession(false,Session.CLIENT_ACKNOWLEDGE);Sessionsession=connection.createSession(false,Session.DUPS_OK_ACKNOWLEDGE);三种应答模式的差别，请参照AMQ相关概念中的应答模式。MessageProducersmessageproducer是由Session创建的一个对象，它是用来向Destinations发送信息的。MessageProducerproducer=session.createProducer(myDest);MessageProducerproducer=session.createProducer(myTopic);或者：QueueSendersender=queueSession.createSender(messageQueue);也可以创建时不指定Destinations，发送时再指定。创建好后，可以继续创建消息，然后调用发送方法发送消息。TextMessagemessage=session.createTextMessage();message.setText(msg_text);producer.send(message);当发送时指定Destinations，你的使用方法如下：MessageProduceranon_prod=session.createProducer(null);anon_prod.send(myQueue,message);MessageConsumersmessageconsumer是指从destination接收消息的对象MessageConsumerconsumer=session.createConsumer(myQueue);MessageConsumerconsumer=session.createConsumer(myTopic);你还能使用Session.createDurableSubscriber创建一个持久的Topic注册对象。它和普通的Topic注册对象的区别是不管当前是否在线，均能收到对应Destination的全部信息StringsubName=MySub;MessageConsumertopicSubscriber=session.createDurableSubscriber(myTopic,subName);停止持久的Topic注册对象使用以下方法。topicSubscriber.close();session.unsubscribe(MySub);使用Consumer前应该先调用connection.start()方法connection.start();Messagem=consumer.receive();//或者connection.start();Messagem=consumer.receive(3000);//3秒后超时MessageListeners这是为异步调用设计的一种方式。它是把消息监听器注册到consumer中，然后消息监听器实现一个onMesseag()方法，当消息到达是，会调用消息监听器的onMesseag()方法。ListenermyListener=newListener();consumer.setMessageListener(myListener);MessageSelectorsAMQ允许消息选择性的接收，语法符合SQL92。例如：客户只收取消息属性NewsType为Sports和Opinion的消息。NewsType='Sports'ORNewsType='Opinion'然后再创建Consumer时指定Selectors。例如createConsumer(Destinationdestination,StringmessageSelector)createConsumer(Destinationdestination,StringmessageSelector,booleanNoLocal)createDurableSubscriber(Topictopic,Stringname,StringmessageSelector,booleannoLocal)把其中的messageSelector参数变成NewsType='Sports'ORNewsType='Opinion'即可。6.3.客户端方式使用JMS6.3.1.开发JMSClient的步骤一般来说，一个JMS应用是几个JMS客户端交换消息，开发JMS客户端应用由以下几步构成：用JNDI寻找得到ConnectionFactory对象；用JNDI寻找或者Session的CreateQueue（CreateTopic）方法得到目标队列或主题对象，即Destination对象；用ConnectionFactory创建Connection对象；用Connection对象创建一个或多个JMSSession；用Session和Destination创建MessageProducer和MessageConsumer；通知Connection开始传递消息。6.3.2.PTP模型应用PTP模型主要包含消息的发送和接收，下面我们举例说明：发送消息1.首先编辑发送节点的config/jms.xml文件，加上以下一段配置：...connection-factorydescriptionThedefaultqueueconnectionfactory/descriptiondisplay-nameQueueConnectionFactory/display-namejndi-namejms/QueueConnectionFactory/jndi-name/connection-factory...（如果同名的connection-factory已经存在，就不用再创建）同时还要编辑消息的目的节点的config/jms.xml文件，加上以下一段配置：...queueclustered=falsequeue-nametestQueue/queue-namejndi-nametestQueue/jndi-name/queue...（如果同名的queue已经存在，就不用再创建）2.启动AMQ由于AMQ目前对配置文件的读取是在启动阶段，所以需要重新启动服务器。3.编写发送消息客户端代码：Send.javaimportjavax.jms.*;importjavax.naming.*;importjava.io.*;importjava.util.*;importjava.rmi.RemoteException;/***Title:JMS*Description:JMSTest*Copyright:Copyright(c)2003*Company:Apusic*@authorApusic*@version1.0*/publicclassSend{StringqueueName=testQueue;//消息的目的队列的队列名StringrouterName=RouterB;//消息的目的节点的路由名QueueConnectionFactoryqueueConnectionFactory=null;Queuequeue=null;QueueConnectionqueueConnection=null;QueueSessionqueueSession=null;QueueSenderqueueSender=null;TextMessagemessage=null;