AI 助理
备案控制台
开发者社区 数据库 文章 正文

RabbitMQ快速入门 | 帮助快速上手

2023-05-25 122
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
容器服务 Serverless 版 ACK Serverless，952元额度 多规格
容器服务 Serverless 版 ACK Serverless，317元额度 多规格
简介： 🐇🐇🐇前言:RabbitMQ是Apache公司的顶级项目之一, 也是目前各大互联网公司常用的🔥主流MQ🔥之一, 因此在这里向大家介绍RabbitMQ的基础知识和相关应用, 供大家参考学习, 也望大佬给出指点与建议,谢谢大家❤️❤️❤️

♨️本篇文章记录的为RabbitMQ知识中快速入门相关内容，适合在学Java的小白,帮助新手快速上手,也适合复习中，面试中的大佬🙉🙉🙉。
♨️如果文章有什么需要改进的地方还请大佬不吝赐教❤️🧡💛

💖个人主页 : 阿千弟
💖点击这里👉👉👉: RabbitMQ专栏学习

🐇🐇🐇前言:RabbitMQ是Apache公司的顶级项目之一, 也是目前各大互联网公司常用的🔥主流MQ🔥之一, 因此在这里向大家介绍RabbitMQ的基础知识和相关应用, 供大家参考学习, 也望大佬给出指点与建议,谢谢大家❤️❤️❤️
@[TOC]
一 . 消息中间件概述-了解

如果这部分内容您对此有详细的认识,可以跳过

🐇1、 什么是消息中间件

（1）MQ是什么

MQ全称为Message Queue，消息队列是应用程序和应用程序之间的通信方法。先进先出

（2）为什么使用MQ--面试

在项目中，可将一些无需即时返回且耗时的操作提取出来，进行异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的吞吐量。

现在，应用开发和部署---微服务。

（3）MQ优势

a、应用解耦

MQ相当于一个中介，生产方通过MQ与消费方交互，它将应用程序进行解耦合。
在这里插入图片描述

b、异步提速

将不需要同步处理的并且耗时长的操作由消息队列通知消息接收方进行异步处理。提高了应用程序的响应时间。
在这里插入图片描述
order_table status 0 已下单 status 1 支付成功 status 2 已通知商家发货 status 3 商家发货 status 4 已经收货

c、削峰填谷

如订单系统，在下单的时候就会往数据库写数据。但是数据库只能支撑每秒1000左右的并发写入，并发量再高就容易宕机。低峰期的时候并发也就100多个，但是在高峰期时候，并发量会突然激增到5000以上，这个时候数据库肯定卡死了。

消息被MQ保存起来了，然后系统就可以按照自己的消费能力来消费，比如每秒1000个数据，这样慢慢写入数据库，这样就不会卡死数据库了。

在这里插入图片描述
但是使用了MQ之后，限制消费消息的速度为1000，但是这样一来，高峰期产生的数据势必会被积压在MQ中，高峰就被“削”掉了。但是因为消息积压，在高峰期过后的一段时间内，消费消息的速度还是会维持在1000QPS，直到消费完积压的消息,这就叫做“填谷”
在这里插入图片描述

（4）MQ劣势

  • 系统可用性降低

系统引入的外部依赖越多，系统稳定性越差。一旦 MQ 宕机，就会对业务造成影响。如何保证MQ的高可用？

  • 系统复杂度提高

MQ 的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过 MQ 进行异步调用。
如何保证消息没有被重复消费？怎么处理消息丢失情况？那么保证消息传递的顺序性？

  • 一致性问题

A 系统处理完业务，通过 MQ 给B、C、D三个系统发消息数据，如果 B 系统、C 系统处理成功，D 系统处理失败。如何保证消息数据处理的一致性。

最终一致性。

（5）什么时候用MQ

🐇2、AMQP 和 JMS

MQ是消息通信的模型；实现MQ的大致有两种主流方式：AMQP、JMS。

（1）AMQP

AMQP是一种协议，更准确的说是一种binary wire-level protocol（链接协议）。这是其和JMS的本质差别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。

（2）JMS

JMS即Java消息服务（JavaMessage Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。

（3）AMQP 与 JMS 区别

JMS是定义了统一的接口，来对消息操作进行统一；AMQP是通过规定协议来统一数据交互的格式
JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。
JMS规定了两种消息模式；而AMQP的消息模式更加丰富。

🐇3. RabbitMQ

