消息中间件:ActiveMQ

ActiveMQ(Message Queue) 来自apache, 开源的消息总线.

完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现，非常快速

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

 

一, 特性

• 多种语言和协议编写客户端。语言: Java, C, C++, C#, Ruby, Perl, Python, PHP。应用协议: OpenWire,Stomp REST,WS Notification,XMPP,AMQP
• 完全支持JMS1.1和J2EE 1.4规范 (持久化,XA消息,事务)
• 对Spring的支持,ActiveMQ可以很容易内嵌到使用Spring的系统里面去,而且也支持Spring2.0的特性
• 通过了常见J2EE服务器(如 Geronimo,JBoss 4, GlassFish,WebLogic)的测试,其中通过JCA 1.5 resource adaptors的配置,可以让ActiveMQ可以自动的部署到任何兼容J2EE 1.4 商业服务器上
• 支持多种传送协议:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA
• 支持通过JDBC和journal提供高速的消息持久化
• 从设计上保证了高性能的集群,客户端-服务器,点对点
• 支持Ajax
• 支持与Axis的整合
• 可以很容易得调用内嵌JMS provider,进行测试

二, 什么情况使用

• 多项目之间集成 
• 降低系统间模块的耦合度，解耦 
• 系统前后端隔离

三, 启动

   windows: 双击bin目录下activemq.bat脚本

   linux: ./activemq start    ./activemq stop

 

四, 测试

      ActiveMQ默认连接端口是61616,

     可通过查看该端口的信息可以测试ActiveMQ是否成功启动 netstat -an|find "61616"

 

五, 监控

   ActiveMQ默认启动时，启动了内置的jetty服务器，提供一个用于监控ActiveMQ的admin应用。 

 

   admin：http://127.0.0.1:8161/admin/             user/pwd: admin/admin

 

六, 通信方式

  1, publish-subscribe:  发布订阅模式,一对多的关系。

   2, P2P

       在p2p的场景里，相互通信的双方是通过一个类似于队列的方式来进行交流。

       和ub-sub的区别在于一个topic有一个发送者和多个接收者，

         而在p2p里一个queue只有一个发送者和一个接收者。

   3, request-response

      需要双方都能给对方发送消息

  备注: 参考: http://shmilyaw-hotmail-com.iteye.com/blog/1897635

                    https://www.cnblogs.com/zhuxiaojie/p/5564187.html

 

七, 安全配置

activemq/conf/jetty.xml:

   <pre name="code" class="html"> <bean id="securityConstraint" class="org.eclipse.jetty.util.security.Constraint">
        <property name="name" value="BASIC" />
        <property name="roles" value="admin" />
         <!-- 把这个改为true,当然，高版本的已经改为了true -->
        <property name="authenticate" value="true" />
  </bean>

 找到activemq/conf/activemq.xml,并打开

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

 然后账号密码的配置在activemq/conf/credentials.properties文件中

#账号
activemq.username=admin
#密码
activemq.password=123456
guest.password=password

 

八, 示例

流程:

     1. 获得JMS connection factory. 通过我们提供特定环境的连接信息来构造factory。

     2. 利用factory构造JMS connection

     3. 启动connection

     4. 通过connection创建JMS session.

     5. 指定JMS destination.

     6. 创建JMS producer或者创建JMS message并提供destination.

     7. 创建JMS consumer或注册JMS message listener.

     8. 发送和接收JMS message.

     9. 关闭所有JMS资源，包括connection, session, producer, consumer等。

 

public class Producter {

    //ActiveMq 的默认用户名
    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
    //ActiveMq 的默认登录密码
    private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
    //ActiveMQ 的链接地址
    private static final String BROKEN_URL = ActiveMQConnection.DEFAULT_BROKER_URL;

    AtomicInteger count = new AtomicInteger(0);
    //链接工厂
    ConnectionFactory connectionFactory;
    //链接对象
    Connection connection;
    //事务管理
    Session session;
    ThreadLocal<MessageProducer> threadLocal = new ThreadLocal<>();

    public void init(){
        try {
            //创建一个链接工厂
            connectionFactory = new ActiveMQConnectionFactory(USERNAME,PASSWORD,BROKEN_URL);
            //从工厂中创建一个链接
            connection  = connectionFactory.createConnection();
            //开启链接
            connection.start();
            //创建一个事务（这里通过参数可以设置事务的级别）
            session = connection.createSession(true,Session.SESSION_TRANSACTED);
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    public void sendMessage(String disname){
        try {
            //创建一个消息队列
            Queue queue = session.createQueue(disname);
            //消息生产者
            MessageProducer messageProducer = null;
            if(threadLocal.get()!=null){
                messageProducer = threadLocal.get();
            }else{
                messageProducer = session.createProducer(queue);
                threadLocal.set(messageProducer);
            }
           while(true){
                Thread.sleep(1000);
                int num = count.getAndIncrement();
                //创建一条消息
                TextMessage msg = session.createTextMessage(Thread.currentThread().getName()+
                        "productor:我是大帅哥，我现在正在生产东西！,count:"+num);
                System.out.println(Thread.currentThread().getName()+
                        "productor:我是大帅哥，我现在正在生产东西！,count:"+num);
                //发送消息
                messageProducer.send(msg);
                //提交事务
                session.commit();
            }
        } catch (JMSException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

 

public class Comsumer {

    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;

    private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;

    private static final String BROKEN_URL = ActiveMQConnection.DEFAULT_BROKER_URL;

    ConnectionFactory connectionFactory;

    Connection connection;

    Session session;

    ThreadLocal<MessageConsumer> threadLocal = new ThreadLocal<>();
    AtomicInteger count = new AtomicInteger();

    public void init(){
        try {
            connectionFactory = new ActiveMQConnectionFactory(USERNAME,PASSWORD,BROKEN_URL);
            connection  = connectionFactory.createConnection();
            connection.start();
            session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }


    public void getMessage(String disname){
        try {
            Queue queue = session.createQueue(disname);
            MessageConsumer consumer = null;

            if(threadLocal.get()!=null){
                consumer = threadLocal.get();
            }else{
                consumer = session.createConsumer(queue);
                threadLocal.set(consumer);
            }
            while(true){
                Thread.sleep(1000);
                TextMessage msg = (TextMessage) consumer.receive();
                if(msg!=null) {
                    msg.acknowledge();
                    System.out.println(Thread.currentThread().getName()+": Consumer:我是消费者，我正在消费Msg"+msg.getText()+"--->"+count.getAndIncrement());
                }else {
                    break;
                }
            }
        } catch (JMSException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}