MQ
MQ（Message Queue）消息队列，是基础数据结构中“先进先出”的一种数据结构。一般用来解决应用解耦，异步消息，流量削峰等问题，实现高性能，高可用，可伸缩和最终一致性架构。
主要的MQ产品包括：RabbitMQ、ActiveMQ、RocketMQ、ZeroMQ、Kafka、IBM WebSphere 等。
AMQP
AMQP简介：AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。 AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。
RabbitMQ
RabbitMQ是AMQP服务器的一种。
它是应用层协议的一个开放标准，为面向消息的中间件设计，基于此协议的客户端与消息中间件可传递消息，并不受产品、开发语言等条件的限制
RabbitMQ 是一个开源的消息代理和队列服务器，用来通过普通协议在完全不同的应用之间共享数据，RabbitMQ是使用 Erlang语言来编写的，并且RabbitMQ是基于AMQP协议的。
RabbitMQ是一个消息代理 - 一个消息系统的媒介。它可以为你的应用提供一个通用的消息发送和接收平台，并且保证消息在传输过程中的安全。
优点
- 开源、性能优秀、稳定性保障
- 提供可靠性消息投递模式(confirm)、返回模式(return)
- 与SpringAMQP完美的整合、API丰富
- 集群模式丰富，表达式配置，HA模式，镜像队列模型
- 保证数据不丢失的前提下做到高可靠性、可用性
架构
RabbitMQ运行机制
组成部分说明如下:
- Broker:消息队列服务进程，此进程包括两个部分:Exchange和Queue.
- Exchange :消息队列交换机，按一定的规则将消息路由转发到某个队列，对消息进行过虑。
- Queue :消息队列，存储消息的队列，消息到达队列并转发给指定的消费方。
- Producer:消息生产者，即生产方客户端，生产方客户端将消费发送到MQ…
- Consumer:消息消费者，即消费方客户端，接收MQ转发的消息。
消息发布接收流程∶
-----发送消息-----
6. 生产者和Broker建立TCP连接。
7. 生产者和Broker建立通道。
8. 生产者通过通道消息发送Broker，由Exchange将消息进行转发。
9. Exchange将消息转发到指定的Queue(队列)
linux安装RabbitMQ
docker拉取RabbitMQ镜像
docker pull rabbitmq:3-management
查看RabbitMQ镜像
docker images
运行RabbitMQ
docker run -d -p 5672:5672 -p 15672:15672 --name myrabbitmq imageid
其中:
-d:代表后台运行
-p 5672:5672:代表将主机端口号5672映射到docker的5672端口
-p 15672:15672:代表将主机端口号15672映射到docker的15672端口
–name myrabbitmq:代表启动的rabbitmq的名字是myrabbitmq
imageid:代表代表docker镜像的镜像id的值.
查看docker进程
docker ps
使用本地浏览器访问
地址：localhost（ip）:15672 默认账号：guest 密码：guest
springboot整合RabbitMQ
官方文档：https://www.rabbitmq.com/getstarted.html
pom.xml文件
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit-test</artifactId>
<scope>test</scope>
</dependency>
application.properties
spring.rabbitmq.host=66.112.211.169
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/
配置类
package com.etc.demomq1;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* 创建一个Queue
*/
@Configuration
public class DemoQueue {
@Bean
public Queue hello() {
return new Queue("hello");
}
}
发送消息
package com.etc.demomq1;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
@Component
public class MessageSender {
@Autowired
private AmqpTemplate amqpTemplate;
@Autowired
private Queue queue;
// @Autowired
private RabbitTemplate template;
/**
* 发送消息的方法
* @param username
*/
public void sendMessage(String username){
String msg = username + ": "+ LocalDateTime.now();
amqpTemplate.convertAndSend(queue.getName(),msg);
}
}
接收消息
package com.etc.demomq1;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
/**
* RabbitListener 监听的是hello的队列
*/
@Component
@RabbitListener(queues = "hello")
public class MessageReceiver {
/**
* 消息的处理
* @param msg
*/
@RabbitHandler
public void getMessage(String msg){
System.out.println("MessageReceiver :"+msg);
}
}
可以直接运行了
