ActiveMQ

ActiveMQ

　　 MQ是消息中间件，是一种在分布式系统中应用程序借以传递消息的媒介，常用的有ActiveMQ，RabbitMQ，kafka。ActiveMQ是Apache下的开源项目，完全支持JMS1.1和J2EE1.4规范的JMS Provider实现。 
特点： 
　　1、支持多种语言编写客户端 
　　2、对spring的支持，很容易和spring整合 
　　3、支持多种传输协议：TCP,SSL,NIO,UDP等 
　　4、支持AJAX 

　　消息形式： 
　　1、点对点（queue） 
　　2、一对多（topic） 

安装启动

　 　官网下载压缩包，解压到相应目录 ， 这里解压到/opt

　　启动mq ， ./bin/activemq start

　　关闭mq,   ./bin/activemq stop

　　访问 ：127.0.0.1:8161/admin/     账号admin   密码admin

　　采用61616端口提供JMS服务，采用8161提供管理控制台服务

　　通信协议：TCP 

生产者生产消息

package com.steak.activemq.test;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class mqTest {

　　private static final StringACTIVE_URL ="tcp://127.0.0.1:61616";

    private static final StringQUEUE ="queue";

    public static void main(String[] args)throws JMSException {

//创建连接工厂

        ActiveMQConnectionFactory activeMQConnectionFactory =new ActiveMQConnectionFactory(ACTIVE_URL);

        //通过连接工厂，获得连接

        Connection connection = activeMQConnectionFactory.createConnection();

        connection.start();

        //创建session

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

        //创建目的地

        Queue queue = session.createQueue(QUEUE);

        //创建消息的生产者

        MessageProducer messageProducer = session.createProducer(queue);

        //通过使用messageProducer生产消息发送到MQ队列里

        for (int i =0 ; i <3 ; i++){

//创建消息

            TextMessage textMessage = session.createTextMessage("消息 "+i);

            //通过messageProducer发送消息

            messageProducer.send(textMessage);

        }

//关闭资源

        messageProducer.close();

        session.close();

        connection.close();

        System.out.println("消息发送完成");

    }

}

　　启动消费者发送消息，此时等待消费的消息为3条，消费者为0个，进队的消息为3条，出队消息为0条

　  　Number Of Pending Messages:等待消费的消息

　　Number of Conumers : 消费者数量

　　Message Enqueued ： 进队消息数

　　Message Dequeued :    出队消息数

消费者消费消息

package com.steak.activemq.test;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class Consumer {

private static final StringACTIVE_URL ="tcp://127.0.0.1:61616";

    private static final StringQUEUE ="queue";

    public static void main(String[] args)throws JMSException {

//创建连接工厂

        ActiveMQConnectionFactory activeMQConnectionFactory =new ActiveMQConnectionFactory(ACTIVE_URL);

        //通过连接工厂，获得连接

        Connection connection = activeMQConnectionFactory.createConnection();

        connection.start();

        //创建session

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

        //创建目的地

        Queue queue = session.createQueue(QUEUE);

        //创建消费者

        MessageConsumer messageConsumer = session.createConsumer(queue);

        //通过同步阻塞接受消息

        while (true){

//接受消息的类型要和发送消息的一样 ,

            TextMessage textMessage = (TextMessage) messageConsumer.receive();

            if (null != textMessage){

System.out.println("消息 "+textMessage.getText());

            }else {

break;

            }

}

messageConsumer.close();

        session.close();

        connection.close();

    }

}

　　此时消费者有一个，并且消费了3条消息

　　消费者接收消息时可以一直等待（耗费系统资源），也可以设置时间，

　　当为receive()(同步阻塞)时,消费者一直阻塞等待消费消息

　　　　为recevice(long varl)(异步阻塞)时，当超时后，消费者就消失

　　　　通过异步监听方式来接收消息

//通过异步监听方式来接收消息

    messageConsumer.setMessageListener(new MessageListener() {

@Override

        public void onMessage(Message message) {

if (null != message && messageinstanceof TextMessage){

TextMessage textMessage = (TextMessage) message;

                try {

System.out.println("消息  "+((TextMessage) message).getText());

                }catch (JMSException e) {

e.printStackTrace();

                }

}

}

});

    System.in.read(); //保证控制台不关闭，作用是消费完消息以后才能关闭

    messageConsumer.close();

    session.close();

    connection.close();

}

　　　　

消费者消费情况

         1，先启动生产者，再启动多个消费者，一定是第一个消费者消费完所有消息，后面的都消费不到

         2，先启动多个消费者，在启动生产者，消息基本是平均消费的，消息个数是基数的时候也是一个的差异

 

主体模式

　　当为（主题模式）Topic时，所有消费者收到的消息都是一样的,前提是要先订阅，订阅后才能收到消息，消息是无状态的，发送消息后就什么都不管了

　　主题模式

 

 　　如：两人订阅

 　　生产者发布消息，此时发布了三条消息，但是为主题模式，有两个消费者，所以一共消费6条