ActiveMQ 安装和使用

时间：2021-04-12 12:37:34 收藏：0 阅读：0

一、ActiveMQ的安装

1、下载ActiveMQ，这里使用的版本为5.5.1

https://cwiki.apache.org/confluence/display/ACTIVEMQ/ActiveMQ+5.5.1+Release

2、启动ActiveMQ

cd D:\software\mq\apache-activemq-5.5.1-bin\apache-activemq-5.5.1\bin

输入activemq.bat (我的电脑是64位)

技术图片

启动后如下图所示：说明启动成功

技术图片

3、打开ActiveMQ

localhost:8161/admin/admin

技术图片

二、ActiveMQ的使用（点对点方式）

1、创建生产者

/**
 * 消息生产者
 */
public class ActiveMQProducer {

    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
    private static final String PASSWORD  = ActiveMQConnection.DEFAULT_PASSWORD;//默认密码
    private static final String BROKERURL = ActiveMQConnection.DEFAULT_BROKER_URL; //默认的连接地址
    private static final int SENDNUM = 10; //发送的消息数量

    public static void main(String[] args) throws JMSException {
        ConnectionFactory connectionFactory; //连接工厂
        Connection connection = null; //连接
        Session session; //会话 接受或者发送消息的线程
        Destination destination; // 消息的目的地
        MessageProducer messageProducer; //消息生产者
        //实例化连接工厂
        connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKERURL);
        try{
            //通过连接工厂获取连接
            connection = connectionFactory.createConnection();
            //启动连接
            connection.start();
            //创建session
            session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
            //创建消息队列
            destination = session.createQueue("FirstQueue1");
            //创建消息生产者
            messageProducer = session.createProducer(destination);
            //发送消息
            sendMessage(session, messageProducer);
            session.commit();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if(connection != null){
                connection.close();
            }
        }

    }

    /**
     * 发送消息
     * @param session
     * @param messageProducer
     */
    public static  void sendMessage(Session session, MessageProducer messageProducer) throws JMSException {
        for(int i = 0; i <ActiveMQProducer.SENDNUM; i++){
            String msg = "ActiveMQ 发送的消息"+ i;
            TextMessage message = session.createTextMessage(msg);
            System.out.println("发送消息： " + msg );
            messageProducer.send(message);
        }
    }

}

2、创建消费者

/**
 * 消息消费者
 */
public class ActiveMQConsumer {

    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
    private static final String PASSWORD  = ActiveMQConnection.DEFAULT_PASSWORD;//默认密码
    private static final String BROKERURL = ActiveMQConnection.DEFAULT_BROKER_URL; //默认的连接地址

    public static void main(String[] args) throws JMSException {
        ConnectionFactory connectionFactory; //连接工厂
        Connection connection = null; //连接
        Session session; //会话 接受或者发送消息的线程
        Destination destination; // 消息的目的地
        MessageConsumer messageConsumer; //消息消费者
        //实例化连接工厂
        connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKERURL);
        try{
            //通过连接工厂获取连接
            connection = connectionFactory.createConnection();
            //启动连接
            connection.start();
            //创建session
            session = connection.createSession(Boolean.FALSE,Session.AUTO_ACKNOWLEDGE);
            //创建消息队列
            destination = session.createQueue("FirstQueue1");
            //创建消息消费者
            messageConsumer= session.createConsumer(destination);
            while (true){
              TextMessage textMessage  = (TextMessage)  messageConsumer.receive(100000); // 100秒接收1次
                if(textMessage != null){
                    System.out.println("接收的消息：" + textMessage.getText());
                }else {
                    break;
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if(connection != null){
                connection.close();
            }
        }

    }



}

运行后，查看ActiveMQ

技术图片

缺点： while(true) 一直监听，使用Listener消费

3、使用Listerer消费

/**
 * 消息消费者
 */
public class ActiveMQConsumer2 {

    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
    private static final String PASSWORD  = ActiveMQConnection.DEFAULT_PASSWORD;//默认密码
    private static final String BROKERURL = ActiveMQConnection.DEFAULT_BROKER_URL; //默认的连接地址

    public static void main(String[] args) throws JMSException, InterruptedException {
        ConnectionFactory connectionFactory; //连接工厂
        Connection connection = null; //连接
        Session session; //会话 接受或者发送消息的线程
        Destination destination; // 消息的目的地
        MessageConsumer messageConsumer; //消息消费者
        //实例化连接工厂
        connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKERURL);
        try{
            //通过连接工厂获取连接
            connection = connectionFactory.createConnection();
            //启动连接
            connection.start();
            //创建session
            session = connection.createSession(Boolean.FALSE,Session.AUTO_ACKNOWLEDGE);
            //创建消息队列
            destination = session.createQueue("FirstQueue1");
            //创建消息消费者
            messageConsumer= session.createConsumer(destination);
            //注册消息监听
            messageConsumer.setMessageListener(new Listener());
        }catch (Exception e){
            e.printStackTrace();
        }

    }

}

