ActiveMQ安装与使用

时间 2020-04-03

原文原文链接

ActiveMQ简介：http://www.javashuo.com/article/p-krkwopju-bm.html
SpringBoot整合ActiveMQ：http://www.javashuo.com/article/p-segkuliy-ge.html
MQ理论知识：http://www.javashuo.com/article/p-ypffqfyx-s.html

1.安装

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

百度网盘地址:
连接：https://pan.baidu.com/s/1A55X-vchhsaxaoaIXDUw8g 
提取码：qby6
复制这段内容后打开百度网盘手机App，操做更方便哦

1.1 下载

进入官网，向下滑动到这个位置java

点击下载最新windows版apache

1.2 运行

1.2.1 解压并启动

将下载好的压缩文件，根据本身的需求放到指定目录，而后解压，
解压后到apache-activemq-5.15.12\bin\win64目录 启动activemq.bat 文件

1.2.2 登陆

访问:http://localhost:8161
点击 Manage ActiveMQ broker,点击去下面标识的便可看到

帐号和密码都是admin
http://localhost:8161/admin/

登陆成功,看到以下页面

1.3 小小介绍一下,后面详细介绍

ActiveMQ使用的是标准生产者和消费者模型windows

1. Queue队列(P2P消息模型)，生产者生产了一个消息，只能由一个消费者进行消费。如给微信好友发消息。
2. Topic主题/广播(Pub/Sub消息模型)，生产者生产了一个消息，能够由多个消费者进行消费。如微信公众号给粉丝发消息。

JMS和ActiveMQ的对应微信

JMS消息模型	P2P模式	pub/sub模式
ActiveMQ消息	Queue队列	Topic队列
特色	一对一,一我的发送,只容许一我的接收	一对多,一我的发送, 容许多我的接收

发送的人: 生产者
接收的人: 消费者

2.使用Java操做ActiveMQ

2.1 建立普通maven项目

简单的截几张建立时的图

2.1.1 添加pom依赖

<dependencies>
  <dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-all</artifactId>
    <version>5.15.0</version>
  </dependency>
</dependencies>

若是本身仓库没有依赖, 导入后会本身下载, 稍等便可session

2.1.2 编写生产者

使用JMS原生API编写测试类, 小消息中间件写入消息的开发步骤以下:负载均衡

1. 建立链接工厂
2. 从链接工厂中获取链接
3. 启动链接
4. 获取会话
5. 建立Queue队列
6. 建立生产者
7. 建立消息
8. 发送消息
9. 提交请求
10. 关闭各类资源

2.1.2.1 建立包com.manlu.queue , 建立类ActiveMQProducter

package com.manlu.queue;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;

/**
 * @author 漫路h
 * 生成者发送消息
 */
public class ActiveMQProducter {
    public static void main(String[] args) throws Exception {
        // 链接工厂
        // 使用默认用户名、密码、路径
        // 由于: 底层实现: final String defaultURL = "tcp://"+DEFAULT_BROKER_HOST+ ":" + DEFAULT_BROKER_PORT;
        // 因此: 路径 tcp://host:61616
        // 1. 建立链接工厂
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
        // 2. 建立链接
        Connection connection = connectionFactory.createConnection();
        // 3. 打开链接
        connection.start();
        // 4. 建立会话
        //第一个参数: 是否开启事务
        //第二个参数: 消息是否自动确认
        Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
        //建立队列
        Queue queue = session.createQueue("hello");
        // 5. 建立生产者
        MessageProducer producer = session.createProducer(queue);
        // 6. 建立消息
        TextMessage message = session.createTextMessage("hi i am manlu");
        // 7. 发送消息
        producer.send(message);
        // 8. 关闭资源
        session.commit();
        producer.close();
        session.close();
        connection.close();
        System.out.println("消息生成成功");
    }
}

2.1.2.2 查看控制台,消息发送成功

2.1.2.3 查看消息是否发送成功(访问安装时登陆的网址)

默认tcp链接activeMQ端口 61616 ！！maven

2.1.3 编写消费者

使用JMS原生API编写测试类，向消息中间件消费消息的开发步骤：tcp

1. 建立连接工厂
2. 建立连接
3. 启动连接
4. 获取会话
5. 建立队列
6. 建立消费者
7. 消费消息
8. 提交
9. 关闭资源