(1）RabbitMQ特点：

1、使用简单，功能强大。
2、基于AMQP协议。 跨语言 c node.js->mq->java python
3、社区活跃，文档完善。
4、高并发性能好，这主要得益于Erlang语言。 c 底层语言，性能强。java 好开发。构建一个web。
5、Spring Boot默认已集成RabbitMQ

(了解即可)
JMS是java提供的一套消息服务API标准，其目的是为所有的java应用程序提供统一的消息通信的标准，类似java的jdbc，只要遵循jms标准的应用程序之间都可以进行消息通信。

它和AMQP有什么 不同，jms是java语言专属的消息服务标准，它是在api层定义标准，并且只能用于java应用；而AMQP是在协议层定义的标准，是跨语言的 。

（2）工作模式

RabbitMQ提供了6种模式：简单模式，work模式，Publish/Subscribe发布与订阅模式，Routing路由模式，Topics主题模式，RPC远程调用模式（远程调用，不太算MQ；暂不作介绍）；

官网对应模式介绍：https://www.rabbitmq.com/getstarted.html

在这里插入图片描述

二 . RabbitMQ工作原理--重点

1. 相关概念介绍

重点看这张图：
在这里插入图片描述
http 协议的 三次握手 四次挥手 比较浪费时间
不如Producer, Consumer 建立的长连接 处理消息的速度

一个消费者监听“一个”“队列”

  • Broker：消息队列服务进程，此进程包括两个部分：Exchange和Queue。
  • Exchange：消息队列交换机，按一定的规则将消息路由转发到某个队列，对消息进行过虑。
  • Queue：消息队列，存储消息的队列，消息到达队列并转发给指定的消费方。
  • Producer：消息生产者，即生产方客户端，生产方客户端将消息发送到MQ。
  • Consumer：消息消费者，即消费方客户端，接收MQ转发的消息。

消息发布接收流程：

-----发送消息-----

1、生产者和Broker建立TCP连接。

2、生产者和Broker建立通道。

3、生产者通过通道消息发送给Broker，由Exchange将消息进行转发。

4、Exchange将消息转发到指定的Queue（队列）

----接收消息-----

1、消费者和Broker建立TCP连接

2、消费者和Broker建立通道

3、消费者监听指定的Queue（队列）

4、当有消息到达Queue时Broker默认将消息推送给消费者。

5、消费者接收到消息。

2、RabbitMQ运转流程

在入门案例中：

生产者发送消息 生产者创建连接（Connection），开启一个信道（Channel），连接到RabbitMQ Broker；
声明队列并设置属性；如是否排它，是否持久化，是否自动删除；
将路由键（空字符串）与队列绑定起来；
发送消息至RabbitMQ Broker；
关闭信道；
关闭连接；

消费者接收消息 消费者创建连接（Connection），开启一个信道（Channel），连接到RabbitMQ Broker 向Broker
请求消费相应队列中的消息，设置相应的回调函数；
等待Broker回应闭关投递响应队列中的消息，消费者接收消息；
确认（ack，自动确认）接收到的消息；
RabbitMQ从队列中删除相应已经被确认的消息；
关闭信道；
关闭连接；

在这里插入图片描述

3、生产者流转过程说明

  1. 客户端与代理服务器Broker建立连接。会调用newConnection() 方法,这个方法会进一步封装Protocol Header 0-9-1 的报文头发送给Broker ，以此通知Broker 本次交互采用的是AMQPO-9-1 协议，紧接着Broker 返回Connection.Start 来建立连接，在连接的过程中涉及Connection.Start/.Start-OK 、Connection.Tune/.Tune-Ok ，Connection.Open/ .Open-Ok 这6 个命令的交互。
  2. 客户端调用connection.createChannel方法。此方法开启信道，其包装的channel.open命令发送给Broker,等待channel.basicPublish方法，对应的AMQP命令为Basic.Publish,这个命令包含了content Header 和content Body()。content Header 包含了消息体的属性，例如:投递模式，优先级等，content Body 包含了消息体本身。
  3. 客户端发送完消息需要关闭资源时，涉及到Channel.Close和Channl.Close-Ok 与Connetion.Close和Connection.Close-Ok的命令交互。

    4、消费者流转过程说明

  4. 消费者客户端与代理服务器Broker建立连接。会调用newConnection() 方法,这个方法会进一步封装Protocol Header 0-9-1 的报文头发送给Broker ，以此通知Broker 本次交互采用的是AMQPO-9-1 协议，紧接着Broker 返回Connection.Start 来建立连接，在连接的过程中涉及Connection.Start/.Start-OK 、Connection.Tune/.Tune-Ok ，Connection.Open/ .Open-Ok 这6 个命令的交互。
  5. 消费者客户端调用connection.createChannel方法。和生产者客户端一样，协议涉及Channel . Open/Open-Ok命令。
  6. 在真正消费之前，消费者客户端需要向Broker 发送Basic.Consume 命令(即调用channel.basicConsume 方法〉将Channel 置为接收模式，之后Broker 回执Basic . Consume - Ok 以告诉消费者客户端准备好消费消息。
  7. Broker 向消费者客户端推送(Push) 消息，即Basic.Deliver 命令，这个命令和Basic.Publish 命令一样会携带Content Header 和Content Body。
  8. 消费者接收到消息并正确消费之后，向Broker 发送确认，即Basic.Ack 命令。
  9. 客户端发送完消息需要关闭资源时，涉及到Channel.Close和Channl.Close-Ok 与Connetion.Close和Connection.Close-Ok的命令交互。

三 . 安装及配置RabbitMQ

更详细安装请点这里

1、下载安装RabbitMQ

​ RabbitMQ由Erlang语言开发，Erlang语言用于并发及分布式系统的开发，在电信领域应用广泛，OTP（Open Telecom Platform）作为Erlang语言的一部分，包含了很多基于Erlang开发的中间件及工具库，安装RabbitMQ需要安装Erlang/OTP，并保持版本匹配，如下图：

RabbitMQ的下载地址：http://www.rabbitmq.com/download.html

在这里插入图片描述

2、下载erlang

安装软件：路径不要有中文和空格。管理员

1. 地址如下：

https://www.erlang.org/downloads

2. erlang安装完成需要配置erlang环境变量：

​ ERLANG_HOME=D:\soft\erl9.3

​ 在path中添加%ERLANG_HOME%\bin

3、安装RabbitMQ

https://github.com/rabbitmq/rabbitmq-server/releases/tag/v3.7.3

或去老师提供的软件包中找到 rabbitmq-server-3.7.3.exe，以管理员方式运行此文件，安装。
在这里插入图片描述

4、启动

安装成功后会自动创建RabbitMQ服务并且启动。

（1）从开始菜单启动RabbitMQ

完成在开始菜单找到RabbitMQ的菜单：
在这里插入图片描述

RabbitMQ Service-install :安装服务

RabbitMQ Service-remove 删除服务

RabbitMQ Service-start 启动

RabbitMQ Service-stop 启动

（2）如果没有开始菜单则进入安装目录下sbin目录手动启动：
在这里插入图片描述
（3）安装并运行服务

rabbitmq-service.bat install 安装服务

rabbitmq-service.bat stop 停止服务

rabbitmq-service.bat start 启动服务

（4）安装管理插件

安装rabbitMQ的管理插件，方便在浏览器端管理RabbitMQ

在sbin目录下，管理员身份运行cmd

rabbitmq-plugins.bat enable rabbitmq_management

5、启动成功 登录RabbitMQ

进入浏览器，输入：http://localhost:15672

在这里插入图片描述

初始账号和密码：guest/guest

在这里插入图片描述

6、注意事项

1、安装erlang和rabbitMQ以管理员身份运行。

2、当卸载重新安装时会出现RabbitMQ服务注册失败，此时需要进入注册表清理erlang

搜索RabbitMQ、ErlSrv，将对应的项全部删除。

🐇四 . RabbitMQ工作模式--重点

🍓1、Work queues工作队列模式（包工头）

（1） 模式说明
在这里插入图片描述
Work Queues与入门程序的简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。
在一个队列中如果有多个消费者，那么消费者之间对于同一个队列消息的关系是竞争的关系。
应用场景：对于 任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

🍓2、publish/subscribe 订阅发布模式类型（微博）

atm取100元，短信，邮件。

订阅模式示例图：

在这里插入图片描述

前面2个案例中，只有3个角色：

  • P：生产者，也就是要发送消息的程序
  • C：消费者：消息的接受者，会一直等待消息到来。
  • queue：消息队列，图中红色部分

而在订阅模型中，多了一个exchange角色，而且过程略有变化：

  • P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
  • C：消费者，消息的接受者，会一直等待消息到来。
  • Queue：消息队列，接收消息、缓存消息。
  • Exchange：交换机，图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。
  • Exchange有常见以下3种类型：
    • Fanout：广播，将消息交给所有绑定到交换机的队列
    • Direct：定向，把消息交给符合指定routing key 的队列
    • Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列
    • Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

发布订阅模式：

1、每个消费者监听自己的队列。
2、生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收 到消息

代码

生产者

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class Producer_pubsub {
   
   
    //交换机名称
    static final String FANOUT_EXCHAGE = "fanout_exchange";
    //队列名称
    static final String FANOUT_QUEUE_1 = "fanout_queue_1";
    //队列名称
    static final String FANOUT_QUEUE_2 = "fanout_queue_2";

    public static void main(String[] args) throws Exception {
   
   

        //1创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //主机地址;默认为 localhost
        connectionFactory.setHost("localhost");
        //连接端口;默认为 5672
        connectionFactory.setPort(5672);
        //虚拟主机名称;默认为 /
        connectionFactory.setVirtualHost("/");
        //连接用户名；默认为guest
        connectionFactory.setUsername("guest");
        //连接密码；默认为guest
        connectionFactory.setPassword("guest");

        //2创建连接
        Connection connection = connectionFactory.newConnection();

        //3创建频道
        Channel channel = connection.createChannel();

        /**
         * 声明交换机
         * 参数1：交换机名称
         * 参数2：交换机类型，fanout、topic、direct、headers
         * 参数3：是否定义持久化
         * 参数4：是否在不使用的时候自动删除
         */
        channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT,true,true,null);

        // 声明（创建）队列
        /**
         * 参数1：队列名称
         * 参数2：是否定义持久化队列
         * 参数3：是否独占本次连接
         * 参数4：是否在不使用的时候自动删除队列
         * 参数5：队列其它参数
         */
        channel.queueDeclare(FANOUT_QUEUE_1, true, false, false, null);
        channel.queueDeclare(FANOUT_QUEUE_2, true, false, false, null);

        //队列绑定交换机
        channel.queueBind(FANOUT_QUEUE_1, FANOUT_EXCHAGE, "");
        channel.queueBind(FANOUT_QUEUE_2, FANOUT_EXCHAGE, "");

        for (int i = 1; i <= 10; i++) {
   
   
            // 发送信息
            String message = "你好；小兔子！发布订阅模式--" + i;
            /**
             * 参数1：交换机名称，如果没有指定则使用默认Default Exchage
             * 参数2：路由key,简单模式可以传递队列名称
             * 参数3：消息其它属性
             * 参数4：消息内容
             */
            channel.basicPublish(FANOUT_EXCHAGE, "", null, message.getBytes());
            System.out.println("已发送消息：" + message);
        }

        // 关闭资源
        channel.close();
        connection.close();
    }
}

消费者1

import com.rabbitmq.client.*;

import java.io.IOException;

public class Consumer1 {
   
   
    //交换机名称
    static final String FANOUT_EXCHAGE = "fanout_exchange";
    //队列名称
    static final String FANOUT_QUEUE_1 = "fanout_queue_1";
    //队列名称
    static final String FANOUT_QUEUE_2 = "fanout_queue_2";

    public static void main(String[] args) throws Exception {
   
   
        //1创建连接工厂
        ConnectionFactory connectionFactory=new ConnectionFactory();
        //连接的ip
        connectionFactory.setHost("localhost");
        //连接的端口
        connectionFactory.setPort(5672);
        //设置虚拟主机
        connectionFactory.setVirtualHost("/");
        //设置用户名
        connectionFactory.setUsername("guest");
        //设置密码
        connectionFactory.setPassword("guest");

        //2创建长连接
        Connection connection = connectionFactory.newConnection();
        //3创建channel
        Channel channel = connection.createChannel();

        //声明队列
        //String queue,  队列名
        // boolean durable, 持久化
        // boolean exclusive, 排他的
        // boolean autoDelete, 自动删除
        // Map<String, Object> arguments 属性
        channel.queueDeclare(FANOUT_QUEUE_1,true,false,false,null);
        channel.queueDeclare(FANOUT_QUEUE_2,true,false,false,null);

        // 声明交换机
        // String exchange,  交换机名称
        // BuiltinExchangeType type, 交换机类型
        // boolean durable,  持久化
        // boolean autoDelete, 自动删除
        // Map<String, Object> arguments 属性
        channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT,true,false,null);

        //队列绑定交换机
        // String queue, 队列名称
        // String exchange, 交换机名称
        // String routingKey 路由键
        channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"");
        channel.queueBind(FANOUT_QUEUE_2,FANOUT_EXCHAGE,"");


        //4监听某个队列
        // String queue, 监听的队列名
        // boolean autoAck, 是否自动应答
        // Consumer callback 回调函数，收到消息，我要干啥
        com.rabbitmq.client.Consumer consumer=new DefaultConsumer(channel){
   
   
            // 回调函数，收到消息，我要干啥
            //  String consumerTag, 消费者标签
            // Envelope envelope, 信封 保存很多信息
            // AMQP.BasicProperties properties, 属性
            // byte[] body  消息字节数组
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
   
   
                //业务逻辑

                //现在的业务逻辑就是打印
//                System.out.println("consumerTag:"+consumerTag);
//                System.out.println("Exchange:"+envelope.getExchange());
//                System.out.println("RoutingKey:"+envelope.getRoutingKey());
//                System.out.println("DeliveryTag:"+envelope.getDeliveryTag()); //消息id

                System.out.println(new String(body));
            }
        };
        channel.basicConsume(FANOUT_QUEUE_1,true,consumer);

        //5 千万别关闭连接，要不然queue有了消息 推不过来了
