JMS概念及实例

矫情吗；* 2022-04-22 12:08 190阅读 0赞

**1、JMS API 体系结构**

1)、JMS Provider : 实现了JMS API ，提供管理和控制的组件。如JAVA EE platform

2)、JMS Client:  发送和接收消息。

3)、Messages: 消息体。

4)、Administered Object : 管理对象，主要包括 destinations 和 connection factories

如图：

![1356153759_7617.gif][]

**2、JMS 类型(Messaging Domain)：PTP和pub/sub**

1)、PTP：Point to Point  发送者和接收者中有一个消息队列(messages quene),发送者发送消息则把消息加入到队列中，接收者接收消息则把消息从队列中取出；如果接收者没有接收，则这条消息永远保存在队列中(除非已过期)。如下图：

![1356155949_5445.gif][]

特性：

A、每条消息只能有一个接收者

B、发送者和接收者之间可以异步（no timing dependencies）

C、接收者成功接收答复机制

2)、Publish/Subscribe   发送者把消息挂在一个主题下（类似电子公告板），接收者先订阅这个主题，当这个主题有新消息发布时，接收者就可以接收这个主题下的消息了，这个消息一直保持到所有订阅这个消息的人（在线的）都接收了才删除。如图：

![1356157783_6600.gif][]

特性：

A、一条消息可以有多个接收者接收

B、接收者和发送者之间必须同步。

为了弥补这种类型的时间依赖（timing dependencies）劣势,JMS API 提供了创建持久订阅的机制，这样不管接收者是否在线，发送者都可以发送，接收者也可以接收。

**3、JMS API 程序设计模型，包括： **

*  Administered objects: connection factories and destinations
 *  Connections
 *  Sessions
 *  Message producers
 *  Message consumers
 *  Messages

如图：

![1356158106_8932.gif][]

1)、JMS Connection Factories

封装了使用者所有的配置，用来创建Connection 的对象，可以是ConnectionFactory,QueueConnectionFactory, orTopicConnectionFactory的一个实例。可以用JNDI命名空间来管理connection Factories

如：

@Resource(lookup = "jms/ConnectionFactory")
    private static ConnectionFactory connectionFactory;

2)、JMS Destinations

消息发送的目的地和来源地，在PTP类型中，Destination 是quene,而在pub/sub类型中，Destination是topic,一个JMS应用可以有多个quenes或者topics。和connection factory一样也可以用JNDI命名空间来管理Destinations,如：

@Resource(lookup = "jms/Queue")
    private static Queue queue;
    
    @Resource(lookup = "jms/Topic")
    private static Topic topic;

3)、JMS Connections

Connections其实就是一个虚拟的TCP/IP链接，在客户端和JMS提供者之间建立链接，通过它来创建Session。在应用关闭之前，你必须关闭connection,否则会造成资源不会释放。

4)、JMS Session

session是一个单线程的实例，它可以创建以下对象：

*  Message producers
 *  Message consumers
 *  Messages
 *  Queue browsers
 *  Temporary queues and topics

session提供了消息事务管理功能，如：

Session session = connection.createSession(false,
         Session.AUTO_ACKNOWLEDGE);

第一个参数表示非事务性，如果要使创建的消息具有事务性必须用true,第二个参数表示接收消息后会自动答复。

5)、JMS Message Producers

用法：

MessageProducer producer = session.createProducer(dest);
         MessageProducer producer = session.createProducer(queue);
         MessageProducer producer = session.createProducer(topic);

也要以这样用

MessageProducer anon_prod = session.createProducer(null);
         anon_prod.send(dest, message);

在消息发送的时候指定目的地。

6)、JMS Message Consumers

用法：

MessageConsumer consumer = session.createConsumer(dest);
        MessageConsumer consumer = session.createConsumer(queue);
        MessageConsumer consumer = session.createConsumer(topic);

connection.start();
        Message m = consumer.receive();
        connection.start();
        Message m = consumer.receive(1000); // time out after a second

