jms activeMQ 应用

  一、JMS的理解

JMS概述：

jms即Java消息服务（Java Message Service）应用程序接口是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通 信。Java消息服务是一个与具体平台无关的API，绝大多数MOM提供商都对JMS提供支持。

JMS元素：

JMS有以下元素组成。

　　JMS提供者

　　连接面向消息中间件的，JMS接口的一个实现。提供者可以是Java平台的JMS实现，也可以是非Java平台的面向消息中间件的适配器

JMS客户

　　生产或消费消息的基于Java的应用程序或对象。

JMS生产者

　　创建并发送消息的JMS客户。

JMS消费者

　　接收消息的JMS客户。

JMS消息

　　包括可以在JMS客户之间传递的数据的对象

JMS队列

　　一个容纳那些被发送的等待阅读的消息的区域。队列暗示，这些消息将按照顺序发送。一旦一个消息被阅读，该消息将被从队列中移走。

JMS主题

　　一种支持发送消息给多个订阅者的机制。

JMS模型：

Java消息服务应用程序结构支持两种模型：

　　点对点或队列模型

　　发布者/订阅者模型

　　在点对点或队列模型下，一个生产者向一个特定的队列发布消息，一个消费者从该队列中读取消息。这里，生产者知道消费者的队列，并直接将消息发送到消费者的队列。这种模式被概括为：

　　只有一个消费者将获得消息

　　生产者不需要在接收者消费该消息期间处于运行状态，接收者也同样不需要在消息发送时处于运行状态。

　　每一个成功处理的消息都由接收者签收

　　发布者／订阅者模型支持向一个特定的消息主题发布消息。0或多个订阅者可能对接收来自特定消息主题的消息感兴趣。在这种模型下，发布者和订阅者彼此不知道对方。这种模式好比是匿名公告板。这种模式被概括为：

　　多个消费者可以获得消息

　　在发布者和订阅者之间存在时间依赖性。发布者需要建立一个订阅（subscription），以便客户能够购订阅。订阅者必须保持持续的活动状态以接收消息，除非订阅者建立了持久的订阅。在那种情况下，在订阅者未连接时发布的消息将在订阅者重新连接时重新发布。

　　使用Java语言，JMS提供了将应用与提供数据的传输层相分离的方式。同一组Java类可以通过JNDI中关于提供者的信息，连接不同的JMS提供者。这一组类首先使用一个连接工厂以连接到队列或主题，然后发送或发布消息。在接收端，客户接收或订阅这些消息。

二、        JMS的实现提供者（ActiveMQ）：

1.windows版：

    A．在http://activemq.apache.org/ 官网上下载apache-activemq-5.3.0-bin.zip

    B．下载后解压缩，执行bin文件夹中的activemq.bat 将启动默认配置的activemq

    C．通过http://localhost:8161/admin访问activemq的管理界面。查看队列（queue）和主题（topic）下的消息内容。

2.linux版：

   A．在http://activemq.apache.org/ 官网上下载apache-activemq-5.3.0-bin.tar.gz

B．解压

将下载下来的apache-activemq-5.3.0-bin.tar.gz放置于/usr/local
cd /usr/local/
tar xvfz apache-activemq-5.3.0-bin.tar.gz
则ActiveMQ的安装目录为：/usr/local/apache-activemq-5.3.0

C．修改activemq运行文件的权限
cd /usr/local/apache-activemq-5.2.0/bin
chmod 755 activemq

D．运行
(1).普通启动
   bin/activemq
(2).指定日志文件的启动方式
   bin/activemq >/tmp/smlog 2>&1 &
(3)后台启动方式
   nohup bin/activemq >/tmp/smlog 2>&1 &

这里需要注意：前两种启动方式在linix命令行或者通过ssh客户端启动时在关闭对应的窗口时activemq会关闭，甚至直接ctrl+c也会导致activemq退出；第三种则不会出现这种情况。

