码迷,mamicode.com
首页 > 其他好文 > 详细

JMS消息服务器(二)——点对点消息传送模型

时间:2015-05-17 23:40:36      阅读:272      评论:0      收藏:0      [点我收藏+]

标签:jms   服务器   消息   发送   点对点模型   

一、点对点模型概览

当你只需要将消息发布送给唯一的一个消息消费者是,就应该使用点对点模型。虽然可能或有多个消费者在队列中侦听统一消息,但是,只有一个且仅有一个消费者线程会接受到该消息

在p2p模型中,生产者称为发送者,而消费者则称为接受者。点对点模型最重要的特性如下:

  • 消息通过称为队列的一个虚拟通道来进行交换。队列是生产者发送消息的目的地和接受者消费消息的消息源。
  • 每条消息通仅会传送给一个接受者。可能会有多个接受者在一个队列中侦听,但是每个队列中的消息只能被队列中的一个接受者消费。
  • 消息存在先后顺序。一个队列会按照消息服务器将消息放入队列中的顺序,把它们传送给消费者当消息已被消费时,就会从队列头部将它们删除。
  • 生产者和消费者之间没有耦合。接受者和发送者可以在运行时动态添加,这使得系统的复杂性可以随着时间而增长或降低(这是消息传送系统的普遍特性)。

点对点消息传送模型有两种类型:异步即发即弃(fire-and-forget)处理异步请求/应答处理。使用即发即弃处理时,消息生产者向某个队列发送一条消息,而且它并不会期望接受到一个响应(至少不是立刻接收到响应)。这类处理可用于触发一个事件,或者用于向接受者发出请求来执行一个并不需要响应的特定活动。异步即发即弃处理如图4-1所示:
技术分享

>
使用异步请求/应答处理时,消息生产者向队里发送一条消息,然后阻塞等待(blocking wait)应答队列,该应答队列正在等待来自接受者的响应。请求/应答处理实现了生产者和消费者之间的高度去耦,允许消息生产者和消费者组件采用不同的语言或平台。异步请求/应答处理如下图所示:
技术分享

用于连接、创建、发送和接受的特定p2p接口见表:

公共API 点对点模型API
ConnectionFactory QueueConnectionFactory
Destination Queue
Connection QueueConnection
Session QueueSession
MessageConsumer QueueSender
MessageProducer QueueReceiver

1.1 何时使用点对点消息传送模型

JMS的初衷是要提供一种公共API的方法,用于访问现有的消息传送系统。在提出JMS规范概念的时候,一些消息传送系统厂商使用的是P2P模型,而另一些厂商使用的则是发布/订阅模型。

当你想让接受者对某个指定的消息进行一次而且仅仅一次处理时,就必须使用点对点模型。这可能是这两种模型之间的最重要的区别:点对点模型只会保证只有一个消费者来处理一条指定的消息。在消息要移除分别接受处理时,要在多个JMS客户端之间均衡消息处理的负载,这是极为重要的。点对点模型的另一优点就是,它所提供的QueueBrowser允许JMS客户端对队列进行快照(Snapshot),以查看正在等待被消费的消息。发布/订阅模型则没有这种浏览特性。

点对点消息传送模型的另一个用例是:您需要在组件之间进行同步通信,而那些组件却是用不同的编程语言编写的,或者是在不同的技术平台(如J2EE或.NET)上实现的。

使用点对点消息传送模型的另一个充分理由是:使用基于消息的负载均衡,可以让服务端的组件实现更大的吞吐量,特别是对于同构组件来说更是如此。

1.2 QBorrower和Qlender应用程序

为了说明点对点消息传送模型是如何工作的,我们将使用一个简单去耦的请求/应答用例。其中,QBorrower类使用点对点消息传送,向QLender类发出了一个简单的抵押贷款申请。QBorrower类使用LoanRequest队列,向QLender类发送贷款申请,而且根据特定的业务规则,QLender类使用LoanResponseQ队列向QBorrower类发回一个响应,表明该LoanRequest是被批准还是拒绝。由于QBorrower感兴趣的是要马上弄清楚贷款批准与否,一旦LoanRequest被发送出去,QBorrower类就会阻塞,并一直等待来自QLender类的响应,无响应就不再继续进行工作。