** 4 、 实例:实现了JMS1.1 和J2EE1.4的开源框架:ActiveMq  (持久化用mysql )**

**        ** 第一步：下载activemq 5.7.0.zip，解压放在一个目录下，如D:\\activemq\_5.7.0

第二步：持久化配置：在安装目录\\conf下找到activemq.xml,

加入如下配置：

第二步：启动：打开一个cmd窗口，把当前目录切换到activemq的安装目录：D:\\activemq\_5.7.0

进入bin目录，输入activemq命令回车 OK了（要先配置好java 的环境变量java\_home、path、

classpath）,在浏览器窗口中输入[http://127.0.0.1:8161/admin/][http_127.0.0.1_8161_admin]**  ,**出现activemq的后台管理界

面，表示安装成功！

**        ** 第三步：在eclipse 创建一个web project testActiveMq,同时把activemq-all-5.7.0.jar

(在activemq的安装目录下)导入到项目中

1)、在com.jms.servlet包下创建MyPublish类

package com.jms.servlet;
    
     
    
    import java.io.IOException;
    
    import javax.jms.Connection;
    
    import javax.jms.ConnectionFactory;
    
    import javax.jms.DeliveryMode;
    
    import javax.jms.Destination;
    
    import javax.jms.JMSException;
    
    import javax.jms.Message;
    
    import javax.jms.MessageProducer;
    
    import javax.jms.Session;
    
    import javax.naming.Context;
    
    import javax.naming.InitialContext;
     
    import javax.naming.NamingException;
    
    import javax.servlet.ServletException;
    
    import javax.servlet.http.HttpServlet;
    
    import javax.servlet.http.HttpServletRequest;
    
    import javax.servlet.http.HttpServletResponse;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
     
    
    public class MyPublish extends HttpServlet {
    
     
    
          private static final long serialVersionUID = 8861449351626383534L;
    
          private InitialContext initialContext;
    
          private Context context;
    
          private ConnectionFactory connectionFactory;
    
          private Connection connection;
    
          private Session session;
    
          private Destination destination;
    
          private MessageProducer messageProducer;
    
     
    
          public void init() throws ServletException {
    
                try {
    
    //                  initialContext = new InitialContext();
    //
    //                  context = (Context) initialContext.lookup("java:comp/env");
    //
    //                  connectionFactory = (ConnectionFactory) context
    //
    //                              .lookup("jms/NormalConnectionFactory");
                	  connectionFactory = new ActiveMQConnectionFactory(null, null, "failover:(tcp://localhost:61616)");
                      connection = connectionFactory.createConnection();
                      session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                      destination = session.createTopic("TOOL.DEFAULT");
                      messageProducer = session.createProducer(destination);
                } catch (JMSException e) {
                      e.printStackTrace();
                }
          }
    
     
    
          public void destroy() {
                super.destroy();
          }
    
     
    
          public void doGet(HttpServletRequest request, HttpServletResponse response)
                                                throws ServletException, IOException {
                doPost(request, response);
          }
    
     
    
          public void doPost(HttpServletRequest request, HttpServletResponse response)
                      throws ServletException, IOException {
                String content = request.getParameter("content");
                try {
    
                      // 设置持久方式
                      messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
                      Message testMessage = session.createMessage();
                      // 发布刷新文章消息
                      testMessage.setStringProperty("RefreshArticleId", content);
                      messageProducer.send(testMessage);
                      // 发布刷新帖子消息
                      testMessage.clearProperties();
                      testMessage.setStringProperty("RefreshTopicId", content);
                      messageProducer.send(testMessage);
                } catch (Exception e) {
                      e.printStackTrace();
                }
    
          }
    
    }

2)、在com.jms.servlet包下创建JMSListener类