//        channel.close();
//        connection.close();
    }
}

消费者2

import com.rabbitmq.client.*;

import java.io.IOException;


public class Consumer2 {
   
   
    //交换机名称
    static final String FANOUT_EXCHAGE = "fanout_exchange";
    //队列名称
    static final String FANOUT_QUEUE_1 = "fanout_queue_1";
    //队列名称
    static final String FANOUT_QUEUE_2 = "fanout_queue_2";

    public static void main(String[] args) throws Exception {
   
   
        //1创建连接工厂
        ConnectionFactory connectionFactory=new ConnectionFactory();
        //连接的ip
        connectionFactory.setHost("localhost");
        //连接的端口
        connectionFactory.setPort(5672);
        //设置虚拟主机
        connectionFactory.setVirtualHost("/");
        //设置用户名
        connectionFactory.setUsername("guest");
        //设置密码
        connectionFactory.setPassword("guest");

        //2创建长连接
        Connection connection = connectionFactory.newConnection();
        //3创建channel
        Channel channel = connection.createChannel();

        //声明队列
        //String queue,  队列名
        // boolean durable, 持久化
        // boolean exclusive, 排他的
        // boolean autoDelete, 自动删除
        // Map<String, Object> arguments 属性
        channel.queueDeclare(FANOUT_QUEUE_1,true,false,false,null);
        channel.queueDeclare(FANOUT_QUEUE_2,true,false,false,null);

        // 声明交换机
        // String exchange,  交换机名称
        // BuiltinExchangeType type, 交换机类型
        // boolean durable,  持久化
        // boolean autoDelete, 自动删除
        // Map<String, Object> arguments 属性
        channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT,true,false,null);