2.1.3.1 建立类ActiveMQConsumer,使用MessageConsumer完成消费

package com.manlu.queue;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;

/**
 * @author manlu
 * 使用MessageConsumer完成消费
 */
public class ActiveMQConsumer {
    public static void main(String[] args) throws Exception {
        // 1.建立链接工厂
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
        // 2.建立链接
        Connection connection = connectionFactory.createConnection();
        // 3.开启链接
        connection.start();
        // 4.建立会话
        /*
         *  第一个参数,是否使用事务
         * 若是设置true,操做消息队列后, 必须使用 session.commit();
         * 若是设置false, 操做消息队列后, 不使用 session.commit();
         */
        Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
        // 5.建立队列
        Queue queue = session.createQueue("hello");
        // 6.建立消费者
        MessageConsumer consumer = session.createConsumer(queue);
        while (true){
            //失效时间, 若是10秒内没有收到新的消息,说明没有消息存在,此时能够退出当前循环
            TextMessage message = (TextMessage) consumer.receive(10000);
            if (message!=null) {
                System.out.println(message.getText());
            }else{
                break;
            }
        }
        // 7.关闭链接
        session.commit();
        session.close();
        connection.close();
        System.out.println("消费结束");
    }
}

2.1.3.2 查看控制台,能够发现消息已经被消费

2.1.3.3 查看页面效果(访问安装时登陆的网址)

消费前

消费后

2.1.4 监听器消费消息

package com.manlu.queue;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;

/**
 * @author 漫路
 * 监听器消费消息
 * 重复测试生成和消费的过程。实现一边生成，一边消费的系统
 */
public class ActiveMQMonitorConsumer {
    public static void main(String[] args) throws Exception {
        // 1.建立链接工厂
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
        // 2.建立链接
        Connection connection = connectionFactory.createConnection();
        // 3.开启链接
        connection.start();
        // 4.建立会话
        /*
         *  第一个参数,是否使用事务
         * 若是设置true,操做消息队列后, 必须使用 session.commit();
         * 若是设置false, 操做消息队列后, 不使用 session.commit();
         */
        Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
        // 5.建立队列
        Queue queue = session.createQueue("hello");
        // 6.建立消费者
        MessageConsumer messageConsumer = session.createConsumer(queue);
        messageConsumer.setMessageListener(new MessageListener() {
            //每次接受消息,自动调用 onMessage
            @Override
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage) message;
                try {
                    System.out.println(textMessage.getText());
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
        //咱们不让程序结束,由于若是结束, 监听就结束了
        while (true){
            //目的: 不让程序死掉
        }
    }
}

2.2 多消费模式-queue

P2P消息模型中的多消费者模式,得出结论以下:
	- 一个消息只能被一个消费者消费, 不可重复消费
	- 多个消费者均分消息(负载均衡策略)
	- 当消费者在消费某个消息的时候, mq必定要等到它的成功回执, 才会分发下一个消息

注意: 测试的时候必定要先启动消费者,再启动生产者ide

2.3 Topic

Topic：主题模式、广播模式、pus/sub模式测试

// 1. 建立链接工厂
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
        // 2. 建立链接
        Connection connection = connectionFactory.createConnection();
        // 3. 打开链接
        connection.start();
        // 4. 建立会话
        //第一个参数: 是否开启事务
        //第二个参数: 消息是否自动确认
        Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
	// 5.建立Topic主题模式
	Topic topic = session.createTopic("manlu123");
	// 6.建立生产者
	MessageProducer producer = session.createProducer(topic);

2.4 多消费模式-topic

2.4.1 queue和topic模式的比较

相同点:
  1. 都只有一个生产者
  2. 均可以有多个消费者
不一样点:
  1. queue队列模式, 一个消息只能别一个消费者消费, 不能重复消费
  	当消费者消费某个消息的时候,必定要获得这个消息被成功消费的回执,才会分发下一个消息
	queue入队以后,不管等待多久, 消息都会一直等待消费者来处理
  2. topic广播模式, 一个消息能够被多个消费者消费
  	这个消息不管被成功消费与否,都无所谓
	topic要求时间要一致, 我正好发, 你正好收