E．验证服务端是否开启，有几种方法
(1).查看控制台输出或者日志文件
(2).直接访问activemq的管理页面：http://localhost:8161/admin/

F．关闭activemq
如果开启方式是使用(1)或(2)，则直接ctrl+c或者关闭对应的终端即可
如果开启方式是(3),则稍微麻烦一点：
先查找到activemq对应的进程：
ps -ef | grep activemq
然后把对应的进程杀掉，假设找到的进程编号为 168168
kill 168168

3．ActiveMQ的activemq.xml详细配置讲解

   <beans

xmlns="http://www.springframework.org/schema/beans"

xmlns:amq="http://activemq.org/config/1.0"

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd

http://activemq.org/config/1.0 http://activemq.apache.org/schema/activemq-core.xsd

http://activemq.apache.org/camel/schema/spring>

<!-- persistent="true"表示要持久化存储消息，和子元素persistenceAdapter结合使用 -->

<!-- dataDirectory默认的存储持久化数据的目录 -->

<!-- brokerName 设置broker的name，在注意在网络上必须是唯一的-->

<broker xmlns="http://activemq.org/config/1.0" brokerName="192.168.1.148" persistent ="true" dataDirectory="${activemq.base}/data" useShutdownHook="false">

<!-- Destination specific policies using destination names or wildcards -->

<!-- wildcards意义见http://activemq.apache.org/wildcards.html -->

<destinationPolicy>

<policyMap>

<policyEntries>

<!-- 这里使用了wildcards，表示所有以EUCITA开头的topic -->

<policyEntry topic="EUCITA.>" producerFlowControl="false" memoryLimit="10mb">

<!-- 分发策略 -->

<dispatchPolicy>

<!-- 按顺序分发 -->

<strictOrderDispatchPolicy/>

</dispatchPolicy>

<!--  恢复策略-->

<subscriptionRecoveryPolicy>

<!-- 只恢复最后一个message -->

<lastImageSubscriptionRecoveryPolicy/>

</subscriptionRecoveryPolicy>

</policyEntry>

</policyEntries>

</policyMap>

</destinationPolicy>

<!-- The transport connectors ActiveMQ will listen to -->

<transportConnectors>

<transportConnector name="openwire" uri="tcp://192.168.1.148:61616" discoveryUri="multicast://default"/>

<transportConnector name="ssl"     uri="ssl://192.168.1.148:61617"/>

<transportConnector name="stomp"   uri="stomp://192.168.1.148:61613"/>

<transportConnector name="xmpp"    uri="xmpp://192.168.1.148:61222"/>

</transportConnectors>

<!-- 消息持久化方式 -->

<persistenceAdapter>

<amqPersistenceAdapter directory="${activemq.base}/data"/>

</persistenceAdapter>

</broker>

<!-- lets create a command agent to respond to message based admin commands on the ActiveMQ.Agent topic -->

<commandAgent xmlns="http://activemq.org/config/1.0"/>

<!-- An embedded servlet engine for serving up the Admin console -->

<jetty xmlns="http://mortbay.com/schemas/jetty/1.0">

<connectors>

<nioConnector port="8161" />

</connectors>

<handlers>

<webAppContext contextPath="/admin" resourceBase="${activemq.base}/webapps/admin" logUrlOnStart="true" />

<webAppContext contextPath="/demo" resourceBase="${activemq.base}/webapps/demo" logUrlOnStart="true" />

</handlers>

</jetty>

</beans>

4．ActiveMQ的安全验证配置

ActiveMQ的安全性，主要表现在两个方面，一个是它的JMS服务的安全性，一个是它的管理控制台的安全性。在缺省情况下，ActiveMQ在这两个方面都是没有配置安全性的。

JMS服务安全性

A．如果没有配置JMS服务的安全性，任何连入网络的人，只要知道ActiveMQ服务器的具体地址(包括IP地址，端口，消息地址[队列或者主题地址])，就可以肆无忌惮的发送、接收消息。那么要解决JMS服务的安全性，最关键的问题就是解决身份认证的问题。