        //队列绑定交换机
        // String queue, 队列名称
        // String exchange, 交换机名称
        // String routingKey 路由键
        channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"");
        channel.queueBind(FANOUT_QUEUE_2,FANOUT_EXCHAGE,"");


        //4监听某个队列
        // String queue, 监听的队列名
        // boolean autoAck, 是否自动应答
        // Consumer callback 回调函数，收到消息，我要干啥
        Consumer consumer=new DefaultConsumer(channel){
   
   
            // 回调函数，收到消息，我要干啥
            //  String consumerTag, 消费者标签
            // Envelope envelope, 信封 保存很多信息
            // AMQP.BasicProperties properties, 属性
            // byte[] body  消息字节数组
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
   
   
                //业务逻辑

                //现在的业务逻辑就是打印
//                System.out.println("consumerTag:"+consumerTag);
//                System.out.println("Exchange:"+envelope.getExchange());
//                System.out.println("RoutingKey:"+envelope.getRoutingKey());
//                System.out.println("DeliveryTag:"+envelope.getDeliveryTag()); //消息id

                System.out.println(new String(body));
            }
        };
        channel.basicConsume(FANOUT_QUEUE_2,true,consumer);

        //5 千万别关闭连接，要不然queue有了消息 推不过来了
//        channel.close();
//        connection.close();

    }
}

