ActiveMQ及整合spring boot

维的爪哇 2020-03-10

365

一、 ActiveMQ 消息队列的使用

MQ:MessageQueue 消息队列,就是用来在系统之间进行消息传递的这个队列有一系列具体的实现技术:ActiveMQ、rabbitMQ、kafka、RocketMQ(阿里巴巴)

消息队列在大型电子商务类网站，如京东、淘宝、去哪儿等网站有着深入的应用，队列的主要作用是消除高并发访问高峰，加快网站的响应速度。在不使用消息队列的情况下，用户的请求数据直接写入数据库，在高并发的情况下，会对数据库造成巨大的压力，同时也使得系统响应延迟加剧。

HttpCilent和MQ的比较

1 HttpClient只能算是两个系统间调用的技术,支持跨操作系统跨语言调用

2 MQ可以在两个系统间进行信息交互，并且支持高并发

ActiveMQ支持跨操作系统的java语言中通信
rabbitMQ支持跨操作系统跨编程语言间的通信

1、ActiveMQ简介

什么是ActiveMQ

ActiveMQ工作原理：在本系统中的调用

1、解决服务之间代码耦合

2、使用消息队列，增加系统并发处理量

ActiveMQ应用场景分析：

1、当系统使用短信平台、邮件平台的时候。

用户注册，重点使用用户信息数据库保存，而发短信、发邮件，增加业务处理复杂度，这时候使用MQ，将发短信、发邮箱，通知MQ，由另外服务平台完成。解决了代码的耦合问题。

2、当系统使用搜索平台、缓存平台的时候。

查询数据，建立缓存、索引，当再次查询相同数据的时候，不从数据库查询，从缓存或者索引库查询

当增加、修改、删除数据时，发送消息给MQ，缓存平台、索引平台从MQ获取到这个信息，更新缓存或者索引

总结：使用MQ作为系统间数据调用的中转站。

2、ActiveMQ安装和使用

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

下载完成后

进行apache-activemq-5.14.0\bin\win64目录启动activemq.bat 文件

访问：http://localhost:8161/ 点击：Manage ActiveMQ broker

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

有两种数据结构 Queue、Topic

1、Queue队列(P2P消息模型),生产者生产了一个消息，只能由一个消费者进行消费

2、Topic主题/广播(Pub/Sub消息模型),生产者生产了一个消息,可以由多个消费者进行消费

总结来说就是:queue队列就是微信好友发消息,topic就是微信群发消息

JMS和ActiveMQ的对应

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

2、使用Java程序操作ActiveMQ

2.1、Queue-HelloWorld

2.1.1、pom

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

2.1.2、编写生产者

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

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

第一步：在test/java包中，创建包activeMQ_helloworld，创建类ActiveMQProducter

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("hello1115");
        //5 创建生产者
        MessageProducer producer = session.createProducer(queue);
        //6 创建消息
        Message message = session.createTextMessage("hi i am boy");
        //7 发送消息
        producer.send(message);


        //8 关闭消息
        session.commit();
        producer.close();
        session.close();
        connection.close();
        System.out.println("消息生产成功");
    }
}

第二步：运行代码，在控制台提示：

第三步：查看页面效果，默认tcp连接activeMQ端口 61616

2.1.3、编写消费者

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

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

第一步：使用MessageConsumer完成消费ActiveMQConsumer.java

public class ActiveMQConsumer {
    public static void main(String[] args) throws Exception {
        //创建连接工厂
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
        //创建连接
        Connection connection = connectionFactory.createConnection();
        //开启连接
        connection.start();
        //创建会话
        /** 第一个参数，是否使用事务
         如果设置true，操作消息队列后，必须使用 session.commit();
         如果设置false，操作消息队列后，不使用session.commit();
         */
        Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
        //创建队列
        Queue queue = session.createQueue("hello1115");
        //创建消费者
        MessageConsumer consumer = session.createConsumer(queue);
        while(true){
            //失效时间，如果10秒内没有收到新的消息，说明没有消息存在，此时可以退出当前循环
            TextMessage message = (TextMessage) consumer.receive(10000);
            if(message!=null){
                System.out.println(message.getText());
            }else {
                break;
            }
        }


        //关闭连接
        session.commit();
        session.close();
        connection.close();


        System.out.println("消费结束0");
    }
}

第二步：查看控制台，发现信息已经被消费

第三步：查看页面效果

访问：http://localhost:8161/ 点击：Manage ActiveMQ broke

用户名和密码 都是admin

消费前：表示没有消费

消费后：表示已经消费

2.1.4、监听器消费消息