1.2.1 配置并运行应用程序

  1. 安装下载ActiveMQ并运行,如下图所示
    技术分享
  2. 配置发送接收队列:进入activeMQ的conf目录中,打开activemq.xml文件,在其中增加如下代码:
      <destinations>
            <queue name="LoanRequestQ" physicalName="jms.LoanRequestQ"/>
                <queue name="LoanResponseQ" physicalName="jms.LoanResponseQ"/>
            </destinations>

重新启动MQ,启动成功后打开浏览器输入网址:http://127.0.0.1:8161/,选择Queues可看到我们刚才加进来的队列,如下图
技术分享

3.在程序中生成一个配置文件jndi.properties,内容如下:
技术分享

4.我们再来看一张我们的程序运行示意图:
技术分享

1.2.2 QBorrower类

QBorrowerl 类负责向包含工资额和贷款额的一个队列发送LoanRequest消息。这个类非常简单:构造函数建立一个到JMS提供者的连接,创建一个QueueSession,并使用JNDI查找获得请求和响应队列。

package cn.com.paner.jms.p2p;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.nio.Buffer;
import java.util.StringTokenizer;
import java.util.UUID;

import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueReceiver;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

public class QBorrower {


    private QueueConnection qConnect = null;
    private QueueSession qSession = null;
    private Queue responseQ  =  null;
    private Queue requestQ  = null;

    public QBorrower(String ququecf,String requestQueue,
            String responseQueue)
    {


        try {
            //连接提供者并获取JMS连接
            Context ctxContext = new InitialContext();
                    QueueConnectionFactory qFactory = (QueueConnectionFactory)
            ctxContext.lookup(ququecf);
            qConnect = qFactory.createQueueConnection();

            //创建JMS会话
            qSession = qConnect.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);

            //查找请求队列和响应队列

            requestQ = (Queue)ctxContext.lookup(requestQueue);
            responseQ = (Queue)ctxContext.lookup(responseQueue);

            //现在完成创建,启动连接
            qConnect.start();

        } catch (NamingException | JMSException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

    }