3） 测试
启动所有消费者，然后使用生产者发送消息；在每个消费者对应的控制台可以查看到生产者发送的所有消息；到达广播的效果。

在执行完测试代码后，其实到RabbitMQ的管理后台找到Exchanges选项卡，点击 fanout_exchange 的交换机，可以查看到如下的绑定：

在这里插入图片描述
测试发送一条消息：

两个消费者都接受到同样的消息
在这里插入图片描述
在这里插入图片描述

小结:

交换机需要与队列进行绑定，绑定之后；一个消息可以被多个消费者都收到。

发布订阅模式与工作队列模式的区别

1、工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机。

2、发布/订阅模式的生产方是面向交换机发送消息，工作队列模式的生产方是面向队列发送消息(底层使用默认交换机)。

3、发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队列绑 定到默认的交换机 。

🍓3、Routing路由模式（分布式日志收集系统）

路由模式特点：

  1. 队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
  2. 消息的发送方在 向 Exchange发送消息时，也必须指定消息的 RoutingKey。

  3. Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息
    在这里插入图片描述

    图解：

P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列
C1：消费者，其所在队列指定了需要routing key 为 error 的消息
C2：消费者，其所在队列指定了需要routing key为 info、error、warning 的消息

代码
在编码上与 Publish/Subscribe发布与订阅模式 的区别是交换机的类型为：Direct，还有队列绑定交换机的时候需要指定routing key。

a、生产者

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;


public class Producer {
   
   

    //交换机名称
    static final String DIRECT_EXCHAGE = "direct_exchange";
    //队列名称
    static final String DIRECT_QUEUE_1 = "direct_queue_1";
    //队列名称
    static final String DIRECT_QUEUE_2 = "direct_queue_2";

    public static void main(String[] args) throws Exception {
   
   
        //1创建连接工厂
        ConnectionFactory connectionFactory=new ConnectionFactory();
        //连接的ip
        connectionFactory.setHost("localhost");
        //连接的端口
        connectionFactory.setPort(5672);
        //设置虚拟主机
        connectionFactory.setVirtualHost("/");
        //设置用户名
        connectionFactory.setUsername("guest");
        //设置密码
        connectionFactory.setPassword("guest");

        //2创建长连接
        Connection connection = connectionFactory.newConnection();
        //3创建channel
        Channel channel = connection.createChannel();
        //声明队列
        //String queue,  队列名
        // boolean durable, 持久化
        // boolean exclusive, 排他的
        // boolean autoDelete, 自动删除
        // Map<String, Object> arguments 属性
        channel.queueDeclare(DIRECT_QUEUE_1,true,false,false,null);
        channel.queueDeclare(DIRECT_QUEUE_2,true,false,false,null);

        // 声明交换机
        // String exchange,  交换机名称
        // BuiltinExchangeType type, 交换机类型
        // boolean durable,  持久化
        // boolean autoDelete, 自动删除
        // Map<String, Object> arguments 属性
        channel.exchangeDeclare(DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT,true,false,null);

        //队列绑定交换机
        // String queue, 队列名称
        // String exchange, 交换机名称
        // String routingKey 路由键
        channel.queueBind(DIRECT_QUEUE_1,DIRECT_EXCHAGE,"error");
        channel.queueBind(DIRECT_QUEUE_2,DIRECT_EXCHAGE,"info");
        channel.queueBind(DIRECT_QUEUE_2,DIRECT_EXCHAGE,"error");
        channel.queueBind(DIRECT_QUEUE_2,DIRECT_EXCHAGE,"warning");

        //4发消息
        // String exchange,  交换机
        // String routingKey, 路由键
        // AMQP.BasicProperties props, 属性
        // byte[] body 消息      string byte[] char[]如何相互转换的？
        String msg="hello rabbitmq!routing error";
        channel.basicPublish(DIRECT_EXCHAGE,"error",null,msg.getBytes());

        String msg1="hello rabbitmq!routing info";
        channel.basicPublish(DIRECT_EXCHAGE,"info",null,msg1.getBytes());

        String msg2="hello rabbitmq!routing warning";
        channel.basicPublish(DIRECT_EXCHAGE,"warning",null,msg2.getBytes());

        //5关闭连接  资源关闭的顺序，先关后出来的资源，最后关，第一个资源
        channel.close();
        connection.close();
    }
}

