消息队列，即MQ，Message Queue。
消息队列是典型的：生产者、消费者模型。生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入，这样就实现了生产者和消费者的解耦。
如果以后有其它系统也依赖商品服务的数据，同样监听消息即可，商品服务无需任何代码修改。
MQ是消息通信的模型，并不是具体实现。现在实现MQ的有两种主流方式：AMQP、JMS。
两者间的区别和联系：
RabbitMQ 和 Kafka 用的多一些
RabbitMQ是基于AMQP的一款消息管理系统
**官网： **http://www.rabbitmq.com/
官方教程：http://www.rabbitmq.com/getstarted.html
** **
RabbitMQ提供了6种消息模型，但是第6种其实是RPC，并不是MQ，因此不予学习。那么也就剩下5种。
但是其实3、4、5这三种都属于订阅模型，只不过进行路由的方式不同。
炳汉的独白
docker pull rabbitmq:3.8.9-management
注意：如果docker pull rabbitmq 后面不带management，启动rabbitmq后是无法打开管理界面的，所以我们要下载带management插件的rabbitmq.两个端口 管理15672,5672** **
docker images
docker run -di -p 5672:5672 -p 15672:15672 --name rabbitmq rabbitmq:3.8.9-management
访问管理界面的地址就是 http://192.168.153.130:15672，可以使用默认的账户登录，用户名和密码都guest
创建一个空项目 rabbit_demo
创建 SpringBoot 子模块
rabbit_base_demo
RabbitMQ是一个消息代理：它接受和转发消息。 你可以把它想象成一个邮局：当你把邮件放在邮箱里时，你可以确定邮差先生最终会把邮件发送给你的收件人。 在这个比喻中，RabbitMQ是邮政信箱，邮局和邮递员。
RabbitMQ与邮局的主要区别是它不处理纸张，而是接受，存储和转发数据消息的二进制数据块。
** **
P（producer/ publisher）：生产者，一个发送消息的用户应用程序。
C（consumer）：消费者，消费和接收有类似的意思，消费者是一个主要用来等待接收消息的用户应用程序
队列（红色区域）：RabbitMQ内部类似于邮箱的一个概念。虽然消息流经RabbitMQ和你的应用程序，但是它们只能存储在队列中。队列只受主机的内存和磁盘限制，实质上是一个大的消息缓冲区。许多生产者可以发送消息到一个队列，许多消费者可以尝试从一个队列接收数据。
总之：
生产者将消息发送到队列，消费者从队列中获取消息，队列是存储消息的缓冲区。
我们将用SpringBoot编写两个程序;发送单个消息的生产者，以及接收消息并将其打印出来的消费者。
我们将调用我们的消息发布者（发送者）Send和我们的消息消费者（接收者）Recv。发布者将连接到RabbitMQ，发送一条消息，然后退出。
<packaging>pom</packaging>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!--<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit-test</artifactId>
<scope>test</scope>
</dependency>-->
</dependencies>
**packaging 为 pom ，建立子模块 provider 和 consumer **
application.yml，代码如下
spring:
rabbitmq:
host: 192.168.153.130
port: 5672
username: guest
password: guest
server:
port: 10001 #消费者为 10002
@SpringBootApplication
public class ProviderApp {
public static void main(String[] args) {
SpringApplication.run(ProviderApp.class);
}
}
@Configuration
public class BaseQueueConfig {
/**
* 定义简单队列
*/
private String simpleQueue = "bh.simple.queue";
@Bean
public Queue simpleQueue() {
return new Queue(simpleQueue);
}
}
@RestController
public class SendController {
@Autowired
private RabbitTemplate rabbitTemplate;
@RequestMapping("/")
public void send() {
for(int i=0;i<5;i++) {
rabbitTemplate.convertAndSend("bh.simple.queue", "简单队列消息"+i);
}
}
}
@Component
public class ConsumerListener {
@RabbitListener(queues= "bh.simple.queue")
public void recieve(String message) {
System.out.println(message);
}
}
工作队列或者竞争消费者模式
** **
创建一个工作队列，在多个工作者之间分配耗时任务。
工作队列，又称任务队列。主要思想就是避免执行资源密集型任务时，必须等待它执行完成。相反我们稍后完成任务，我们将任务封装为消息并将其发送到队列。 在后台运行的工作进程将获取任务并最终执行作业。当你运行许多工人时，任务将在他们之间共享，但是一个消息只能被一个消费者获取。
这个概念在Web应用程序中特别有用，因为在短的HTTP请求窗口中无法处理复杂的任务。
接下来我们来模拟这个流程：
** P：生产者：任务的发布者**
** C1：消费者，领取任务并且完成任务，假设完成速度较快**
** C2：消费者2：领取任务并完成任务，假设完成速度慢 **
面试题：避免消息堆积？
1）采用workqueue，多个消费者监听同一队列。
2）接收到消息以后，而是通过线程池，异步消费。
@Configuration
public class WorkQueueConfig {
/**
* 定义简单队列
*/
private String simpleQueue = "work_queue";
@Bean
public Queue simpleQueue() {
return new Queue(simpleQueue);
}
}
@RestController
public class SendController {
@Autowired
private RabbitTemplate rabbitTemplate;
@RequestMapping("/")
public void send() {
for (int i = 0; i < 8; i++) {
String message = "工作队列消息" + i;
rabbitTemplate.convertAndSend("work_queue", message);
}
}
}
@Component
public class ReceiveListener {
@RabbitListener(queues = "work_queue")
public void recieve1(String message) {
System.out.println("listener1:" + message);
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@RabbitListener(queues = "work_queue")
public void recieve2(String message) {
System.out.println("listener2:" + message);
}
}
我们在消费者1里设置延时时间，模式消费耗时
控制台输出
默认平均分配/统筹分配 任务一分为二**** 每个人 分配的消息数量相同
刚才的实现有问题吗？
现在的状态属于是把任务平均分配，正确的做法应该是消费越快的人，消费的越多。
怎么实现呢？
我们可以设置每次只处理一个信息，告诉RabbitMQ一次不要向消费端发送多于一条消息。 或者换句话说，不要向消费端发送新消息，直到它处理并确认了前一个消息。 相反，它会将其分派给不是处于忙碌状态的下一个消费端。
修改配置文件，如下：
spring:
rabbitmq:
host: 192.168.153.130
port: 5672
username: guest
password: guest
listener:
simple:
prefetch: 1 # 每次只处理一个信息 实现能者多劳
server:
port: 10003
交换机实现多个队列消息传递
**在之前的模式中，我们创建了一个工作队列。 工作队列背后的假设是：每个任务只被传递给一个工作人员。 在这一部分，我们将做一些完全不同的事情 - 我们将会传递一个信息给多个消费者。 这种模式被称为“发布/订阅”。 **
解读：
1、1个生产者，多个消费者
2、每一个消费者都有自己的一个队列
3、生产者没有将消息直接发送到队列，而是发送到了交换机
4、每个队列都要绑定到交换机
5、生产者发送的消息，经过交换机到达队列，实现一个消息被多个消费者获取的目的
X（Exchanges）：交换机一方面：接收生产者发送的消息。另一方面：知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。
Exchange类型有以下几种：
** Fanout：广播，将消息交给所有绑定到交换机的队列**
** Direct：定向，把消息交给符合指定routing key 的队列 (默认)**
** Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列**
我们这里先学习
** **Fanout：即广播模式
Exchange（交换机）只负责转发消息，不具备存储消息的能力**，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！**
@RestController
public class SendController {
@Autowired
private RabbitTemplate rabbitTemplate;
@RequestMapping("/")
public void send() {
for (int i = 0; i < 5; i++) {
String message = "订阅模式消息" + i;
// 第二个参数路由 不需要
rabbitTemplate.convertAndSend("fanoutExchange", "", message);
}
}
}
@Component
public class ReceiveListener {
@RabbitListener(bindings =
@QueueBinding(
value = @Queue, // 临时创建队列
exchange = @Exchange(value = "fanoutExchange",type="fanout") // 交换机
)
)
public void recieve1(String message) {
System.out.println("listener1:" + message);
}
@RabbitListener(bindings =
@QueueBinding( // 绑定队列
value = @Queue, // 临时创建队列
exchange = @Exchange(value = "fanoutExchange",type="fanout") // 交换机
)
)
public void recieve2(String message) {
System.out.println("listener2:" + message);
}
}
**访问 **http://localhost:10004/
输出
有选择性的接收消息
在订阅模式中，生产者发布消息，所有消费者都可以获取所有消息。
在路由模式中，我们将添加一个功能 - 只能订阅一部分消息。 例如，我们只能将重要的错误消息引导到日志文件（以节省磁盘空间），同时仍然能够在控制台上打印所有日志消息。
但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
在Direct模型下，队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
消息的发送方在向Exchange发送消息时，也必须指定消息的routing key。
** **
P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列
C1：消费者，其所在队列指定了需要routing key 为 error 的消息
C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息
@RestController
public class SendController {
@Autowired
private RabbitTemplate rabbitTemplate;
@RequestMapping("/send01")
public void sendMessageA() {
for (int i = 0; i < 5; i++) {
String message = "路由模式--routingKey=update消息" + i;
System.out.println("我是生产信息的：" + message);
rabbitTemplate.convertAndSend("directExchange", "update", message);
}
}
@RequestMapping("/send02")
public void sendMessageB() {
for (int i = 0; i < 5; i++) {
String message = "路由模式--routingKey=add消息" + i;
System.out.println("我是生产信息的：" + message);
rabbitTemplate.convertAndSend("directExchange", "add", message);
}
}
}
@Component
public class ReceiveListener {
@RabbitListener(bindings =
@QueueBinding(
value = @Queue, // 临时创建队列
// 交换机 direct是默认值，可以不定义
exchange = @Exchange(value = "directExchange",type = "direct"),
key = {"update","add"} // 接收的路由
)
)
public void recieve_direct(String message) {
System.out.println("listener_direct_1:" + message);
}
@RabbitListener(bindings =
@QueueBinding(
value = @Queue, // 临时创建队列
// 交换机 direct是默认值，可以不定义
exchange = @Exchange(value = "directExchange",type = "direct"),
key = {"add"} // 接收的路由
)
)
public void recieve_direct2(String message) {
System.out.println("listener_direct_2:" + message);
}
}
访问http://localhost:10005/send01
访问http://localhost:10005/send02** **
Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key** 的时候使用通配符！**
Routingkey** 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： **item.insert
** 通配符规则：**
** #：匹配一个或多个词**
** *：匹配不多不少恰好1个词**
举例：
** audit.#：能够匹配**audit.irs.corporate** 或者 **audit.irs
** audit.*：只能匹配**audit.irs
** **
在这个例子中，我们将发送所有描述动物的消息。消息将使用由三个字（两个点）组成的routing key发送。路由关键字中的第一个单词将描述速度，第二个颜色和第三个种类：“
我们创建了三个绑定：Q1绑定了绑定键“ .orange.***”，Q2绑定了“.*.rabbit”和“lazy.＃”。
Q1匹配所有的橙色动物。
Q2匹配关于兔子以及懒惰动物的消息。
@RestController
public class SendController {
@Autowired
private RabbitTemplate rabbitTemplate;
@RequestMapping("/send01")
public void send01() {
for (int i = 0; i < 5; i++) {
String message = "通配符模式--routingKey=topic.keyA消息" + i;
System.out.println("我是生产信息的：" + message);
rabbitTemplate.convertAndSend("topicExchange", "topic.keyA", message);
}
}
@RequestMapping("/send02")
public void send02(){
for (int i = 0; i < 5; i++) {
String message = "通配符模式--routingKey=topic.#消息" + i;
System.out.println("我是生产信息的：" + message);
rabbitTemplate.convertAndSend("topicExchange", "topic.keyD.keyE", message);
}
}
}
@Component
public class ReceiveListener {
@RabbitListener(bindings =
@QueueBinding(
value = @Queue, // 临时创建队列
exchange = @Exchange(type = "topic" , name = "topicExchange"), // 交换机
key = {"topic.*"} // 接收的路由 topic.后有一个词
)
)
public void recieve_direct(String message) {
System.out.println("listener_direct_1:" + message);
}
@RabbitListener(bindings =
@QueueBinding(
value = @Queue, // 临时创建队列
exchange = @Exchange(type = "topic" , name = "topicExchange"), // 交换机
key = {"topic.#"} // 接收的路由 topic.后有多个词
)
)
public void recieve_direct2(String message) {
System.out.println("listener_direct_2:" + message);
}
}
访问http://localhost:10006/send01
输出，发现两个消费者都能消费
访问http://localhost:10006/send02
输出，发现只有消费者2 可以消费
查看RabbitMq 服务器：