1)增加plugin

<plugins>

    <!--use JAAS to authenticate using the login.config file on the classpath to configure JAAS -->

    <jaasAuthenticationPlugin configuration="activemq-domain" />

    <!--  lets configure a destination based authorization mechanism -->

    <authorizationPlugin>

        <map>

            <authorizationMap>

                <authorizationEntries>

                    <authorizationEntry queue=">" read="admins" write="admins" admin="admins" />

                    <authorizationEntry topic=">" read="admins" write="admins" admin="admins" />

                    <authorizationEntry queue="ActiveMQ.Advisory.>" read="admins" write="admins" admin="admins" />

                    <authorizationEntry topic="ActiveMQ.Advisory.>" read="admins" write="admins" admin="admins" />

                </authorizationEntries>

            </authorizationMap>

        </map>

    </authorizationPlugin>

</plugins>

2)在conf中增加login.config文件，内容如下：

    activemq-domain {

        org.apache.activemq.jaas.PropertiesLoginModule required

        debug=true

        org.apache.activemq.jaas.properties.user="users.properties"

        org.apache.activemq.jaas.properties.group="groups.properties";

    };

3)在conf中增加groups.properties文件，内容如下：

   admins=system

4)在conf中增加users.properties文件，内容如下：

   system=manager

管理控制台安全性

  ActiveMQ缺省的管理是通过内置的jetty服务器，只要在浏览器中输入http://localhost:8161/admin，不需要登录，就 可以对队列、主题及消息等进行管理，可以想象这非常不安全。那么要解决管理控制台的安全性，除了通过修改管理端口号以及应用名称之外，最关键的也是需要进 行配置，必须通过身份认证才能登录。本文要说的方式主要是在不改变内置jetty的方式下，通过配置基本认证的方式来实现安全登录。

1)在activemq.xml中增加realm设置

   <userRealms>

       <jaasUserRealm name="adminRealm" loginModuleName="adminLoginModule">

      </jaasUserRealm>

   </userRealms>

2)修改login.config文件，参考上面已见的login.config文件

   adminLoginModule {

       org.mortbay.jetty.plus.jaas.spi.PropertyFileLoginModule required

       debug="true"

       file="C:/software/apache-activemq-5.2.0/conf/realm.properties";

   };

3)在lib\web中增加jetty-plus-6.1.9.jar；

4)在conf中增加realm.properties文件

    system: MD5:1d0258c2440a8d19e716292b231e3190,admins

注意，上面是用户名为system，而密码为manager，如果用户密码不一致，请根据如下命令获得相关信息

java -cp jetty-6.1.9.jar;jetty-util-6.1.9.jar org.mortbay.jetty.security.Password system manager

