ActiveMQ入门以及整合spring boot

青旅半醒 2022-04-17 00:16 231阅读 0赞

# 一、 ActiveMQ 消息队列的使用 #

## MQ ##

MQ:Message Queue 消息队列，就是用来在系统之间进行消息传递的

这个队列有一系列具体的实现技术：ActiveMQ、rabbitMQ、kafka、RocketMQ(alibaba)

HttpCilent和MQ的比较

1 HttpClient只能算是两个系统间调用的技术

HttpClient支持跨操作系统跨语言调用

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

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

## 1、ActiveMQ简介 ##

什么是ActiveMQ  
![在这里插入图片描述][20181110094050394.jpg]

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70]

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

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 1]

ActiveMQ应用场景分析：

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

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

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

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

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

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 2]

## 2、**ActiveMQ安装和使用** ##

官网：[http://activemq.apache.org/][http_activemq.apache.org]

下载

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 3]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 4]

下载windows版本

![在这里插入图片描述][20181110094158131.jpg]

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

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

用户名和密码 都是admin

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 5]

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

有两种数据结构 Queue、Topic (详见4.2)

1、Queue 队列（P2P消息模型） ，生产者生产了一个消息，只能由一个消费者进行消费 ：给微信好友发消息

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

微信公众号给粉丝发消息

JMS和ActiveMQ的对应

<table> 
 <thead> 
  <tr> 
   <th>JMS消息模型</th> 
   <th>P2P模式</th> 
   <th>pub/sub模式</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>ActiveMQ消息</td> 
   <td>Queue队列</td> 
   <td>Topic队列</td> 
  </tr> 
  <tr> 
   <td>特点</td> 
   <td>一对一，一个人发送，只允许一个人接收</td> 
   <td>一对多，一个人发送，允许多个人接收</td> 
  </tr> 
 </tbody> 
</table>

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

## 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("消息生产成功");
        }
    }

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

![在这里插入图片描述][20181110094409937.jpg]

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 6]

##### 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");
        }
    }

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

![在这里插入图片描述][20181110094438377.png]

第三步：查看页面效果

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

用户名和密码 都是admin

消费前：表示没有消费

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 7]

消费后：表示已经消费

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 8]

##### 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 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 9]

P2P消息模型中的多消费者模式，得出结论如下：

*  一个消息只能被一个消费者消费，不可重复消费
 *  多个消费者均分消息(负载均衡策略)
 *  当消费者在消费某个消息的时候，mq一定要等到它的成功回执，才会分发下一个消息

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 10]

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

\#\#\#3.3、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 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 11]

#### 2.5 queue和topic模式的比较 ####

相同点：

1、都只有一个生产者

2、都可以有多个消费者

不同点：

1、queue队列模式，一个消息只能被一个消费者消费，不能重复消费

当消费者消费某个消息的时候，一定要得到这个消息被成功消费的回执，才会分发下一个消息  
queue入队之后，无论等待多久，消息都会一直等待消费者来处理

2、topic广播模式，一个消息可以被多个消费者消费

这个消息无法被成功消费与否，都无所谓

topic要求时间要一致，我正好发，你正好收

### 3、SpringBoot整合ActiveMQ ###

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

我们采用外置ActiveMQ。

### 3.1、创建工程maven子模块 ###

1、在common\_parent中导入activeMQ的启动器

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-activemq</artifactId>
    </dependency>

2、创建bos-mq子模块

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 12]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 13]

![在这里插入图片描述][20181110094653760.png]

3、添加项目依赖

<dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-activemq</artifactId>
        </dependency>
    </dependencies>

4、创建程序入口

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 14]

代码如下：

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

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 15]

# 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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 16]

代码如下：

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

![在这里插入图片描述][2018111009490168.png]

/** * 消息的生产者 * @author Administrator * */
    @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

![在这里插入图片描述][20181110094914823.png]

/** * 消息的消费者 * @author Administrator */
    @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

7 应用场景：

*  用在高并发的请求中
    
     *  注册短信的发送
     *  注册邮件的发送
     *  秒杀
     *  两个系统间进行消息传递

#### 4.1、消息队列应用场景 ####

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

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

##### 4.1.1异步处理 #####

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

1.串行方式；2.并行方式

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

![在这里插入图片描述][20181110095339781.jpg]

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 17]

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

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

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

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

![在这里插入图片描述][20181110095403937.jpg]

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

##### 4.1.2应用解耦 #####

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

![在这里插入图片描述][20181110095413892.jpg]

传统模式的缺点：

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

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

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

![在这里插入图片描述][20181110095423736.jpg]

*  订单系统：用户下单后，订单系统完成持久化处理，将消息写入消息队列，返回用户订单下单成功
 *  库存系统：订阅下单的消息，采用pub/sub(发布/订阅)的方式，获取下单信息，库存系统根据下单信息，进行库存操作
 *  假如：在下单时库存系统不能正常使用。也不影响正常下单，因为下单后，订单系统写入消息队列就不再关心其他的后续操作了。实现订单系统与库存系统的应用解耦

##### 4.1.3流量削锋 #####

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

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

*  可以控制活动的人数
 *  可以缓解短时间内高流量压垮应用

![在这里插入图片描述][2018111009543822.jpg]

*  用户的请求，服务器接收后，首先写入消息队列。假如消息队列长度超过最大数量，则直接抛弃用户请求或跳转到错误页面
 *  秒杀业务根据消息队列中的请求信息，再做后续处理

