一：什么是MQ
MQ，中文是消息队列（MessageQueue），字面来看就是存放消息的队列。也就是事件驱动架构中的Broker。比较常见的MQ实现：
- ActiveMQ
- RabbitMQ
- RocketMQ
- Kafka
几种常见MQ的对比：
| RABBITMQ | ACTIVEMQ | ROCKETMQ | KAFKA | |
| 公司/社区 | Rabbit | Apache | 阿里 | Apache |
| 开发语言 | Erlang(并发能力强，性能及其好) | Java | Java | Scala&Java |
| 协议支持 | AMQP，XMPP，SMTP，STOMP | OpenWire,STOMP，REST,XMPP,AMQP | 自定义协议 | 自定义协议 |
| 可用性 | 高 | 一般 | 高 | 高 |
| 单机吞吐量(性能承载能力) | 一般 | 差 | 高 | 非常高 |
| 消息延迟 | 微秒级 | 毫秒级 | 毫秒级 | 毫秒以内 |
| 消息可靠性 | 高 | 一般 | 高 | 一般 |
追求可用性(当需要处理数据时，资源处于可用状态的程度)：Kafka、 RocketMQ 、RabbitMQ
追求可靠性：RabbitMQ、RocketMQ
追求吞吐能力(十万级别的)：RocketMQ、Kafka
追求消息低延迟：RabbitMQ、Kafka
二：安装RabbitMQ
我们将RabbitMQ安装在linux上使用以下两种方式可以安装（安装过程保证自己的docker容器一直是开着的）
第一种方法使用docker镜像，直接拉取安装即可
命令：docker pull rabbitmq:3.8-management
第二种方法 使用 本地下载
RabbitMQ的镜像包我已上传在我的个人博客资源中，将镜像包拖到tmp目录下
使用命令
//在线解压即可
docker load -i mq.tar
如下： 
解压好后，接着输入以下命令
docker run \
# 设置环境变量 用户名是 dlwlrma
-e RABBITMQ_DEFAULT_USER=itcast \
# 设置环境变量 密码是 zhien0516
-e RABBITMQ_DEFAULT_PASS=123321 \
# 挂载数据卷，后面高级会用到下面的插件
-v mq-plugins:/plugins \
# mq的名字
--name mq \
# 主机名 这里不配置也可以，后期如果是集群必须配置
--hostname mq \
# web可视化终端监控端口；mq的ui界面管理平台端口
-p 15672:15672 \
# 程序与mq交互的访问端口；发消息和收消息的端口
-p 5672:5672 \
# 后端运行
-d \
# 镜像名称
rabbitmq:3.8-management
精简命令如下：
docker run \
-e RABBITMQ_DEFAULT_USER=dlwlrma\
-e RABBITMQ_DEFAULT_PASS=zhien0516 \
-v mq-plugins:/plugins \
--name mq \
--hostname mq \
-p 15672:15672 \
-p 5672:5672 \
-d \
rabbitmq:3.8-management
安装好后，可以在浏览器中输入自己的linux的ip地址访问即可出现以下画面，输入自己刚刚设置的用户名和密码即可成功登录
三：在Java中如何实现MQ的使用
RabbitMQ的五种消息模型
1.基本消息队列（BasicQueue）
P（producer/ publisher）：生产者，一个发送消息的用户应用程序。我们自己书写代码发送。
C（consumer）：消费者，消费和接收有类似的意思，消费者是一个主要用来等待接收消息的用户应用程序。我们自己书写代码接收。
队列（红色区域）：存在于rabbitmq内部。许多生产者可以发送消息到一个队列，许多消费者可以尝试从一个队列接收数据。
总之：生产者将消息发送到队列，消费者从队列中获取消息，队列是存储消息的缓冲区。
2.工作消息队列（WorkQueue）
工作消息队列是基本消息队列的增强版，具有多个消费者消费队列的消息。假设消息队列中积压了多个消息，那么此时可以使用多个消费者来消费队列中的消息。效率要比基本消息队列模型高。
3. 发布订阅（Publish、Subscribe）
又根据交换机类型不同分为三种：
但是三种类型都有一个共性：
1、1个生产者，多个消费者
2、每一个消费者都有自己的一个队列
3、生产者没有将消息直接发送到队列，而是发送到了交换机
4、每个队列都要绑定到交换机
5、生产者发送的消息，经过交换机到达队列，实现一个消息被多个消费者获取的目的
X（Exchanges）：交换机一方面：接收生产者发送的消息。另一方面：知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失
广播类型：
将消息交给所有绑定到交换机的队列,生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。交换机把消息发送给绑定过的所有队列.队列的消费者都能拿到消息。实现一条消息被多个消费者消费.
4.Direct Exchange：路由
1.在广播模式中，生产者发布消息，所有消费者都可以获取所有消息。
2.在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。在Direct模型下，队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）.消息的发送方在向Exchange发送消息时，也必须指定消息的routing key。
3.P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
4.X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列
5.Topic Exchange：主题
1.Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！
2.Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert
3.通配符规则：
#：匹配一个或多个词
*：匹配恰好1个词
在Java中实现五种模型
1.基本消息队列
消费者：
import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; //将当前类放到SpringIOC容器中 @Component public class SpringRabbitListener { /** * * @author dlwlrma * @date 2024/6/17 20:18 * @param msg * TODO: @RabbitListener 为定义监听哪一个队列 */ @RabbitListener(queues = {"simple.queue"}) //Spring自动将接收的消息给方法参数msg public void listenSimpleQueueMessage(String msg) throws InterruptedException { System.out.println("spring 消费者接收到消息：【" + msg + "】"); } }
发布者：
import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; //TODO:是一个测试启动器，可以加载SpringBoot测试注解,让测试方法在Spring容器环境下执行 @RunWith(SpringRunner.class) //TODO:目的是加载ApplicationContext，启动spring容器 @SpringBootTest public class SpringAmqpTest { //自动装配RabbitTemplate模板对象 @Autowired private RabbitTemplate rabbitTemplate; /** * * @author dlwlrma * @date 2024/6/17 20:23 * TODO： Basic Queue 简单队列模型 */ @Test public void testSimpleQueue() { //1.定义变量保存队列的名字 String queueName = "simple.queue"; //2.定义变量保存要发送的消息 String message = "hello啊，我是mq,我发送的是简单队列模型"; //3.发送消息 rabbitTemplate.convertAndSend(queueName,message); } }
2.工作消息队列
消费者：
@RabbitListener(queues = {"simple.queue"}) public void listenWorkQueue1Message(String msg) throws InterruptedException { System.out.println("用户1已收到了消息："+msg); //休眠20毫秒 Thread.sleep(20); } @RabbitListener(queues = {"simple.queue"}) public void listenWorkQueue2Message(String msg) throws InterruptedException { System.out.println("用户2已收到了消息："+msg); //休眠100毫秒 Thread.sleep(100); }
发布者:
/** * * @author dlwlrma * @date 2024/6/17 20:24 * TODO：workQueue 向队列中不停发送消息，模拟消息堆积。 */ @Test public void testWorkQueue() throws InterruptedException { //1.定义变量保存队列的名字 String queueName = "simple.queue"; //2.定义变量保存要发送的消息 String message = "我是mq,我发送的是hello_message_"; //3.发送消息 for (int i = 1; i <=50 ; i++) { rabbitTemplate.convertAndSend(queueName,message+i); //4.模拟每隔20毫秒发送一次 Thread.sleep(20); } }
3.发布订阅模型--广播
定义绑定类：将交换机与多个队列绑定起来
import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.FanoutExchange; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class FanoutConfig { //1.定义交换机的名字 @Bean public FanoutExchange fanoutExchange(){ return new FanoutExchange("dlwlrma.fanout"); } //2.定义队列1 @Bean public Queue fanoutQueue1(){ return new Queue("fanout.queue1"); } //3.将交换机与队列绑定起来 @Bean public Binding bindingExchangeToFanoutQueue1(FanoutExchange fanoutExchange,Queue fanoutQueue1){ //将队列1与交换机绑定起来 return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange); } //4.定义队列2 @Bean public Queue fanoutQueue2(){ return new Queue("fanout.queue2"); } //3.将交换机与队列绑定起来 @Bean public Binding bindingExchangeToFanoutQueue2(FanoutExchange fanoutExchange,Queue fanoutQueue2){ //将队列2与交换机绑定起来 return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange); } }
消费者：
/** * * @author dlwlrma * @date 2024/6/17 22:14 * @param msg * TODO:模拟订阅发布：广播类型 */ @RabbitListener(queues = {"fanout.queue1"}) public void listenFAnoutQueue1Message(String msg) { System.out.println("用户1已收到了消息："+msg); } @RabbitListener(queues = {"fanout.queue2"}) public void listenFAnoutQueue2Message(String msg) { System.out.println("用户2已收到了消息："+msg); }
发布者：
/** * * @author dlwlrma * @date 2024/6/17 20:24 * TODO： Fanout Exchange：广播 */ @Test public void testFanoutExchange() { //1.定义变量保存交换机名字 String exchangeName = "dlwlrma.fanout"; //2.定义变量保存要发送的信息 String message = "hello，我是订阅发布"; //3.发送消息 //第二个参数为广播类型 rabbitTemplate.convertAndSend(exchangeName,"",message); }
4.发布订阅模型---路由
用基于注解的方式定义交换机和队列在consumer的SpringRabbitListener中添加两个消费者
/** * * @author dlwlrma * @date 2024/6/18 23:42 * @param msg * TODO: 1. value = @Queue(name = "direct.queue1") 表示绑定的第一个队列 * 2.exchange = @Exchange(name = "dlwlrma.direct", type = ExchangeTypes.DIRECT) 表示交换机名和类型 * 3.key = {"red", "blue"} 表示路由key */ @RabbitListener(bindings = @QueueBinding( value = @Queue(name = "direct.queue1"), exchange = @Exchange(name = "dlwlrma.direct", type = ExchangeTypes.DIRECT), key = {"red", "blue"} )) public void listenDirectQueue1(String msg){ System.out.println("消费者接收到direct.queue1的消息：【" + msg + "】"); } @RabbitListener(bindings = @QueueBinding( value = @Queue(name = "direct.queue2"), exchange = @Exchange(name = "dlwlrma.direct", type = ExchangeTypes.DIRECT), key = {"red", "yellow"} )) public void listenDirectQueue2(String msg){ System.out.println("消费者接收到direct.queue2的消息：【" + msg + "】"); }
发布者：
/** * * @author dlwlrma * @date 2024/6/17 20:24 * TODO：Direct Exchange：路由 */ @Test public void testSendDirectExchange() { // 交换机名称 String exchangeName = "dlwlrma.direct"; // 消息 String message = "红色警报！惊险大怪兽！"; // 发送消息 rabbitTemplate.convertAndSend(exchangeName, "red", message); }
5.发布订阅模型----主题
消费者：
/** * * @author dlwlrma * @date 2024/6/18 23:42 * @param msg * TODO: 1.value = @Queue(name = "topic.queue1") 表示绑定的第一个队列 * 2.exchange = @Exchange(name = "dlwlrma.topic", type = ExchangeTypes.TOPIC) 表示交换机名和类型 * 3. key = "china.#" 表示路由key只要以china开始都会接收 */ @RabbitListener(bindings = @QueueBinding( value = @Queue(name = "topic.queue1"), exchange = @Exchange(name = "dlwlrma.topic", type = ExchangeTypes.TOPIC), key = "china.#" )) public void listenTopicQueue1(String msg) { System.out.println("消费者接收到topic.queue1的消息：【" + msg + "】"); } @RabbitListener(bindings = @QueueBinding( value = @Queue(name = "topic.queue2"), exchange = @Exchange(name = "dlwlrma.topic", type = ExchangeTypes.TOPIC), key = "#.news" )) public void listenTopicQueue2(String msg) { System.out.println("消费者接收到topic.queue2的消息：【" + msg + "】"); }
发布者：
/** * * @author dlwlrma * @date 2024/6/17 20:25 * Topic Exchange：主题 */ @Test public void testSendTopicExchange() { // 交换机名称 String exchangeName = "dlwlrma.topic"; // 消息 String message = "喜报！孙悟空大战哥斯拉，胜!"; // 发送消息 rabbitTemplate.convertAndSend(exchangeName, "china.news", message); }
