JMS一些实例十一到实例十二
?
深入掌握JMS(十一):TemporaryQueue和TemporaryTopic
TemporaryQueue和TemporaryTopic,从字面上就可以看出它们是“临时”的目的地。可以通过Session来创建,例如:
??? TemporaryQueue replyQueue = session.createTemporaryQueue();
??? 虽然它们是由Session来创建的,但是它们的生命周期确实整个Connection。如果在一个Connection上创建了两个Session,则一个Session创建的TemporaryQueue或TemporaryTopic也可以被另一个Session访问。那如果这两个Session是由不同的Connection创建,则一个Session创建的TemporaryQueue不可以被另一个Session访问。
??? 另外,它们的主要作用就是用来指定回复目的地, 即作为JMSReplyTo。
??? 在下面的例子中,先创建一个Connection,然后创建两个Session,其中一个Session创建了一个TemporaryQueue,另一个Session在这个TemporaryQueue上读取消息。
import javax.jms.*; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.command.ActiveMQQueue; public class TemporaryQueueTest { public static void main(String[] args) throws Exception { ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost"); Connection connection = factory.createConnection(); connection.start(); Queue queue = new ActiveMQQueue("testQueue2"); final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //使用session创建一个TemporaryQueue。 TemporaryQueue replyQueue = session.createTemporaryQueue(); //接收消息,并回复到指定的Queue中(即replyQueue) MessageConsumer comsumer = session.createConsumer(queue); comsumer.setMessageListener(new MessageListener(){ public void onMessage(Message m) { try { System.out.println("Get Message: " + ((TextMessage)m).getText()); MessageProducer producer = session.createProducer(m.getJMSReplyTo()); producer.send(session.createTextMessage("ReplyMessage")); } catch (JMSException e) { } } }); //使用同一个Connection创建另一个Session,来读取replyQueue上的消息。 Session session2 = connection.createSession(true, Session.AUTO_ACKNOWLEDGE); MessageConsumer replyComsumer = session2.createConsumer(replyQueue); replyComsumer.setMessageListener(new MessageListener(){ public void onMessage(Message m) { try { System.out.println("Get reply: " + ((TextMessage)m).getText()); } catch (JMSException e) { } } }); MessageProducer producer = session.createProducer(queue); TextMessage message = session.createTextMessage("SimpleMessage"); message.setJMSReplyTo(replyQueue); producer.send(message); } } 运行结果为: Get Message: SimpleMessage Get reply: ReplyMessage 如果将: Session session2 = connection.createSession(true, Session.AUTO_ACKNOWLEDGE); 更改为: Connection connection2 = factory.createConnection(); Session session2 = connection2.createSession(true, Session.AUTO_ACKNOWLEDGE); 就会得到类似于下面的异常: Exception in thread "main" javax.jms.InvalidDestinationException: Cannot use a Temporary destination from another Connection。
?
?
?
?
深入掌握JMS(十二):MDB
在EJB3中,一个MDB(消息驱动Bean)就是一个实现了MessageListener接口的POJO。下面就是一个简单的MDB。
@MessageDriven(activationConfig={ @ActivationConfigProperty(propertyName="destinationType", propertyValue="javax.jms.Queue"), @ActivationConfigProperty(propertyName="destination", propertyValue="queue/testQueue")}) public class SimpleMDB implements MessageListener { public void onMessage(Message message) { try { System.out.println("Receive Message : " + ((TextMessage)message).getText()); } catch (JMSException e) { e.printStackTrace(); } } } 它要求必须标注为@MessageDriven。它所监听Destination通过标注属性来注入。 下面是一个发送消息的StatelessBean: @Remote public interface IMessageSender { public void sendMessage(String content) throws Exception; } @Stateless @Remote public class MessageSender implements IMessageSender { @Resource(mappedName="ConnectionFactory") private ConnectionFactory factory; @Resource(mappedName="queue/testQueue") private Queue queue; public void sendMessage(String content) throws Exception { Connection cn = factory.createConnection(); Session session = cn.createSession(false, Session.AUTO_ACKNOWLEDGE); MessageProducer producer = session.createProducer(queue); producer.send(session.createTextMessage(content)); } } 这个EJB只有一个方法SendMessage。ConnectionFactory和Queue通过标注注入。 接下来是客户端: public class MessageSenderClient { public static void main(String[] args) throws Exception { Properties props = new Properties(); props.setProperty(Context.INITIAL_CONTEXT_FACTORY, "org.jnp.interfaces.NamingContextFactory"); props.setProperty(Context.PROVIDER_URL, "localhost:2099"); Context context = new InitialContext(props); IMessageSender messageSender = (IMessageSender) context.lookup("MessageSender/remote"); messageSender.sendMessage("Hello"); } } 它通过JNDI查找到上面的EJB,然后调用sengMessage.
?