    private void sendLoanRequest(double salary,double loanAmt){


        try {
            //创建JMS消息‘
            MapMessage msg  = qSession.createMapMessage();
            msg.setDouble("Salary", salary);
            msg.setDouble("LoanAmount", loanAmt);
            msg.setJMSReplyTo(responseQ);
            UUID uuid = UUID.randomUUID();
            msg.setStringProperty("UUID", uuid.toString());

            //创建发送者并发送消息
            QueueSender qSender = qSession.createSender(requestQ);
            qSender.send(msg);

            //等待查看贷款申请被接受或拒绝
            //String filter  ="JMSCorrelationID = ‘"+ msg.getJMSMessageID()+"‘";
            String filter  ="JMSCorrelationID=‘"+ uuid.toString()+"‘";
            System.out.println(filter);
            //String slecector = "CustomerType = ‘GOLD‘ OR JMSPriority BETWEEN 5 AND 9";
            QueueReceiver qReceiver = qSession.createReceiver(responseQ,filter);
            TextMessage tmsg = (TextMessage) qReceiver.receive(3000);
            System.out.println(tmsg);
            if (tmsg == null) {
                System.out.println("QLender not responding.");
            }else {
                System.out.println("Loan request was "+tmsg.getText());
            }

        } catch (JMSException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    private void exit()
    {
        try {
            qConnect.close();
        } catch (JMSException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        System.exit(0);
    }

    public static void main(String[] args) {
        // TODO Auto-generated method stub

        String queuecf = null;
        String requestq = null;
        String responseq = null;

        if (args.length == 3) {
          queuecf = args[0];
          requestq = args[1];
          responseq = args[2];
        }else {
            System.out.println("Usage:queueFactoy requestQueue responseQueue.");
            System.exit(0);
        }

        QBorrower borrower = new QBorrower(queuecf, requestq, responseq);
        try {

            BufferedReader stdin = new BufferedReader
                    (new InputStreamReader(System.in));
            System.out.println("QBorrower Application Started");
            System.out.println("Press enter to quit application");
            System.out.println("Enter : Salary,Loan_Amount");
            System.out.println("\neg,g, 5000 , 12000");

            while (true) {

                System.out.println(">");
                String loanRequest = stdin.readLine();
                if (loanRequest == null || 
                        loanRequest.trim().length() <=0) {
                    borrower.exit();
                }

                //解析交易说明】
                StringTokenizer st = new StringTokenizer(loanRequest, ",");
                double salary = Double.valueOf(st.nextToken().trim()).doubleValue();
                double loanAmt = Double.valueOf(st.nextToken().trim()).doubleValue();

                borrower.sendLoanRequest(salary, loanAmt);

            }
        } catch (Exception e) {
            // TODO: handle exception
        }

    }

}

QBorrower类的面方法从命令行接收3个参数:队列连接工厂的JNDI名称、贷款申请队列的JNDI名称,最后是贷款响应队列的JNDI名称,这个响应队列将接收来自QLender类的响应。
java -jar QBorrowe.jar QueueCF LoanRequestQ LoanResponseQ
java -jar QLender.jar QueueCF LoanResponseQ

JMS初始化

在QBorrower类中,所有的JMS初始化逻辑都在构造函数中处理。构造函数要做到第一件事就是:通过创建一个InitalContext,建立一个到JMS提供者的连接:

//连接提供者并获取JMS连接
            Context ctxContext = new InitialContext();
                    QueueConnectionFactory qFactory = (QueueConnectionFactory)
            ctxContext.lookup(ququecf);
            qConnect = qFactory.createQueueConnection();

当创建QueueConnection时,该连接最初是处于停止模式的。这就意味着虽然你可以将消息发送给队列,但是没有消息消费者能够从这个连接接受到消息,直到它被启动为止。

QueueConnection对象用于创建一个JMS Session对象,该对象时JMS中的一个工作线程和事务性工作单元。

通常来说,应用程序会在应用程序启动时创建一个单独的JMS connection,并维护一个Session对象池,供生产或消费者使用。

QueuSession对象通过QueueConnection对象上的工厂对象来创建。关闭Connection很重要,关闭Connection对象也将关闭所有打开的、和该连接有关的Session对象。创建QueueSession的语句如下:
//创建JMS会话
qSession = qConnect.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);

请注意:createQueueSession方法使用两个参数,第一个参数表示QueueSession是否为事务性的。true表示是事务性的,这意味着,在QueueSession预期试用期内发送到队列的消息,将不会传送给接受者,直到QueueSession上调用commit方法为止。同样,在QueueSession上调用rollback方法,会删除事务性回话期间发送的所有消息。第二个参数表示确认模式。

最后一行代码启动连接,自此允许在该连接上接受消息。通常来说,在启动该连接之前执行所有的初始化逻辑,是一个明智之举。

发送消息和接受消息

JMS消息时通过和消息类型相匹配的一个工厂方法,是从Session对象中创建的。使用new来时实例化一个新的JMS消息将不会奏效;他必须从Session对象中创建。在创建并加载消息对象之后,我们还为响应队列设置了JMSReplyTo消息头属性,这会进一步解决生产者和消费者之间的耦合。使用请求/应答模型时,在消息生产者中设置JMSReplyTo消息头属性,而不是在消息消费者中指定应答队列,这是一种同行的标准做法。

//创建JMS消息‘
            MapMessage msg  = qSession.createMapMessage();
            msg.setDouble("Salary", salary);
            msg.setDouble("LoanAmount", loanAmt);
            msg.setJMSReplyTo(responseQ);
            UUID uuid = UUID.randomUUID();
            msg.setStringProperty("UUID", uuid.toString());

在创建消息之后,接下来我们将创建QueuSender对象,指定希望发送消息的队列,然后在使用send方法消息;

//创建发送者并发送消息
            QueueSender qSender = qSession .createSender(requestQ);
                   qSender.send(msg );

在QueueSender对象中,有若干种可用的重写send方法。

一旦消息已被发送出去,QBorrower类就会被阻塞,并等待QLender关于贷款被批准或拒绝的响应。这个过程的第一步就是去创建一个消息选择器,以便我们能够将响应消息和发送的消息关联起来。这是很有必要的,因为申请贷款时,可能同时还有许多其他的贷款申请正被发送到贷款申请队列,或者从中出发。为了确保能够得到准确的响应消息,我们使用一种消息关联的技术。

在创建QueueReceiver时,我们会指定过滤器,表明只有在JMSCorrelationID和原始的JMSMessageID相等时才会接受消息。由于有QueueReceiver,我们能够调用receive方法进行阻塞等待,直到响应消息被接受为止。

//等待查看贷款申请被接受或拒绝
            //String filter  ="JMSCorrelationID = ‘"+ msg.getJMSMessageID()+"‘";
            String filter  ="JMSCorrelationID=‘"+ uuid.toString()+"‘";
            System.out.println(filter);
            //String slecector = "CustomerType = ‘GOLD‘ OR JMSPriority BETWEEN 5 AND 9";
            QueueReceiver qReceiver = qSession.createReceiver(responseQ,filter);
            TextMessage tmsg = (TextMessage) qReceiver.receive(3000);
            System.out.println(tmsg);
            if (tmsg == null) {
                System.out.println("QLender not responding.");
            }else {
                System.out.println("Loan request was "+tmsg.getText());
            }

始终未receive方法指定一个合理的延时值,这肯定是一个明智之举;否则,它将在那里一直等待。

1.2.3 QLender

QLender类的作用是去侦听贷款申请队列上的贷款申请,判断工资是否满足必要的商业要求,并最终将结果发回给借方。

package cn.com.paner.jms.p2p;

import java.io.BufferedReader;
import java.io.InputStreamReader;

import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueReceiver;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;

public class QLender implements MessageListener {

    private QueueConnection qConnect = null;
    private QueueSession qSession = null;
    private Queue  requestQ = null;


    public QLender(String Queuecf,String requetQueue){

        try {
            //连接到提供者并获得JMS连接
            Context ctxContext = new InitialContext();
            QueueConnectionFactory qFactory = (QueueConnectionFactory) ctxContext.lookup(Queuecf);
            qConnect = qFactory.createQueueConnection();

            //创建JMS会话
            qSession = qConnect.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

            //查找申请队列
            requestQ = (Queue) ctxContext.lookup(requetQueue);

            //启动连接
            qConnect.start();

            //创建消息侦听器
            QueueReceiver qReceiver = qSession.createReceiver(requestQ);
            qReceiver.setMessageListener(this);

            System.out.println("Waitting for loan request ...");

        } catch (Exception e) {
            // TODO: handle exception
            e.printStackTrace();
            System.exit(0);
        }
    }

    @Override
    public void onMessage(Message arg0) {
        // TODO Auto-generated method stub

        try {
            boolean accepted = false;

            if (arg0 instanceof MapMessage) {
                System.out.println("message type is legal.");
            }

            //从消息中获取数据
            MapMessage msg = (MapMessage)arg0;
            double salary = msg.getDouble("Salary");
            double loanAmt = msg.getDouble("LoanAmount");

            //决定是否接受或拒绝贷款申请
            if (loanAmt < 200000) {
                accepted = (salary / loanAmt) > 0.25;
            }else {
                accepted = (salary / loanAmt) > 0.33;
            }

            System.out.println("% = "+(salary / loanAmt)+"loan is ?"
                    +(accepted? "Accepted" : "Declined"));

            //将结果返回
            TextMessage tmsg = qSession.createTextMessage();
            tmsg.setText(accepted? "Accepted" : "Declined");
            //tmsg.setJMSCorrelationID(arg0.getJMSMessageID());
        //  System.out.println("JMSCorrelationID = "+arg0.getStringProperty("UUID"));
            tmsg.setJMSCorrelationID(arg0.getStringProperty("UUID"));

            //创建发送者并发送消息
            QueueSender qSender =
                    qSession.createSender((Queue)arg0.getJMSReplyTo());
            qSender.setTimeToLive(30*60*1000);
            qSender.send(tmsg);

            System.out.println("\nWaiting for loan requests...");

        } catch (Exception e) {
            // TODO: handle exception
            e.printStackTrace();
            System.exit(0);
        }

    }

    private void exit()
    {
        try {
            qConnect.close();
        } catch (JMSException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        System.exit(0);
    }

    public static void main(String argv[])
    {
        String queuecf  = null;
        String requestq = null;

        if (argv.length == 2) {
            queuecf = argv[0];
            requestq = argv[1];
        }else {
            System.out.println("Invalid arguments,Should be:");
            System.out.println("java OLender factory request_queue");
            System.exit(0);;
        }

        QLender lender = new QLender(queuecf, requestq);
        try {
            //
            BufferedReader stdin = new BufferedReader(
                    new InputStreamReader(System.in));

            System.out.println("OLender application stared");
            System.out.println("Press enter to quit application.");
            stdin.readLine();
            lender.exit();

        } catch (Exception e) {
            // TODO: handle exception
            e.printStackTrace();
        }
    }

}

QLender类之所以称为一个异步消息侦听器,意味着它和前面的QBorrower类不同,他在等待消息时,不会阻塞。从QLender类实现MessageListener接口和重写OnMessage方法的事实来看,这一点是显而易见的。

一旦启动连接,QLender类就可以开始接受消息。不过在它能够接受消息之前,必须由QueueReceiver注册一个消息侦听器:

    //创建消息侦听器
            QueueReceiver qReceiver = qSession.createReceiver(requestQ);
            qReceiver.setMessageListener(this);

至此,已经启动了一个单独的侦听线程。该线程将一直等待,直到接受到一个消息为止,而且一旦它接受到一条消息,就会调用侦听器类的onMessage方法。我们可以很容易地将消息传送工作委托给实现了MessageListener接口的另一个类:

qReceiver.setMessageListener(otherclass);

在createReceiver方法指定的队列中接受一条消息时,侦听器线程将异步调用侦听器类的OnMessage方法。OnMessage方法首先将消息造型成一个MapMessage。
为了使它更安全,最好是在另一种类型的消息正被发送到该队列的情况下,再使用关键字instanceof检查一下JMS的消息类型:

 if (arg0 instanceof MapMessage) {
                        System. out.println("message type is legal." );
                  }

一旦贷款申请已被分析并作出决定,QLenderl类就须向借方发回响应。为了完成这个工作,它首先创建一条要发送的JMS消息。响应消息无须和QLender接收的贷款申请消息时相同类型的JMS消息。

1.3 动态队列对受管队列

动态队列是通过使用厂商特定API的应用程序源代码创建的队列。受管队列则是在JMS提供者配置文件或管理工具中定义的队列。

动态队列的生成和配置往往取决于特定的厂商。一个队列可以由一个消费者专用,也可以被多个消费者共享。根据内存共享和溢出到磁盘选项的不同,它可能还会有容量大小的限制。

JMS不会试图为一个队列的所有可能选项定义一组API,而是用厂商特定 的管理方式来管理地设置这些选项,这样应该是可能的。为了运行时管理队列,大多数厂商都会提供命令行管理工具、图形界面管理工具或API。

JMS提供了一个QueueSession.createQueue(string queueName)方法,打这个方法并不是要在消息传送系统中定义一个新的队列。它的设计目的使用用于返回代表某个现有队列的Queue对象。另外还有一个JMS定义的方法,即QueueSession.createTemporaryQueue()方法,JMS客户端可以使用这个方法创建一个临时队里,而这个临时队里也只能由该JMS客户单消费。

如果你有很多队列,它们的数量还可能会随时间而增多,那么创建动态队列就大有用处。

JMS消息服务器(二)——点对点消息传送模型

标签:jms   服务器   消息   发送   点对点模型   

原文地址:http://blog.csdn.net/caidaoqq/article/details/45798529

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!