b、消费者1

import com.rabbitmq.client.*;

import java.io.IOException;


public class Consumer1 {
   
   
    //交换机名称
    static final String DIRECT_EXCHAGE = "direct_exchange";
    //队列名称
    static final String DIRECT_QUEUE_1 = "direct_queue_1";
    //队列名称
    static final String DIRECT_QUEUE_2 = "direct_queue_2";

    public static void main(String[] args) throws Exception {
   
   
        //1创建连接工厂
        ConnectionFactory connectionFactory=new ConnectionFactory();
        //连接的ip
        connectionFactory.setHost("localhost");
        //连接的端口
        connectionFactory.setPort(5672);
        //设置虚拟主机
        connectionFactory.setVirtualHost("/");
        //设置用户名
        connectionFactory.setUsername("guest");
        //设置密码
        connectionFactory.setPassword("guest");

        //2创建长连接
        Connection connection = connectionFactory.newConnection();
        //3创建channel
        Channel channel = connection.createChannel();

        //声明队列
        //String queue,  队列名
        // boolean durable, 持久化
        // boolean exclusive, 排他的
        // boolean autoDelete, 自动删除
        // Map<String, Object> arguments 属性
        channel.queueDeclare(DIRECT_QUEUE_1,true,false,false,null);
        channel.queueDeclare(DIRECT_QUEUE_2,true,false,false,null);

        // 声明交换机
        // String exchange,  交换机名称
        // BuiltinExchangeType type, 交换机类型
        // boolean durable,  持久化
        // boolean autoDelete, 自动删除
        // Map<String, Object> arguments 属性
        channel.exchangeDeclare(DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT,true,false,null);

        //队列绑定交换机
        // String queue, 队列名称
        // String exchange, 交换机名称
        // String routingKey 路由键
        channel.queueBind(DIRECT_QUEUE_1,DIRECT_EXCHAGE,"error");
        channel.queueBind(DIRECT_QUEUE_2,DIRECT_EXCHAGE,"info");
        channel.queueBind(DIRECT_QUEUE_2,DIRECT_EXCHAGE,"error");
        channel.queueBind(DIRECT_QUEUE_2,DIRECT_EXCHAGE,"warning");


        //4监听某个队列
        // String queue, 监听的队列名
        // boolean autoAck, 是否自动应答
        // Consumer callback 回调函数，收到消息，我要干啥
        com.rabbitmq.client.Consumer consumer=new DefaultConsumer(channel){
   
   
            // 回调函数，收到消息，我要干啥
            //  String consumerTag, 消费者标签
            // Envelope envelope, 信封 保存很多信息
            // AMQP.BasicProperties properties, 属性
            // byte[] body  消息字节数组
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
   
   
                //业务逻辑

                //现在的业务逻辑就是打印
//                System.out.println("consumerTag:"+consumerTag);
//                System.out.println("Exchange:"+envelope.getExchange());
//                System.out.println("RoutingKey:"+envelope.getRoutingKey());
//                System.out.println("DeliveryTag:"+envelope.getDeliveryTag()); //消息id

                System.out.println(new String(body));
            }
        };
        channel.basicConsume(DIRECT_QUEUE_1,true,consumer);

        //5 千万别关闭连接，要不然queue有了消息 推不过来了
//        channel.close();
//        connection.close();

    }
}

c、消费者2

import com.rabbitmq.client.*;

import java.io.IOException;

public class Consumer2 {
   
   
    //交换机名称
    static final String DIRECT_EXCHAGE = "direct_exchange";
    //队列名称
    static final String DIRECT_QUEUE_1 = "direct_queue_1";
    //队列名称
    static final String DIRECT_QUEUE_2 = "direct_queue_2";

