RabbitMQ 入门

一：什么是MQ

二：安装RabbitMQ

三：在Java中如何实现MQ的使用

RabbitMQ的五种消息模型

1.基本消息队列（BasicQueue）

2.工作消息队列（WorkQueue）

3. 发布订阅（Publish、Subscribe）

4.Direct Exchange：路由

5.Topic Exchange：主题

在Java中实现五种模型

1.基本消息队列

2.工作消息队列

3.发布订阅模型--广播

4.发布订阅模型---路由

5.发布订阅模型----主题

一：什么是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自动将接收的消息给方法参数msgpublic 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模板对象@Autowiredprivate RabbitTemplate rabbitTemplate;/**** @author dlwlrma* @date 2024/6/17 20:23* TODO： Basic Queue 简单队列模型*/@Testpublic 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 向队列中不停发送消息，模拟消息堆积。*/@Testpublic 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.定义交换机的名字@Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange("dlwlrma.fanout");}//2.定义队列1@Beanpublic Queue fanoutQueue1(){return new Queue("fanout.queue1");}//3.将交换机与队列绑定起来@Beanpublic Binding bindingExchangeToFanoutQueue1(FanoutExchange fanoutExchange,Queue fanoutQueue1){//将队列1与交换机绑定起来return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}//4.定义队列2@Beanpublic Queue fanoutQueue2(){return new Queue("fanout.queue2");}//3.将交换机与队列绑定起来@Beanpublic 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：广播*/@Testpublic 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：路由*/@Testpublic 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：主题*/@Testpublic void testSendTopicExchange() {// 交换机名称String exchangeName = "dlwlrma.topic";// 消息String message = "喜报！孙悟空大战哥斯拉，胜!";// 发送消息rabbitTemplate.convertAndSend(exchangeName, "china.news", message);}