消息队列开发记录笔记-ActiveMQ

1.下载ActiveMQ

去官方网站下载：http://activemq.apache.org/

2.运行ActiveMQ

解压缩apache-activemq-5.5.1-bin.zip，然后双击apache-activemq-5.5.1inactivemq.bat运行ActiveMQ程序。

3.代码： 

　　需要参数：消息队列IP、端口默认61616，用户名，密码，queue。写在配置文件中。

　　消费消息代码，项目启动开启一条线程，线程执行以下代码。一直监听消息队列，有数据就消费。

 //连接消息队列
     ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");   //消息队列的地址
     Connection connection = null;
     Session session = null;
     try {
          //根据用户名和密码创建连接消息队列服务器的连接
          connection = factory.createConnection("1014", "1014");
          connection.start();
           //创建客户端确认的会话
          session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
          //确定目标队列
      Destination dest = new ActiveMQQueue("skyform.queue.1014");
          MessageConsumer consumer = session.createConsumer(dest);
          //接受处理消息
           while(true){
              Message msg = consumer.receive();
              if (msg instanceof TextMessage) {
                   String body = null;
                      try {
                        body = ((TextMessage) msg).getText();
                            //TODO:接受到消息以后的业务处理，业务需求，接收消息后，放到自己的队列里调用下面的sender.供别人消费。
                            msg.acknowledge();//事物处理
                      } catch (JMSException e) {
                            e.printStackTrace();
                      }
               }else{
                        Throw new RuntimeException(“非法的消息”);
                 }
               }
      } catch (JMSException e) {
          e.printStackTrace();
      } finally {
         if (session != null) {
             Try{session.close();}catch(Exception e){e. printStackTrace()}
         } 
        if (connection != null) {
           Try{ connection.close();}catch(Exception e){e. printStackTrace()}
        }
     }

发送代码：把消息放到自己的消息队列，供别人消费

　

  // ConnectionFactory ：连接工厂，JMS 用它创建连接
            ConnectionFactory connectionFactory;
            // Connection ：JMS 客户端到JMS Provider 的连接
            Connection connection = null;
            // Session： 一个发送或接收消息的线程
            Session session;
            // Destination ：消息的目的地;消息发送给谁.
            Destination destination;
            // MessageProducer：消息发送者
            MessageProducer producer;
            // TextMessage message;
            // 构造ConnectionFactory实例对象，此处采用ActiveMq的实现jar
          
            try {
                
                
                  connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
                
                // 构造从工厂得到连接对象
                connection = connectionFactory.createConnection("1072","1072");
                // 启动
                connection.start();
                // 获取操作连接
                session = connection.createSession(Boolean.TRUE,
                        Session.CLIENT_ACKNOWLEDGE);
                // 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置
                destination = session.createQueue("skyform.queue.1072");
                // 得到消息生成者【发送者】
                producer = session.createProducer(destination);
                // 构造消息，此处写死，项目就是参数，或者方法获取
                  sendMessage(session, producer);
                
                
                session.commit();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    if (null != connection)
                        connection.close();
                } catch (Throwable ignore) {
                }
            }
        }

4.队列情况查看

发送队列通过 http://localhost:8161/admin 可监听队列情况以及发送测试数据，如下图：

5.常见异常

开发中常见的2个异常以及解决方法：

javax.jms.IllegalStateException: The Session is closed

网络异常时客户端会报出这个错误

javax.jms.JMSException: Channel was inactive for too long

服务器消息较长时间没有消息发送时，客户端会报这个错误

可以把连接mq的url修改成下面的样子

failover:(tcp://127.0.0.1:61616?wireFormat.maxInactivityDuration=10000)&maxReconnectDelay=10000

failover 失效备援

maxInactivityDuration 允许最大静止（消息服务器无消息）时间

maxReconnectDelay 最大重连间隔

解决方法参考：http://hi.baidu.com/hontlong/blog/item/d475a916ffc8e65df3de32c8.html

 

6.安全配置

1、控制台安全配置，打开conf/jetty.xml文件，找到

    <bean id="securityConstraint" class="org.eclipse.jetty.http.security.Constraint">

        <property name="name" value="BASIC" />

        <property name="roles" value="admin" />

        <property name="authenticate" value="false" />

    </bean>

   将“false”改为“true”即可。用户名和密码存放在conf/jetty-realm.properties文件中。

　　

 2.连接和消费队列的用户名密码配置

apache-activemq-5.5.1-binapache-activemq-5.5.1conf

credentials.properties文件配置用户名、密码

activemq.username=1091
activemq.password=aL^PbQInrzUUY^m9

activemq.xml文件配置：

<simpleAuthenticationPlugin>

<users>

<authenticationUser username="${activemq.username}" password="${activemq.password}" groups="users,admins"/>

</users>

</simpleAuthenticationPlugin>

 安全配置参考网站：http://www.cnblogs.com/CopyPaster/archive/2012/04/27/2473179.html

7.ActiveMQ通过JAAS实现的安全机制

参考博客《转载-ActiveMQ通过JAAS实现的安全机制》