消息队列是应用程序之间的通信方法；
无需即时返回的且耗时的操作进行异步处理从而提高系统的吞吐量；
消息队列相当于一个中介，生产方与消费方之间没有联系，而是通过MQ来进行互动，可以实现程序之间的解耦合。
有两种主流的方式：
- AMQP：AMQP只是协议，不规定实现方式，因此是跨语言的。
- JMS:JMS是api接口，限定必须使用Java语言
activeMQ，zeroMQ，RabbitMQ，RocketMQ，kafka
4. 安装及配置RabbitMQ- 安装erlang（RabbitMQ是基于erlang语言的）:
- 安装RabbitMQ;
- 安装RabbitMQ的图形管理界面插件;
- 创建管理用户
- 创建虚拟主机Virtual Hosts
注意：安装上述组件的时候都需要使用以管理员身份运行。
4.1 用户角色配置RabbitMQ在安装好后，可以访问http://localhost:15672 ；其自带了guest/guest的用户名和密码；如果需要创建自定义用户；那么也可以登录管理界面后，如下操作：
4.2 Virtual Hosts（虚拟主机）配置像MySQL拥有数据库的概念一样，RabbitMQ中可以自定义增加Virtual Hosts（虚拟主机），并且每个虚拟主机都可以设置哪个用户可以管理，各个虚拟主机都是完全隔离的，互不干扰。
- 创建Virtual Hosts
- 设置Virtual Hosts权限
5.2 入门工程-生产者com.rabbitmq amqp-client 5.6.0
目标：编写消息生产者代码，发送消息到队列
分析：生产者发送消息到RabbitMQ的队列（simple_queue）；消费者可以从队列中获取消息。可以使用RabbitMQ的简单模式（simple）。
生产者实现发送消息的步骤：
- 创建连接工厂（设置RabbitMQ的连接参数）;
- 创建连接；
- 创建频道；
- 声明队列；
- 发送消息；
- 关闭资源
package com.itheima.rabbitmq.simple;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Producer {
static final String QUEUE_NAME = "simple-queue";//设置队列名称
public static void main(String[] args) throws IOException, TimeoutException {
//1. 创建连接工厂（设置RabbitMQ的连接方式）
ConnectionFactory connectionFactory = new ConnectionFactory();
//主机；默认localhost
connectionFactory.setHost("localhost");
//连接端口；默认是5672
connectionFactory.setPort(5672);
//虚拟主机：默认是/
connectionFactory.setVirtualHost("/itcast");
//用户名；默认是guest
connectionFactory.setUsername("heima");
//密码；默认是guest
connectionFactory.setPassword("heima");
//2. 创建连接
Connection connection = connectionFactory.newConnection();
//3. 创建频道
Channel channel = connection.createChannel();
//4. 声明队列
channel.queueDeclare(QUEUE_NAME,true,false,false,null);
//5. 发送消息；
String message = "你好！小兔纸。";
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println("已发送消息：" + message);
//6. 关闭资源
channel.close();
connection.close();
}
}
注意：在设置连接工厂的时候；如果没有指定连接的参数则会用默认值。
5.3 入门工程-消费者目标：编写消息消费者代码，从队列中接收消息并消费。
分析：从RabbitMQ中队列（与生产者发送消息时的队列一致；simple_queue）接收消息；
实现消费者步骤：
- 创建连接工厂；
- 创建连接；（抽取一个获取连接的工具类）
- 创建频道；
- 声明队列；
- 创建消费者（接收消息并处理消息）；
- 监听队列
工具类的编写：
package com.itheima.rabbitmq.util;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ConnectionUtil {
public static Connection getConnection() throws IOException, TimeoutException {
//1. 创建连接工厂（设置RabbitMQ的连接方式）
ConnectionFactory connectionFactory = new ConnectionFactory();
//主机；默认localhost
connectionFactory.setHost("localhost");
//连接端口；默认是5672
connectionFactory.setPort(5672);
//虚拟主机：默认是/
connectionFactory.setVirtualHost("/itcast");
//用户名；默认是guest
connectionFactory.setUsername("heima");
//密码；默认是guest
connectionFactory.setPassword("heima");
//2. 创建连接
Connection connection = connectionFactory.newConnection();
return connection;
}
}
package com.itheima.rabbitmq.simple;
import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer {
public static void main(String[] args) throws Exception {
//1. 创建连接工厂；
//2. 创建连接；（抽取一个获取连接的工具类）
Connection connection = ConnectionUtil.getConnection();
//3. 创建频道；
Channel channel = connection.createChannel();
//4. 声明队列；
channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
//5. 创建消费者（接收消息并处理消息）；
DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//路由key
System.out.println("路由key为：" + envelope.getRoutingKey());
//交换机
System.out.println("交换机为：" + envelope.getExchange());
//消息id
System.out.println("消息id为：" + envelope.getDeliveryTag());
//接收到的消息
System.out.println("接收到的消息为：" + new String(body, "utf-8"));
}
};
//6. 监听队列
channel.basicConsume(Producer.QUEUE_NAME, true, defaultConsumer);
}
}
注意：需要持续监听队列消息，所以不要关闭资源
5.4 入门工程测试目标：启动消费者和生产者，到RabbitMQ中查询队列并在消费者端IDEA控制台查看接收到的消息
分析：
生产者：发送消息到RabbitMQ队列（simple_queue）
消费者：接收RabbitMQ队列消息
小结：
简单模式：生产者发送消息到队列中，一个消费者从队列中接收消息。
6. Work queues工作队列模式在RabbitMQ中消费者只能从队列接收消息。
如果接收消息的消费者在同一个队列中有两个或多个时；消息是如何分配的？
工作队列模式：在同一个队列中可以有多个消费者，消费者之间对于消息的接收是竞争关系。
生产者：发送30个消息
消费者：创建两个消费者监听同一个队列，查看两个消费者的接收消息是否存在重复。
生产者
package com.itheima.rabbitmq.work;
import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Producer {
static final String QUEUE_NAME = "work-queue";//设置队列名称
public static void main(String[] args) throws IOException, TimeoutException {
//1. 创建连接工厂（设置RabbitMQ的连接方式）
//2. 创建连接
Connection connection = ConnectionUtil.getConnection();
//3. 创建频道
Channel channel = connection.createChannel();
//4. 声明队列
channel.queueDeclare(QUEUE_NAME,true,false,false,null);
for (int i=1; i<=30; i++) {
//5. 发送消息；
String message = "你好！小兔纸。work模式"+ i;
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println("已发送消息：" + message);
}
//6. 关闭资源
channel.close();
connection.close();
}
}
消费者1
package com.itheima.rabbitmq.work;
import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer1 {
public static void main(String[] args) throws Exception {
//1. 创建连接工厂；
//2. 创建连接；（抽取一个获取连接的工具类）
Connection connection = ConnectionUtil.getConnection();
//3. 创建频道；
final Channel channel = connection.createChannel();
//4. 声明队列；
channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
//每次可以预取多少个消息
channel.basicQos(1);
//5. 创建消费者（接收消息并处理消息）；
DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//路由key
System.out.println("路由key为：" + envelope.getRoutingKey());
//交换机
System.out.println("交换机为：" + envelope.getExchange());
//消息id
System.out.println("消息id为：" + envelope.getDeliveryTag());
//接收到的消息
System.out.println("消费者1---接收到的消息为：" + new String(body, "utf-8"));
try {
//休眠1s钟，可以看的更明显
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
//确认消息
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
//6. 监听队列
channel.basicConsume(Producer.QUEUE_NAME, true, defaultConsumer);
}
}
消费者2
package com.itheima.rabbitmq.work;
import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer2 {
public static void main(String[] args) throws Exception {
//1. 创建连接工厂；
//2. 创建连接；（抽取一个获取连接的工具类）
Connection connection = ConnectionUtil.getConnection();
//3. 创建频道；
final Channel channel = connection.createChannel();
//4. 声明队列；
channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
//每次可以预取多少个消息
channel.basicQos(1);
//5. 创建消费者（接收消息并处理消息）；
DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//路由key
System.out.println("路由key为：" + envelope.getRoutingKey());
//交换机
System.out.println("交换机为：" + envelope.getExchange());
//消息id
System.out.println("消息id为：" + envelope.getDeliveryTag());
//接收到的消息
System.out.println("消费者2---接收到的消息为：" + new String(body, "utf-8"));
try {
//休眠1s钟，可以看的更明显
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
//确认消息
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
//6. 监听队列
channel.basicConsume(Producer.QUEUE_NAME, true, defaultConsumer);
}
}
总结：
工作队列模式：一个消息只能被一个消费者接收，其它消费者是不能接收到同一条消息的
订阅模式与前面的两种模式比较：多了一个角色Exchange交换机，接收生产者发送的消息并决定如何投递消息到其绑定的队列；消息的投递取决于交换机的类型。
交换机的类型：
- Fanout：广播，将消息交给所有绑定到交换机的队列
- Direct：定向，把消息交给符合指定routing key 的队列
- Topic：通配符，把消息交给符合routing pattern（路由模式） 的队
交换机只做消息转发，自身不存储任何数据
8. 发布与订阅模式消息到了队列，只有一个消费者可以拿到消息。如果想要多个消费者拿到同一个消息，要通过交换机设置多个队列。
一个消息可以被多个消费者接收；其实是使用了订阅模式，交换机类型为：fanout广播
- 生产者（发送10个消息）
- 创建连接；
- 创建频道；
- 声明交换机（fanout）；
- 声明队列；
- 队列绑定到交换机；
- 发送消息；
- 关闭资源
- 消费者（至少两个消费者）
- 创建连接；
- 创建频道；
- 声明交换机；
- 声明队列；
- 队列绑定到交换机；
- 创建消费者；
- 监听队列；
小结：
9. Routing路由模式发布与订阅模式：一个消息可以被多个消费者接收；一个消费者对应的队列，该队列只能被一个消费者监听。使用了订阅模式中交换机类型为：广播（fanout）。
路由模式特点：
- 队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）。
- 消息的发送方在向Exchange发送消息时，也必须指定消息的RoutingKey。
- Exchange不再把消息发给每一个绑定的队列，而是根据消息的RoutingKey进行判断，只有队列的RoutingKey与消息的RoutingKey完全一致，才会接收消息。
- 使用了订阅模式中交换机类型为：定向（direct）。
Topic 类型与 Direct 相比，都是可以根据 RoutingKey 把消息路由到不同的队列。只不过 Topic 类型 Exchange 可以让队列在绑定 Routing key 的时候使用通配符！
通配符规则：
#：匹配一个或多个词
例如：item.# ：能够匹配 item.insert.abc 或者 item.insert
*：匹配不多不少恰好一个词
例如：item * ：只能匹配 item.insert
小结：
Topics通配符模式：可以根据路由key将消息传递到对应路由key的队列；队列绑定到交换机的路由key可以有多个；通配符模式中路由key可以使用 * 和 # ；使用了通配符模式之后对于路由Key的配置更加灵活。
- 不直接Exchange交换机（默认交换机）
- simple简单模式：一个生产者生产一个消息到一个队列被一个消费者接收
- work工作队列模式：生产者发送消息到一个队列中，然后可以被多个消费者监听该队列；一个消息只能被一个消费者接收，消费者之间是竞争关系
- 使用Exchange交换机；订阅模式（交换机：广播fanout、定向direct、通配符topic）
- 发布与订阅模式：使用了fanout广播类型的交换机，可以将一个消息发送到所有绑定了该交换机的队列
- 路由模式：使用了direct定向类型的交换机，消费会携带路由key，交换机根据消息的路由key与队列的路由key进行对比，一致的话那么该队列可以接收到消息
- 通配符模式：使用了topic通配符类型的交换机，消费会携带路由key（*， #），交换机根据消息的路由key与队列的路由key进行对比，匹配的话那么该队列可以接收到消息
Spring Boot提供了对于AMQP的整合；可以使用RabbitTemplate发送消息；可以使用@RabbitListener注解接收消息。
12.1 生产者工程
