基于Tomcat + JNDI + ActiveMQ实现JMS的点对点消息传送

170次阅读

共计 5489 个字符，预计需要花费 14 分钟才能阅读完成。

写了一个简单的 JMS 例子，之所以使用 JNDI 是出于通用性考虑，该例子使用 JMS 规范提供的通用接口，没有使用具体 JMS 提供者的接口，这样可以保证我们编写的程序适用于任何一种 JMS 实现 (ActiveMQ、HornetQ…)。

JNDI(Java Naming and Directory Interface) 是一个标准规范，类似于 JDBC，JMS 等规范，为开发人员提供了查找和访问各种命名和目录服务的通用、统一的接口。J2EE 规范要求所有 J2EE 容器都要提供 JNDI 规范的实现，因此 Tomcat 就实现了 JNDI 规范。

找到 Tomcat 安装路径下的 conf 文件夹，打开 context.xml，添加如下配置：

<Resource name="queue/connectionFactory"    
                auth="Container"    
                type="org.apache.activemq.ActiveMQConnectionFactory"  
                description="JMS Connection Factory"  
                factory="org.apache.activemq.jndi.JNDIReferenceFactory"  
                brokerURL="tcp://localhost:61616"  
                brokerName="LocalActiveMQBroker" />  
                  
<Resource name="queue/queue0"    
                auth="Container"    
                type="org.apache.activemq.command.ActiveMQQueue"  
                description="My Queue"  
                factory="org.apache.activemq.jndi.JNDIReferenceFactory"  
                physicalName="TomcatQueue" />

CMD 到 ActiveMQ 安装路径下的 bin 目录，输入“activemq start”指令即可启动，可在浏览器中输入地址 http://localhost:8161/admin，查看队列、话题等信息。

基于 Tomcat + JNDI + ActiveMQ 实现 JMS 的点对点消息传送

eclipse 上新建 web 工程，添加 ActiveMQ 依赖的 jar 包，然后开始编写两个 Servlet，一个用于生产消息，另一个用于消费消息，如下代码：

消息生产者 Servlet：

import java.io.IOException;
import java.io.PrintWriter;

import javax.jms.DeliveryMode;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.InitialContext;
import javax.servlet.ServletException;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

/**
 * Servlet implementation class JMSTest
 */
@WebServlet("/Send")
public class Send extends HttpServlet {private static final long serialVersionUID = 1L;

    /**
     * @see HttpServlet#HttpServlet()
     */
    public Send() {super();
        // TODO Auto-generated constructor stub
    }

    /**
     * @see HttpServlet#doGet(HttpServletRequest request, HttpServletResponse
     *      response)
     */
    protected void doGet(HttpServletRequest request,
            HttpServletResponse response) throws ServletException, IOException {PrintWriter out = response.getWriter();

        try {// get the initial context
            InitialContext context = new InitialContext();

            // lookup the queue object
            Queue queue = (Queue) context.lookup("java:comp/env/queue/queue0");

            // lookup the queue connection factory
            QueueConnectionFactory conFactory = (QueueConnectionFactory) context
                    .lookup("java:comp/env/queue/connectionFactory");

            // create a queue connection
            QueueConnection queConn = conFactory.createQueueConnection();

            // create a queue session
            QueueSession queSession = queConn.createQueueSession(false,
                    Session.DUPS_OK_ACKNOWLEDGE);

            // create a queue sender
            QueueSender queSender = queSession.createSender(queue);
            queSender.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

            // create a simple message to say "Hello World"
            TextMessage message = queSession.createTextMessage("Hello World");

            // send the message
            queSender.send(message);

            // print what we did
            out.write("Message Sent:" + message.getText());

            // close the queue connection
            queConn.close();} catch (Exception e) {// TODO Auto-generated catch block
            e.printStackTrace();}

    }

    /**
     * @see HttpServlet#doPost(HttpServletRequest request, HttpServletResponse
     *      response)
     */
    protected void doPost(HttpServletRequest request,
            HttpServletResponse response) throws ServletException, IOException {// TODO Auto-generated method stub
    }

}

消息消费者 Servlet：

import java.io.IOException;
import java.io.PrintWriter;

import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueReceiver;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.InitialContext;
import javax.servlet.ServletException;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

/**
 * Servlet implementation class Receive
 */
@WebServlet("/Receive")
public class Receive extends HttpServlet {private static final long serialVersionUID = 1L;

    /**
     * @see HttpServlet#HttpServlet()
     */
    public Receive() {super();
        // TODO Auto-generated constructor stub
    }

    /**
     * @see HttpServlet#doGet(HttpServletRequest request, HttpServletResponse
     *      response)
     */
    protected void doGet(HttpServletRequest request,
            HttpServletResponse response) throws ServletException, IOException {PrintWriter out = response.getWriter();

        try {// get the initial context
            InitialContext context = new InitialContext();

            // lookup the queue object
            Queue queue = (Queue) context.lookup("java:comp/env/queue/queue0");

            // lookup the queue connection factory
            QueueConnectionFactory conFactory = (QueueConnectionFactory) context
                    .lookup("java:comp/env/queue/connectionFactory");

            // create a queue connection
            QueueConnection queConn = conFactory.createQueueConnection();

            // create a queue session
            QueueSession queSession = queConn.createQueueSession(false,
                    Session.AUTO_ACKNOWLEDGE);

            // create a queue receiver
            QueueReceiver queReceiver = queSession.createReceiver(queue);

            // start the connection
            queConn.start();

            // receive a message
            TextMessage message = (TextMessage) queReceiver.receive();

            // print the message
            out.write("Message Received:" + message.getText());

            // close the queue connection
            queConn.close();} catch (Exception e) {// TODO Auto-generated catch block
            e.printStackTrace();}
    }

    /**
     * @see HttpServlet#doPost(HttpServletRequest request, HttpServletResponse
     *      response)
     */
    protected void doPost(HttpServletRequest request,
            HttpServletResponse response) throws ServletException, IOException {// TODO Auto-generated method stub
    }

}