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的对比：

追求可用性(当需要处理数据时，资源处于可用状态的程度)：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);
    }