5)修改webapps/admin/WEB-INF/web.xml

   <security-constraint>

       <web-resource-collection>

           <web-resource-name>adminRealm</web-resource-name>

           <url-pattern>/*</url-pattern>

       </web-resource-collection>

       <auth-constraint>

           <role-name>admins</role-name>

       </auth-constraint>

   </security-constraint>

   <login-config>

       <auth-method>BASIC</auth-method>

       <realm-name>adminRealm</realm-name>

   </login-config>



总结

经过如上配置后，必须通过用户安全验证才能连接上消息服务器并进行消息发送和接收。当访问http://localhost:8161 /admin/时，将会弹出窗口以让用户输入登录帐号和密码。特别说明一下，ActiveMQ跟安全相关的有三个地方，分别是上面说的消息服务器本身的身 份认证、Web Console的身份认证，还有一个地方是通过Web Console访问消息服务器的身份认证，也就是conf\credentials.properties，文件内容如下：

       activemq.username=system

       activemq.password=manager

这三个地方的用户名密码最好一致，尤其是credentials.properties中的账号与users.properties中的管理员账号必须一致。



三、        JMS的应用：

1.   JMS应用程序接口

ConnectionFactory 接口（连接工厂）
　　用户用来创建到JMS提供者的连接的被管对象。JMS客户通过可移植的接口访问连接，这样当下层的实现改变时，代码不需要进行修改。 管理员在JNDI名字空间中配置连接工厂，这样，JMS客户才能够查找到它们。根据消息类型的不同，用户将使用队列连接工厂，或者主题连接工厂。

Connection 接口（连接）
　　连接代表了应用程序和消息服务器之间的通信链路。在获得了连接工厂后，就可以创建一个与JMS提供者的连接。根据不同的连接类型，连接允许用户创建会话，以发送和接收队列和主题到目标。

Destination 接口（目标）
　　目标是一个包装了消息目标标识符的被管对象，消息目标是指消息发布和接收的地点，或者是队列，或者是主题。JMS管理员创建这些对象，然后用户通过JNDI发现它们。和连接工厂一样，管理员可以创建两种类型的目标，点对点模型的队列，以及发布者／订阅者模型的主题。

MessageConsumer 接口（消息消费者）
　　由会话创建的对象，用于接收发送到目标的消息。消费者可以同步地（阻塞模式），或异步（非阻塞）接收队列和主题类型的消息。

MessageProducer 接口（消息生产者）
　　由会话创建的对象，用于发送消息到目标。用户可以创建某个目标的发送者，也可以创建一个通用的发送者，在发送消息时指定目标。

Message 接口（消息）
　　是在消费者和生产者之间传送的对象，也就是说从一个应用程序创送到另一个应用程序。一个消息有三个主要部分：

　　   消息头（必须）：包含用于识别和为消息寻找路由的操作设置。

　   　一组消息属性（可选）：包含额外的属性，支持其他提供者和用户的兼容。可以创建定制的字段和过滤器（消息选择器）。

　   　一个消息体（可选）：允许用户创建五种类型的消息（文本消息，映射消息，字节消息，流消息和对象消息）。

　　消息接口非常灵活，并提供了许多方式来定制消息的内容。

Session 接口（会话）
　　表示一个单线程的上下文，用于发送和接收消息。由于会话是单线程的，所以消息是连续的，就是说消息是按照发送的顺序一个一个接收的。会话的好 处是它支持事务。如果用户选择了事务支持，会话上下文将保存一组消息，直到事务被提交才发送这些消息。在提交事务之前，用户可以使用回滚操作取消这些消 息。一个会话允许用户创建消息生产者来发送消息，创建消息消费者来接收消息。



四、        JMS封装接口的应用：



    (以下是自己封装的接口)


Jms消息管理类：JmsServiceManager.java

1.   public void  sendTextMessage (SmsConfig config1,String content)

是发送 “普通字符串消息，包含一个String”的方法

参数说明：

A．String content  ：要发送的普通字符串消息

B．SmsConfig config1 ：是配置jms发送环境的bean类（常用的参数如下：）

                    1)/*消息的transacted模式*/ 

         private boolean transacted = false;

//jms事务要用到  默认是false  是否启用本地事务

                        2) /**

                            * 传递模式， 有两种模式： PERSISTENT(true) 和

                            * NON_PERSISTENT(false)

                            * 消息的发送模式：persistent或nonpersistent。

                            * 前者表示消息在被消费之前，如果JMS提供者DOWN了，

                            * 重新启动后消息仍然存在。

                            * 后者在这种情况下表示消息会被丢失。

                            * */

                private boolean durable = false;  //默认是false

                       3) /**

                            * 消息池d名字

                            * 发送消息的目的地

                            * */

               private String subject = "TOOL.DEFAULT";  

//默认是TOOL.DEFAULT  