// 使用监听器消费
public static void main(String[] args) throws Exception {
    // 连接工厂
    // 使用默认用户名、密码、路径
    // 路径 tcp://host:61616
    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
    // 获取一个连接
    Connection connection = connectionFactory.createConnection();
    // 开启连接
    connection.start();
    // 建立会话
    // 第一个参数，是否使用事务，如果设置true，操作消息队列后，必须使用 session.commit();
    Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
    // 创建队列或者话题对象
    Queue queue = session.createQueue("Hello1115");
    // 创建消费者
    MessageConsumer messageConsumer = session.createConsumer(queue);


    messageConsumer.setMessageListener(new MessageListener() {
        // 每次接收消息，自动调用 onMessage
        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一定要等到它的成功回执，才会分发下一个消息

注意：测试的时候一定先启动消费者，然后再启动生产者

Topic-HelloWorld

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

//4 获取会话
// 第一个参数：是否开启事务  true 开启事务，后面一定要提交commit
// 第二个参数：是否自动确认消息已经被消费
Session session = connection.createSession(true,Session.AUTO_ACKNOWLEDGE);
//5 创建Topic主题模式
//Queue queue = session.createQueue("java1.0913");
Topic topic = session.createTopic("java1.0319");
//6 创建生产者
MessageProducer producer = session.createProducer(topic);

2.4、多消费者模式-topic

2.5、queue和topic模式的比较

相同点：

1、都只有一个生产者

2、都可以有多个消费者

不同点：

1、queue队列模式，一个消息只能被一个消费者消费，不能重复消费,当消费者消费某个消息的时候，一定要得到这个消息被成功消费的回执，才会分发下一个消息,queue入队之后，无论等待多久，消息都会一直等待消费者来处理

2、topic广播模式，一个消息可以被多个消费者消费,这个消息无法被成功消费与否，都无所谓,topic要求时间要一致，我正好发，你正好收

3、SpringBoot整合ActiveMQ

在Spring Boot中集成ActiveMQ相对还是比较简单的，都不需要安装什么服务，默认使用内存的activeMQ，当然配合外置ActiveMQ Server会更好。

3.1、创建工程maven子模块

1、在common_parent中导入activeMQ的启动器

<!-- ActiveMQ的启动器 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>

2、创建bos-mq子模块

3、添加项目依赖

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <!-- ActiveMQ的启动器 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-activemq</artifactId>
    </dependency>
</dependencies>

4、创建程序入口

@SpringBootApplication
public class BosMqApplication {


    public static void main(String[] args) {
        SpringApplication.run(BosMqApplication.class, args);
    }
}

5、创建application.properties，具体配置如下：

# MQ所在的服务器的地址
spring.activemq.broker-url=tcp://127.0.0.1:61616
# 是否使用内置的MQ， true  使用；fale  不使用
spring.activemq.in-memory=false 
# 是否在回滚回滚消息之前停止消息传递。这意味着当启用此命令时，消息顺序不会被保留。
spring.activemq.non-blocking-redelivery=false
# 用户名
spring.activemq.password=admin
# 密码
spring.activemq.user=admin

配置的具体意思

spring.activemq.broker-url=tcp://127.0.0.1:61616
# 在考虑结束之前等待的时间
#spring.activemq.close-timeout=15s 
# 默认代理URL是否应该在内存中。如果指定了显式代理，则忽略此值。
spring.activemq.in-memory=true 
# 是否在回滚回滚消息之前停止消息传递。这意味着当启用此命令时，消息顺序不会被保留。
spring.activemq.non-blocking-redelivery=false
# 密码
spring.activemq.password=admin
# 等待消息发送响应的时间。设置为0等待永远。
spring.activemq.user=admin
# 是否信任所有包
#spring.activemq.packages.trust-all=
# 要信任的特定包的逗号分隔列表（当不信任所有包时）
#spring.activemq.packages.trusted=
# 当连接请求和池满时是否阻塞。设置false会抛“JMSException异常”。
#spring.activemq.pool.block-if-full=true
# 如果池仍然满，则在抛出异常前阻塞时间。
#spring.activemq.pool.block-if-full-timeout=-1ms
# 是否在启动时创建连接。可以在启动时用于加热池。
#spring.activemq.pool.create-connection-on-startup=true
# 是否用Pooledconnectionfactory代替普通的ConnectionFactory。
#spring.activemq.pool.enabled=false 
# 连接过期超时。
#spring.activemq.pool.expiry-timeout=0ms
# 连接空闲超时
#spring.activemq.pool.idle-timeout=30s
# 连接池最大连接数
#spring.activemq.pool.max-connections=1
# 每个连接的有效会话的最大数目。
#spring.activemq.pool.maximum-active-session-per-connection=500
# 当有"JMSException"时尝试重新连接
#spring.activemq.pool.reconnect-on-exception=true
# 在空闲连接清除线程之间运行的时间。当为负数时，没有空闲连接驱逐线程运行。
#spring.activemq.pool.time-between-expiration-check=-1ms
# 是否只使用一个MessageProducer
#spring.activemq.pool.use-anonymous-producers=true

6、创建config

package com.czxy.config;
import javax.jms.Queue;
import javax.jms.Topic;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class ActiveMQConfig {
  @Bean
  public Queue queue() {
     return new ActiveMQQueue("Armyman.queue");
  }
  @Bean
  public Topic topic() {
    return new ActiveMQTopic("Armyman.topic");
  }
}

3.2、编写消费者和生产者

1、编写生产者QueueProducer

/**
     * @Author zhangwei
     * @Description //TODO 消息的生产者
     * @Date 17:16 2020/3/9
     * @Param 
     * @return 
     **/
@Componet
@EnableScheduling
public class QueueProducer {  
  /*
   * @Autowired // 也可以注入JmsTemplate，JmsMessagingTemplate对JmsTemplate进行了封装
   * private JmsMessagingTemplate jmsTemplate; //
   * 发送消息，destination是发送到的队列，message是待发送的消息
   * 
   * @Scheduled(fixedDelay=3000)//每3s执行1次 
     public void sendMessage(Destination destination, final String message){
        jmsTemplate.convertAndSend(destination, message); 
     }
   */


    @Autowired
    private JmsMessagingTemplate jmsMessagingTemplate;


    @Autowired
    private Queue queue;


    @Scheduled(fixedDelay=3000)//每3s执行1次
    public void send() {
       try {


       MapMessage mapMessage = new ActiveMQMapMessage();
       mapMessage.setString("info", "你还在睡觉");


       this.jmsMessagingTemplate.convertAndSend(this.queue, mapMessage);


    } catch (Exception e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }
    }
}

2、编写消费者QueueConsumer

@Component  
public class QueueConsumer {  
    //使用JmsListener配置消费者监听的队列，其中Message是接收到的消息  
  @JmsListener(destination = "Armyman.queue")  
    public void receiveQueue(Message message) {
    try {
      MapMessage mapMessage = (MapMessage) message;
      String info = mapMessage.getString("info");
      System.out.println(info);
    } catch (Exception e) {
      e.printStackTrace();
    }
    } 
}

3、启动，测试，OK

3.3 使用内置ActiveMQ

只需要改变properties配置文件，即可运行

# MQ所在的服务器的地址
# spring.activemq.broker-url=tcp://127.0.0.1:61616
# 是否使用SpringBoot内置的MQ， true  使用；fale  不使用
spring.activemq.in-memory=true 
# 是否在回滚回滚消息之前停止消息传递。这意味着当启用此命令时，消息顺序不会被保留。
spring.activemq.non-blocking-redelivery=false
# 用户名
# 密码
#spring.activemq.user=admin
# spring.activemq.password=admin

应用场景：

用在高并发的请求中

注册短信的发送

注册邮件的发送

秒杀

两个系统间进行消息传递

4.1、消息队列应用场景

以下介绍消息队列在实际应用中常用的使用场景。

异步处理，应用解耦，流量削锋和消息通讯四个场景

4.1.1异步处理

场景说明：用户注册后，需要发注册邮件和注册短信。传统的做法有两种:

串行方式；2.并行方式

（1）串行方式：将注册信息写入数据库成功后，发送注册邮件，再发送注册短信。以上三个任务全部完成后，返回给客户端

（2）并行方式：将注册信息写入数据库成功后，发送注册邮件的同时，发送注册短信。以上三个任务完成后，返回给客户端。与串行的差别是，并行的方式可以提高处理的时间

假设三个业务节点每个使用50毫秒钟，不考虑网络等其他开销，则串行方式的时间是150毫秒，并行的时间可能是100毫秒。

因为CPU在单位时间内处理的请求数是一定的，假设CPU在1秒内吞吐量是100次。则串行方式1秒内CPU可处理的请求量是7次（1000/150）。并行方式处理的请求量是10次（1000/100）

小结：如以上案例描述，传统的方式系统的性能（并发量，吞吐量，响应时间）会有瓶颈。如何解决这个问题呢？

引入消息队列，将不是必须的业务逻辑，异步处理。改造后的架构如下：

按照以上约定，用户的响应时间相当于是注册信息写入数据库的时间，也就是50毫秒。注册邮件，发送短信写入消息队列后，直接返回，因此写入消息队列的速度很快，基本可以忽略，因此用户的响应时间可能是50毫秒。因此架构改变后，系统的吞吐量提高到每秒20 QPS。比串行提高了3倍，比并行提高了2倍

4.1.2应用解耦

场景说明：用户下单后，订单系统需要通知库存系统。传统的做法是，订单系统调用库存系统的接口。如下图

传统模式的缺点：

l 假如库存系统无法访问，则订单减库存将失败，从而导致订单失败

l 订单系统与库存系统耦合

如何解决以上问题呢？引入应用消息队列后的方案，如下图：

订单系统：用户下单后，订单系统完成持久化处理，将消息写入消息队列，返回用户订单下单成功

库存系统：订阅下单的消息，采用pub/sub(发布/订阅)的方式，获取下单信息，库存系统根据下单信息，进行库存操作

假如：在下单时库存系统不能正常使用。也不影响正常下单，因为下单后，订单系统写入消息队列就不再关心其他的后续操作了。实现订单系统与库存系统的应用解耦

4.1.3流量削锋

流量削锋也是消息队列中的常用场景，一般在秒杀或团抢活动中使用广泛

应用场景：秒杀活动，一般会因为流量过大，导致流量暴增，应用挂掉。为解决这个问题，一般需要在应用前端加入消息队列。

可以控制活动的人数

可以缓解短时间内高流量压垮应用

用户的请求，服务器接收后，首先写入消息队列。假如消息队列长度超过最大数量，则直接抛弃用户请求或跳转到错误页面

秒杀业务根据消息队列中的请求信息，再做后续处理

4.1.4日志处理

日志处理是指将消息队列用在日志处理中，比如Kafka的应用，解决大量日志传输的问题。架构简化如下

日志采集客户端，负责日志数据采集，定时写受写入Kafka队列

Kafka消息队列，负责日志数据的接收，存储和转发

日志处理应用：订阅并消费kafka队列中的日志数据[外链图片转存失败(img-gzmTdxhA-1567492404919)(assets/wps3257.tmp.jpg)]

(1)Kafka：接收用户日志的消息队列

(2)Logstash：做日志解析，统一成JSON输出给Elasticsearch

(3)Elasticsearch：实时日志分析服务的核心技术，一个schemaless，实时的数据存储服务，通过index组织数据，兼具强大的搜索和统计功能

(4)Kibana：基于Elasticsearch的数据可视化组件，超强的数据可视化能力是众多公司选择ELK stack的重要原因

4.1.5消息通讯

消息通讯是指，消息队列一般都内置了高效的通信机制，因此也可以用在纯的消息通讯。比如实现点对点消息队列，或者聊天室等

点对点通讯：

客户端A和客户端B使用同一队列，进行消息通讯。

聊天室通讯：

4.2、JMS消息服务

消息队列的JAVAEE规范JMS 。JMS（Java Message Service,java消息服务）API是一个消息服务的标准/规范，允许应用程序组件基于JavaEE平台创建、发送、接收和读取消息。它使分布式通信耦合度更低，消息服务更加可靠以及异步性。

4.2.1 消息模型

在JMS标准中,有两种消息模型P2P,Pub/Sub

4.2.2 P2P模式-队列模式

P2P模式包含三个角色：消息队列（Queue），发送者(Sender)，接收者(Receiver)。每个消息都被发送到一个特定的队列，接收者从队列中获取消息。队列保留着消息，直到他们被消费或超时。

P2P的特点

每个消息只能被一个消费者（Consumer）消费(即一旦被消费，消息就不再存在于消息队列中)

发送者和接收者之间在时间上没有依赖性，也就是说当发送者发送了消息之后，不管接收者有没有正在运行，它不会影响到消息被发送到队列

接收者在成功接收消息之后需向队列应答成功

如果希望发送的每个消息都会被成功处理的话，那么需要P2P模式。

4.2.3 Pub/Sub模式–广播/主题模式

包含三个角色主题(Topic),发布者(Publisher)订阅者(Subscriber)多个发布者将消息发送到Topic,系统将这些消息传递给多个订阅者

Pub/Sub的特点

每个消息可以有多个消费者

发布者和订阅者之间有时间上的依赖性。针对某个主题（Topic）的订阅者，它必须创建一个订阅者之后，才能消费发布者的消息

为了消费消息，订阅者必须保持运行的状态

为了缓和这样严格的时间相关性，JMS允许订阅者创建一个可持久化的订阅。这样，即使订阅者没有被激活（运行），它也能接收到发布者的消息。

如果希望发送的消息可以被多个消费者处理的话，那么可以采用Pub/Sub模型。

4.3.消息消费方式

在JMS中，消息的产生和消费都是异步的。对于消费来说，JMS的消息者可以通过两种方式来消费消息。

（1）同步

订阅者或接收者通过receive方法来接收消息，receive方法在接收到消息之前（或超时之前）将一直阻塞；

（2）异步

订阅者或接收者可以注册为一个消息监听器。当消息到达之后，系统自动调用监听器的onMessage方法。

数据库

文章转载自维的爪哇，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。