    public static void main(String[] args) throws Exception {
   
   
        //1创建连接工厂
        ConnectionFactory connectionFactory=new ConnectionFactory();
        //连接的ip
        connectionFactory.setHost("localhost");
        //连接的端口
        connectionFactory.setPort(5672);
        //设置虚拟主机
        connectionFactory.setVirtualHost("/");
        //设置用户名
        connectionFactory.setUsername("guest");
        //设置密码
        connectionFactory.setPassword("guest");

        //2创建长连接
        Connection connection = connectionFactory.newConnection();
        //3创建channel
        Channel channel = connection.createChannel();

        //声明队列
        //String queue,  队列名
        // boolean durable, 持久化
        // boolean exclusive, 排他的
        // boolean autoDelete, 自动删除
        // Map<String, Object> arguments 属性
        channel.queueDeclare(DIRECT_QUEUE_1,true,false,false,null);
        channel.queueDeclare(DIRECT_QUEUE_2,true,false,false,null);

        // 声明交换机
        // String exchange,  交换机名称
        // BuiltinExchangeType type, 交换机类型
        // boolean durable,  持久化
        // boolean autoDelete, 自动删除
        // Map<String, Object> arguments 属性
        channel.exchangeDeclare(DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT,true,false,null);

        //队列绑定交换机
        // String queue, 队列名称
        // String exchange, 交换机名称
        // String routingKey 路由键
        channel.queueBind(DIRECT_QUEUE_1,DIRECT_EXCHAGE,"error");
        channel.queueBind(DIRECT_QUEUE_2,DIRECT_EXCHAGE,"info");
        channel.queueBind(DIRECT_QUEUE_2,DIRECT_EXCHAGE,"error");
        channel.queueBind(DIRECT_QUEUE_2,DIRECT_EXCHAGE,"warning");


        //4监听某个队列
        // String queue, 监听的队列名
        // boolean autoAck, 是否自动应答
        // Consumer callback 回调函数，收到消息，我要干啥
        Consumer consumer=new DefaultConsumer(channel){
   
   
            // 回调函数，收到消息，我要干啥
            //  String consumerTag, 消费者标签
            // Envelope envelope, 信封 保存很多信息
            // AMQP.BasicProperties properties, 属性
            // byte[] body  消息字节数组
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
   
   
                //业务逻辑

                //现在的业务逻辑就是打印
//                System.out.println("consumerTag:"+consumerTag);
//                System.out.println("Exchange:"+envelope.getExchange());
//                System.out.println("RoutingKey:"+envelope.getRoutingKey());
//                System.out.println("DeliveryTag:"+envelope.getDeliveryTag()); //消息id

                System.out.println(new String(body));
            }
        };
        channel.basicConsume(DIRECT_QUEUE_2,true,consumer);

        //5 千万别关闭连接，要不然queue有了消息 推不过来了
//        channel.close();
//        connection.close();
    }
}

测试
启动所有消费者，然后使用生产者发送消息；在消费者对应的控制台可以查看到生产者发送对应routing key对应队列的消息；到达按照需要接收的效果。

在执行完测试代码后，其实到RabbitMQ的管理后台找到Exchanges选项卡，点击 direct_exchange 的交换机，可以查看到如下的绑定：

在这里插入图片描述
测试：

第一个消费者受到一条消息

在这里插入图片描述
第二个消费者受到三条消息

在这里插入图片描述

小结
Routing模式要求队列在绑定交换机时要指定routing key，消息会转发到符合routing key的队列。

🍓4、Topics通配符模式

Topic类型与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！

Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

通配符规则：

#：匹配一个或多个词

：*匹配不多不少恰好1个词

在这里插入图片描述
在这里插入图片描述
图解：

红色Queue：绑定的是usa.# ，因此凡是以 usa.开头的routing key 都会被匹配到
黄色Queue：绑定的是#.news ，因此凡是以 .news结尾的 routing key 都会被匹配

代码

生产者

public class Send {
   
   
    private final static String EXCHANGE_NAME = "test_topic_exchange";

    public static void main(String[] argv) throws Exception {
   
   
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明exchange，指定类型为topic
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
        // 消息内容
        String message = "这是一只行动迅速的橙色的兔子";
        // 发送消息，并且指定routing key为：quick.orange.rabbit
        channel.basicPublish(EXCHANGE_NAME, "quick.orange.rabbit", null, message.getBytes());
        System.out.println(" [动物描述：] Sent '" + message + "'");

        channel.close();
        connection.close();
    }
}

消费者1

public class Recv {
   
   
    private final static String QUEUE_NAME = "topic_exchange_queue_Q1";
    private final static String EXCHANGE_NAME = "test_topic_exchange";

    public static void main(String[] argv) throws Exception {
   
   
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 绑定队列到交换机，同时指定需要订阅的routing key。订阅所有的橙色动物
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.orange.*");

        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
   
   
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
   
   
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者1] received : " + msg + "!");
            }
        };
        // 监听队列，自动ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

消费者2

public class Recv2 {
   
   
    private final static String QUEUE_NAME = "topic_exchange_queue_Q2";
    private final static String EXCHANGE_NAME = "test_topic_exchange";

    public static void main(String[] argv) throws Exception {
   
   
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 绑定队列到交换机，同时指定需要订阅的routing key。订阅关于兔子以及懒惰动物的消息
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.*.rabbit");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "lazy.＃");

        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
   
   
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
   
   
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者2] received : " + msg + "!");
            }
        };
        // 监听队列，自动ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

结果C1、C2是都接收到消息了：

在这里插入图片描述
在这里插入图片描述

🍓5、 模式总结

RabbitMQ工作模式：

（1）简单模式 HelloWorld

一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）。

（2）工作队列模式 Work Queue

一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）。

（3）发布订阅模式 Publish/subscribe

需要设置类型为fanout的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列。

（4）路由模式 Routing

需要设置类型为direct的交换机，交换机和队列进行绑定，并且指定routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列。

（5）通配符模式 Topic

需要设置类型为topic的交换机，交换机和队列进行绑定，并且指定通配符方式的routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列。

在这里插入图片描述

小结:
好了,简单的RabbitMQ快速入门我们就先说到这里了
接下来请持续关注的springboot整合RabbitMQ高级应用的相关内容

如果这篇【文章】有帮助到你💖，希望可以给我点个赞👍，创作不易，如果有对Java后端或者对redis感兴趣的朋友,请多多关注💖💖💖
💖个人主页
💖阿千弟