4) /**

                           * 连接url

                           * URL的值为"failover:(tcp://localhost:61616)",

                           * 其中failover关键字表示客户端与ActiveMQ Server的

                           * 连接中断后(如ActiveMQ Server重新启动)，

                           * 客户端会自动重新连接，不需要在我们的代码中处理了。

                          * */

                       private String url = "failover:(tcp://localhost:61616)";       //默认是failover:(tcp://localhost:61616)

5)    /**

                          * 发送消息或接收消息时使用的用户名

                          * */

                          private String pwd = "system";   

//默认是system

   

                      6) /**

                           * 发送消息或接收消息时使用的密码

                           * */

private String user = "manager";  

//默认是manager

7) /**

                           * 点对点传输模式或订阅模式 true 订阅模式，false 点对点模式

                           * */

private boolean topic = false;   //默认是false

8)/**

                         * 表示一个消息的有效期。只有在这个有效期内，

*消息消费者才可以消费这个消息。

                         * 默认值为0，表示消息永不过期。

                         * 可以通过下面的方式设置：有效期1小时 （1000毫秒 * 60秒 * 60分）

                         * */

private long timeToLive = 0; //默认值为0，表示消息永不过期。

9)/**

                         * 表示接收到消息后返回的目的地

                         * */

private String replyTo = ""; 

//默认值为空，表示接收到消息不返回。

2.  public void sendObjectMessage(SmsConfig config1,Object object)

是发送 “对象消息，包含一个可序列化的Java 对象”的方法

参数说明：

Object object : 是继承了com.jms.tool. ActionBean

     SmsConfig config1 : 和 1 方法中的参数是一样的

3.   public void sessionCommit()

//提交事务

4.  public void closeSessionAndConnection()

//关闭会话和连接

5.  public void sessionRollback()

//事务回滚

6.  public void reciveMessage(SmsConfig config1)

是发送 “接受监听信息”的方法

参数说明：

SmsConfig config1 : 和 1 方法中的参数是一样的



Jms消息接受处理类：JmsReciveAction.java

1.  public static void actionObjectMessage(Object object)

是“接受监听信息（对象消息）”的处理方法

参数说明：

Object object : 是接收到的消息对象

2.  public static String actionTextMessage(String content)

是“接受监听信息（String消息）”的处理方法

参数说明：

String content : 是接收到的String消息

并返回处理后的结果。（如果没有返回结果可以返回空字符串””）



     调用Jms接口例子：

         1.发送”String消息”

public static void senderText(){

               JmsServiceManager jmsManager = new JmsServiceManager();

SmsConfig config = new SmsConfig();//jms环境配置bean

               config.setTopic(false); //是 点对点 还是 发布订阅 发送模式

               config.setDurable(true);

               config.setSubject("127.0.0.1"); //消息发送目的地

               config.setTransacted(true);//是否启用本地事务

               config.setUrl("failover:(tcp://localhost:61616)");

               //如果ActiveMQ设置了安全验证 还需要设置用户名和密码

               config.setUser("system");

               config.setPwd("manager");

        try{

             jmsManager.sendTextMessage(config, "我是中国人，我爱我的祖国");                      jmsManager.sessionCommit(); //提交事务

        }catch (Exception e) {

             e.printStackTrace();

             jmsManager.sessionRollback();//事务回滚

        }finally{

             jmsManager.closeSessionAndConnection(); //关闭会话和连接

        }

  }

注：如果config.setTransacted(false);//是否启用本地事务 则不用提交事务



2.接收消息 

          public static void reciverText(){

              JmsServiceManager jmsManager = new JmsServiceManager();

             SmsConfig config = new SmsConfig();

             config.setTopic(false);

             //config.setDurable(true);

           config.setSubject("127.0.0.1"); //消息接收目的地

             config.setUrl("failover:(tcp://localhost:61616)");

              //如果ActiveMQ设置了安全验证 还需要设置用户名和密码

              config.setUser("system");

              config.setPwd("manager");

             jmsManager.reciveMessage(config);

         }



转载自：http://blog.csdn.net/jimmylsp/archive/2010/01/21/5220071.aspx