##### 4.1.4日志处理 #####

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

![在这里插入图片描述][20181110095454192.jpg]

*  日志采集客户端，负责日志数据采集，定时写受写入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消息通讯 #####

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

点对点通讯：

![在这里插入图片描述][20181110095510354.jpg]

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

聊天室通讯：

![在这里插入图片描述][20181110095519928.jpg]

客户端A，客户端B，客户端N订阅同一主题，进行消息发布和接收。实现类似聊天室效果。

以上实际是消息队列的两种消息模式，点对点或发布订阅模式。模型为示意图，供参考。

### 4.2、JMS消息服务 ###

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

##### 4.2.1 消息模型 #####

在JMS标准中，有两种消息模型P2P（Point to Point）,Publish/Subscribe(Pub/Sub)。

##### 4.2.2 P2P模式-队列模式 #####

![在这里插入图片描述][20181110095546113.jpg]

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

P2P的特点

*  每个消息只能被一个消费者（Consumer）消费(即一旦被消费，消息就不再存在于消息队列中)
 *  发送者和接收者之间在时间上没有依赖性，也就是说当发送者发送了消息之后，不管接收者有没有正在运行，它不会影响到消息被发送到队列
 *  接收者在成功接收消息之后需向队列应答成功

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

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 18]

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

Pub/Sub的特点

*  每个消息可以有多个消费者
 *  发布者和订阅者之间有时间上的依赖性。针对某个主题（Topic）的订阅者，它必须创建一个订阅者之后，才能消费发布者的消息
 *  为了消费消息，订阅者必须保持运行的状态

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

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

#### 4.3.消息消费方式 ####

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

（1）同步

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

（2）异步

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

[20181110094050394.jpg]: /images/20220417/bbc3984191184387917e491c5dfb004b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70]: /images/20220417/d2e24cab357b4c718dfce2db4f4e0026.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 1]: /images/20220417/8591a134d1a44550a76f82064c0e83b8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 2]: /images/20220417/147020dbd47f4f7d835dfa068b2de9d3.png
[http_activemq.apache.org]: http://activemq.apache.org/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 3]: /images/20220417/194a0e186a9747f58fae72d4e6d18ee2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 4]: /images/20220417/4b6aaa26428b4c84a8d0871355c95662.png
[20181110094158131.jpg]: /images/20220417/8761b741045a49e59272e46ab9edbb0f.png
[http_localhost_8161]: http://localhost:8161/
[Manage ActiveMQ broker]: http://localhost:8161/admin/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 5]: /images/20220417/c8fabe8b537543fa91bd1accc790de4b.png
[20181110094409937.jpg]: /images/20220417/5852500697ae48f5812cd87ecbb4da24.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 6]: /images/20220417/550b82c705494f56ae7ff9b3500a46c8.png
[20181110094438377.png]: /images/20220417/b18919b8b8f644cbbd826834ffd5814f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 7]: /images/20220417/bab958327818453aac4bd26a17d3682e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 8]: /images/20220417/4b1e695cea9e4dbda4f80be26975152d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 9]: /images/20220417/bc822315bd4844c080484ced362e5405.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 10]: /images/20220417/e64e16c6fee04fffa107e423887e229d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 11]: /images/20220417/37cc017a6d5645bb852ab6d539b0955f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 12]: /images/20220417/a270efb66a854163b14affcd182e608a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 13]: /images/20220417/2e9a69c089c5417e9304628d1b7e22f3.png
[20181110094653760.png]: /images/20220417/9dd921e1eaa74d1390e1b3e7a0febf6c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 14]: /images/20220417/4728b9f7e2294e2b956c1b3154392a4f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 15]: /images/20220417/a6daf9f45d494c51b22c15bb5e0b8bee.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 16]: /images/20220417/7cd95af5e4be4bb88b09f1249c21c00a.png
[2018111009490168.png]: /images/20220417/1552bf7c1e18498f995f0410dbd04bf8.png
[20181110094914823.png]: /images/20220417/afb40520f7eb48c6af806f983636c24c.png
[20181110095339781.jpg]: /images/20220417/252d9fa729c342a4923590baca0f1d8e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 17]: /images/20220417/28ec726272884d47916550b956ab500f.png
[20181110095403937.jpg]: /images/20220417/190948b2c5d54be9abe51efa8d5ecac4.png
[20181110095413892.jpg]: /images/20220417/7a97adcb6a9b418ca73c1232f2e2a461.png
[20181110095423736.jpg]: /images/20220417/bf601335d54b49d5b4d375b6b96b1744.png
[2018111009543822.jpg]: /images/20220417/683634c0439343b79619f5ecfb7026ef.png
[20181110095454192.jpg]: /images/20220417/fe3ceb37712b4dfe9f317af6eed914a4.png
[20181110095510354.jpg]: /images/20220417/9d96b98993864a7388e7d593d615b8f5.png
[20181110095519928.jpg]: /images/20220417/a76a9980a809485a95fa6d69cb6ec374.png
[Java]: http://lib.csdn.net/base/17
[20181110095546113.jpg]: /images/20220417/2395112dcd0449f29a94390094221510.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjUyNTA5_size_16_color_FFFFFF_t_70 18]: /images/20220417/11a4281798d448b982b9e3f8c743cdc5.png