package com.jms.servlet;
    
    import javax.servlet.*;
    import javax.servlet.http.*;
    import javax.naming.*;
    import javax.jms.*;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class JMSListener extends HttpServlet implements MessageListener {
        private static final long serialVersionUID = 5088494289145588596L;
    
        public void init(ServletConfig config) throws ServletException {
            try {
    //            InitialContext initialContext = new InitialContext();
    //            Context context = (Context) initialContext.lookup("java:comp/env");
    //            ConnectionFactory connectionFactory = (ConnectionFactory) context
    //                    .lookup("jms/FailoverConnectionFactory");
            	 ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(null, null, 
            			 "failover:(tcp://localhost:61616)");
                Connection connection = connectionFactory.createConnection();
                connection.setClientID("MyClient");
                connection.start();
                Session session = connection.createSession(false,
                        Session.AUTO_ACKNOWLEDGE);
                // 普通消息订阅者，无法接收持久消息
                // Destination destination = (Destination) context
                // .lookup("jms/topic/MyTopic");
                // MessageConsumer consumer = session.createConsumer(destination);
                // 基于Topic创建持久的消息订阅者，前提：Connection必须指定一个唯一的clientId，当前为MyClient
               // Topic topic = (Topic) context.lookup("jms/topic/MyTopic");
                Topic topic=session.createTopic("TOOL.DEFAULT");
                TopicSubscriber consumer = session.createDurableSubscriber(topic,"MySub");
                consumer.setMessageListener(this);
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    
       
        public void onMessage(Message message) {
            if (checkText(message, "RefreshArticleId") != null) {
                String articleId = checkText(message, "RefreshArticleId");
                System.out.println("refresh article, ID=" + articleId);
            } else if (checkText(message, "RefreshTopicId") != null) {
                String topicId = checkText(message, "RefreshTopicId");
                System.out.println("refresh topic, ID=" + topicId);
            } else {
                System.out.println("it's normal message, no need to care");
            }
        }
    
        private static String checkText(Message m, String s) {
            try {
                return m.getStringProperty(s);
            } catch (JMSException e) {
                e.printStackTrace(System.out);
                return null;
            }
        }
    }

3)、配置web.xml

在web.xml 增加以下配置

<servlet> <servlet-name>jms-listener</servlet-name> <servlet-class>com.jms.servlet.JMSListener</servlet-class> <load-on-startup>1</load-on-startup> </servlet> <servlet> <servlet-name>MyPublish</servlet-name> <servlet-class>com.jms.servlet.MyPublish</servlet-class> </servlet> <servlet-mapping> <servlet-name>MyPublish</servlet-name> <url-pattern>/myPublish.do</url-pattern> </servlet-mapping>

4)、新增页面index.jsp

<%@ page language="java" import="java.util.*" pageEncoding="utf-8"%>
    <!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN">
    <html>
      <body>
         <form action="myPublish.do">
                <input type="text" name="content" />
                <input type="submit" value="提交">
          </form>
      </body>
    </html>

5)、把该项目发布到tomcat6下启动，在浏览器中输入[http://127.0.0.1:8080/testActiveMq/][http_127.0.0.1_8080_testActiveMq]

输入要发送的消息，如hello,jms 然后在 eclipse 的控制台会看到

refresh article, ID=hello,jms

refresh topic, ID=hello,jms

同时在本地数据库activemq的activemq\_msgs表中会增加两条记录，表示项目测试成功！

[1356153759_7617.gif]: https://img-my.csdn.net/uploads/201212/22/1356153759_7617.gif
[1356155949_5445.gif]: https://img-my.csdn.net/uploads/201212/22/1356155949_5445.gif
[1356157783_6600.gif]: https://img-my.csdn.net/uploads/201212/22/1356157783_6600.gif
[1356158106_8932.gif]: https://img-my.csdn.net/uploads/201212/22/1356158106_8932.gif
[http_127.0.0.1_8161_admin]: http://127.0.0.1:8161/admin/
[http_127.0.0.1_8080_testActiveMq]: http://127.0.0.1:8080/testActiveMq/