Queue 消息的发送与接收(PTP 消息传递模型)

时间：2024-07-13 21:35:08

上篇博客写到了JMS两种消息模型（P2P、pub/sub）《JMS两种消息模型》。本篇博客通过一个实例来进一步了解P2P模型。

Queue消息的发送与接收——PTP消息传递模型，样例：

Queue消息的发送与接收：

1。MyQueeuMDBBeanClient.java(Queue消息的发送者) 

package com.jinb.cn;

import javax.jms.Queue;

import javax.jms.QueueConnection;

import javax.jms.QueueConnectionFactory;

import javax.jms.QueueSender;

import javax.jms.QueueSession;

import javax.jms.TextMessage;

import javax.naming.InitialContext;

public class MyQueeuMDBBeanClient {

    public static void main(String[] args) throws Exception {

       InitialContext context = new InitialContext();

       //获得QueueConnectionFactory对象

       QueueConnectionFactory factory = (QueueConnectionFactory)context.lookup("ConnectionFactory");

       //创建QueueConnection

       QueueConnection connection = (QueueConnection)factory.createConnection();

       /**

         * 通过connection创建QueueSession对象；

         * 当中第一个參数为是否支持事务。TRUE为支持，false为不支持。

         * 若设为true,则须要手动COMMIT;

         * 第二个參数为响应的模式，普通情况下就设为QueueSession.AUTO_ACKNOWLEDGE

         * */

       QueueSession session = (QueueSession)connection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);

       //获取Destination对象

       Queue queue = (Queue)context.lookup("queue/myqueue");

       //创建文本消息

       TextMessage message = session.createTextMessage("你好，我是韩学敏。这是第一个消息驱动Bean");

       //创建消息发送者,发送到queue 目的对象

       QueueSender sender = session.createSender(queue);

       //发送消息

       sender.send(message);

       //关闭资源

       session.close();

       connection.close();

       System.out.println("消息已经发送!");

    }

}

client发送消息的一般步骤：

(1)创建一个JNDI初始化上下文(Context)

   Properties props = new Properties();

   props.setProperty(“java.naming.factory.initial”,”org.jnp.interfaces.NamingContextFactory”);

   props.setProperty(“java.naming.provider.url”,”localhost:1099”);

   props.setProperty(“java.naming.factory.url.pkgs”,”org.jboss.naming:org.jnp.interfaces”);

   InitialContext ctx = new InitialContext(props);

或，建立一个jndi.properties文件。

java.naming.factory.initial=org.jnp.interfaces.NamingContextFactory

   java.naming.factory.url.pkgs=org.jboss.naming:org.jnp.interfaces

   java.naming.provider.url=localhost

MyQueeuMDBBeanClient.java中写例如以下代码：

InitialContext context = new InitialContext();

(2)依据上下文查找连接工厂TopicConnectFactory/QueueConnectionFactory(有两种连接工厂。依据是topic/queue来使用对应的类型);

//获得QueueConnectionFactory对象

QueueConnectionFactory factory = (QueueConnectionFactory)context.lookup("ConnectionFactory");

(3)从连接工厂得到一个连接(Connect  有两种连接TopicConnection/QueueConnection);

//创建QueueConnection

QueueConnection connection = (QueueConnection)factory.createConnection();

(4)通过连接来建立一个会话(Session);

/**

* 通过connection创建QueueSession对象。

* 当中第一个參数为是否支持事务。TRUE为支持，false为不支持。

* 若设为true,则须要手动COMMIT;

* 第二个參数为响应的模式，普通情况下就设为QueueSession.AUTO_ACKNOWLEDGE

* */

QueueSession session = (QueueSession)connection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);

(注：建立不须要事务的而且能自己主动接收消息收条的会话。在非事务Session中。消息传递的方式有三种：

Session.AUTO_ACKNOWLEGE:当客户机调用的receive方法成功返回，或当MessageListenser成功处理了消息，session将会自己主动接收消息的收条。

Session.CLIENT_ACKNOWLEDGE:Session对象依赖于应用程序对已接收到的消息调用确认方法。一旦调用该方法，会话将确认全部自上次确认后收到的消息。

该方法同意应用程序通过一次调用接收、处理和确认一批消息。

Session.DUPS_OK_ACKNOWLEDGE:一旦消息处理中返回了应用程序接收方法，Session对象即确认消息接收，同意反复确认。就资源利用情况而言，此模式最高效。)

(5)查找目的地(Topic/Queue);

Destination destination =(Queue) ctx.lookup(“queue/foshanshop”);

(6)依据会话以及目的地建立消息制造者MessageProducer(扩展了QueueSender和TopicPublisher这两个基本口)

MessageProducer producer = session.createProducer(destination);

TextMessage msg=session.createTextMessage(“你好，我是韩学敏。第一个消息驱动Bean”);//发送文本

producer.send(msg);

2，MyQueueMDBBean.java(Queue消息的接收者)

package com.jinbo.cn;

import javax.ejb.ActivationConfigProperty;

import javax.ejb.MessageDriven;

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.MessageListener;

import javax.jms.TextMessage;

//通过注解配置 MDB 是一个什么样的消费者. 这里p2p消费者

@MessageDriven(

    activationConfig={

       @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),

       @ActivationConfigProperty(propertyName = "destination", propertyValue = "queue/myqueue")

    }

)

public class MyQueueMDBBean implements MessageListener {

    public void onMessage(Message msg) {

       //转换为 TextMessage

       TextMessage message = (TextMessage)msg;

       try {

           System.out.println("MyQueueMDBBean被调用了:[" + message.getText() +"]");

       } catch (JMSException e) {

           e.printStackTrace();

       }

    }

}

以上代码的凝视解释：

@MessageDriver凝视：指明这是一个消息驱动Bean；

@ActivationConfigProperty凝视：使用此凝视配置消息的各种属性；当中destinationType属性指定消息的类型queue。Destination属性制定消息路径(Destination),消息驱动Bean在公布时，假设路径(Destination)不存在。容器会自己主动创建，当容器关闭时路径将被删除。

执行程序。当一个消息到达queue/muqueue队列，就会触发onMessage方法，消息作为一个參数传入，在onMessage方法里得到消息体并调用print方法把消息内容打印到控制台。

3,JBoss公布。执行结果：

MyQueueMDBBean被调用了:[你好，我是韩学敏。这是第一个消息驱动Bean]

相关文章