三、ActiveMQ发布-订阅模式

一个人发布，多个人订阅

1、生产者

/**
 * 消息生产者-消息发布者
 */
public class ActiveMQProducer {

    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
    private static final String PASSWORD  = ActiveMQConnection.DEFAULT_PASSWORD;//默认密码
    private static final String BROKERURL = ActiveMQConnection.DEFAULT_BROKER_URL; //默认的连接地址
    private static final int SENDNUM = 10; //发送的消息数量

    public static void main(String[] args) throws JMSException {
        ConnectionFactory connectionFactory; //连接工厂
        Connection connection = null; //连接
        Session session; //会话 接受或者发送消息的线程
        Destination destination; // 消息的目的地
        MessageProducer messageProducer; //消息生产者
        //实例化连接工厂
        connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKERURL);
        try{
            //通过连接工厂获取连接
            connection = connectionFactory.createConnection();
            //启动连接
            connection.start();
            //创建session
            session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
            destination = session.createTopic("FirstTopic1");
            //创建消息生产者
            messageProducer = session.createProducer(destination);
            //发送消息
            sendMessage(session, messageProducer);
            session.commit();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if(connection != null){
                connection.close();
            }
        }

    }

    /**
     * 发送消息
     * @param session
     * @param messageProducer
     */
    public static  void sendMessage(Session session, MessageProducer messageProducer) throws JMSException {
        for(int i = 0; i < ActiveMQProducer.SENDNUM; i++){
            String msg = "ActiveMQ 发布的消息"+ i;
            TextMessage message = session.createTextMessage(msg);
            System.out.println("发布消息： " + msg );
            messageProducer.send(message);
        }
    }


}

2、消费者-订阅者1

/**
 * 消息消费者-消息订阅者1
 */
public class ActiveMQConsumer {

    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
    private static final String PASSWORD  = ActiveMQConnection.DEFAULT_PASSWORD;//默认密码
    private static final String BROKERURL = ActiveMQConnection.DEFAULT_BROKER_URL; //默认的连接地址

    public static void main(String[] args) throws JMSException {
        ConnectionFactory connectionFactory; //连接工厂
        Connection connection = null; //连接
        Session session; //会话 接受或者发送消息的线程
        Destination destination; // 消息的目的地
        MessageConsumer messageConsumer; //消息消费者
        //实例化连接工厂
        connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKERURL);
        try{
            //通过连接工厂获取连接
            connection = connectionFactory.createConnection();
            //启动连接
            connection.start();
            //创建session
            session = connection.createSession(Boolean.FALSE,Session.AUTO_ACKNOWLEDGE);

            destination = session.createTopic("FirstTopic1");
            //创建消息消费者
            messageConsumer= session.createConsumer(destination);
            //注册消息监听
            messageConsumer.setMessageListener(new Listener());
        }catch (Exception e){
            e.printStackTrace();
        }

    }


}

Listener

/**
 * 消息监听-订阅1
 */
public class Listener  implements MessageListener {
    @Override
    public void onMessage(Message message) {
        try {
            System.out.println("订阅者1 收到的消息：" + ((TextMessage)message).getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

3、消费者-订阅者2

/**
 * 消息消费者-消息订阅者2
 */
public class ActiveMQConsumer2 {

    private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
    private static final String PASSWORD  = ActiveMQConnection.DEFAULT_PASSWORD;//默认密码
    private static final String BROKERURL = ActiveMQConnection.DEFAULT_BROKER_URL; //默认的连接地址

    public static void main(String[] args) throws JMSException {
        ConnectionFactory connectionFactory; //连接工厂
        Connection connection = null; //连接
        Session session; //会话 接受或者发送消息的线程
        Destination destination; // 消息的目的地
        MessageConsumer messageConsumer; //消息消费者
        //实例化连接工厂
        connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKERURL);
        try{
            //通过连接工厂获取连接
            connection = connectionFactory.createConnection();
            //启动连接
            connection.start();
            //创建session
            session = connection.createSession(Boolean.FALSE,Session.AUTO_ACKNOWLEDGE);

            destination = session.createTopic("FirstTopic1");
            //创建消息消费者
            messageConsumer= session.createConsumer(destination);
            //注册消息监听
            messageConsumer.setMessageListener(new Listener2());
        }catch (Exception e){
            e.printStackTrace();
        }

    }



}

Listener2

/**
 * 消息监听-订阅2
 */
public class Listener2 implements MessageListener {
    @Override
    public void onMessage(Message message) {
        try {
            System.out.println("订阅者2 收到的消息：" + ((TextMessage)message).getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

先运行生产者，再运行订阅者1和订阅者2，可以看到两个订阅者都收到了消息。