文章标签：
云消息队列 MQ
容器服务Kubernetes版
Java
消息中间件
NoSQL
网络协议
Apache
API
数据库
缓存
存储
数据库连接
关键词：
云消息队列 MQ快速入门
相关实践学习
消息队列RocketMQ版：基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能，涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列：消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验：基于MNS，0基础轻松构建 Web Client 本节课带您一起基于MNS，0基础轻松构建 Web Client
阿千弟
目录
相关文章
Cool架构
|
6月前
|
消息中间件 Java 测试技术
快速入门RabbitMQ并且加入项目实战-kuai-su-ru-men-rabbitmq-bing-qie-jia-ru-xiang-mu-shi-zhan（二）
快速入门RabbitMQ并且加入项目实战-kuai-su-ru-men-rabbitmq-bing-qie-jia-ru-xiang-mu-shi-zhan
Cool架构
131 1
浩泽学编程
|
6月前
|
消息中间件 运维 Java
RabbitMQ快速入门（简单收发消息）
RabbitMQ快速入门（简单收发消息）
浩泽学编程
101 0
RabbitMQ快速入门（简单收发消息）
请回答1024
|
6月前
|
消息中间件 存储 Java
RabbitMQ-同步和异步区别&快速入门
RabbitMQ-同步和异步区别&快速入门
请回答1024
194 1
Cool架构
|
6月前
|
消息中间件 Java Docker
快速入门RabbitMQ并且加入项目实战-kuai-su-ru-men-rabbitmq-bing-qie-jia-ru-xiang-mu-shi-zhan（一）
快速入门RabbitMQ并且加入项目实战-kuai-su-ru-men-rabbitmq-bing-qie-jia-ru-xiang-mu-shi-zhan
Cool架构
80 1
陶然同学
|
6月前
|
消息中间件 Java Kafka
【RabbitMQ】RabbitMQ快速入门 通俗易懂 初学者入门
【RabbitMQ】RabbitMQ快速入门 通俗易懂 初学者入门
陶然同学
142 0
赵广陆
|
消息中间件 SQL 负载均衡
RocketMQ快速入门 2
RocketMQ快速入门
赵广陆
198 0
赵广陆
|
消息中间件 Java Apache
RocketMQ快速入门 1
RocketMQ快速入门
赵广陆
172 0
赵广陆
|
消息中间件 Java Spring
RabbitMQ快速入门 3
RabbitMQ快速入门
赵广陆
52 0
赵广陆
|
消息中间件 存储 缓存
RabbitMQ快速入门 2
RabbitMQ快速入门
赵广陆
49 0
赵广陆
|
消息中间件 缓存 Java
RabbitMQ快速入门 1
RabbitMQ快速入门
赵广陆
155 0

热门文章

最新文章

  • 1
    信用算力基于 RocketMQ 实现金融级数据服务的实践
  • 2
    MQTT的心跳保活机制是通过`setKeepAliveInterval()`方法设置的
  • 3
    消息中间件的技术选型心得－RabbitMQ、ActiveMQ和ZeroMQ
  • 4
    【Spring常见错误】Rocketmq 闪退
  • 5
    解决微信小程序MQTT真机连接问题与合法域名配置SSL问题
  • 6
    RocketMQ 消息发送system busy、broker busy原因分析与解决方案
  • 7
    rabbitmq消息件在idc机房监控中的应用
  • 8
    RabbitMQ入门指南(十一)：延迟消息-延迟消息插件
  • 9
    STM32通过esp8266连接WiFi接入MQTT服务器
  • 10
    RabbitMQ队列中unacked消息持续时间很久
  • 1
    RabbitMQ的springboot项目集成使用-01
    143
  • 2
    深入理解MQ消息队列的高可用与可靠性策略
    1340
  • 3
    rabbitmq插件升级-延时队列为列
    664
  • 4
    rabbitmq安装erlang环境后没生效
    1023
  • 5
    Springboot 集成Rabbitmq之延时队列
    186
  • 6
    RabbitMQ高级
    45
  • 7
    rabbitMQ集群
    51
  • 8
    rabbitmq与kafka的区别
    91
  • 9
    涉及rocketMQ，jemeter等性能测试服务器的安装记录
    72
  • 10
    用docker方式 安装rabbitmq 并配置MQTT
    295

    • 相关课程

    更多
  • 消息队列 RocketMQ 消息集成
  • 消息队列 MNS 入门课程
  • RocketMQ知识精讲与项目实战（第一阶段）
  • RocketMQ知识精讲与项目实战（第二阶段）
  • RocketMQ知识精讲与项目实战（第三阶段）
  • 微服务实战-RocketMQ Binder

    • 相关电子书

    更多
  • RocketMQ Client-GO 介绍
  • RocketMQ Prometheus Exporter 打造定制化 DevOps 平台
  • 基于 RocketMQ Prometheus Exporter 打造定制化 DevOps 平台

    • 相关实验场景

    更多
  • 快速体验云消息队列RocketMQ版
  • 快速入门轻量消息队列（原MNS）
  • 快速学会使用云消息队列RabbitMQ版
  • 部署RabbitMQ
  • Rocketmq-spring入门与实践
  • RocketMQ一站式入门使用
    • 下一篇
    无影云桌面