首页 > 其他分享 >RabbitMQ

RabbitMQ

时间:2022-12-05 23:13:46浏览次数:32  
标签:订阅 队列 factory RabbitMQ 消息 channel

初识MQ

1.同步和异步通讯

微服务间通讯有同步和异步两种方式:

同步通讯:就像打电话,需要实时响应。

异步通讯:就像发邮件,不需要马上回复。

两种方式各有优劣,打电话可以立即得到响应,但是你却不能跟多个人同时通话。发送邮件可以同时与多个人收发邮件,但是往往响应会有延迟。

1.1.同步通讯

我们之前学习的Feign调用就属于同步方式,虽然调用可以实时得到结果,但存在下面的问题:

 

总结:

同步调用的优点:

  • 时效性较强,可以立即得到结果

同步调用的问题:

  • 耦合度高

  • 性能和吞吐能力下降

  • 有额外的资源消耗

  • 有级联失败问题

1.2.异步通讯

异步调用则可以避免上述问题:

 

我们以购买商品为例,用户支付后需要调用订单服务完成订单状态修改,调用物流服务,从仓库分配响应的库存并准备发货。

在事件模式中,支付服务是事件发布者(publisher),在支付完成后只需要发布一个支付成功的事件(event),事件中带上订单id。

订单服务和物流服务是事件订阅者(Consumer),订阅支付成功的事件,监听到事件后完成自己业务即可。

 为了解除事件发布者与订阅者之间的耦合,两者并不是直接通信,而是有一个中间人(Broker)。发布者发布事件到Broker,不关心谁来订阅事件。订阅者从Broker订阅事件,不关心谁发来的消息。

 

Broker 是一个像数据总线一样的东西,所有的服务要接收数据和发送数据都发到这个总线上,这个总线就像协议一样,让服务间的通讯变得标准和可控。

 

好处:

  • 吞吐量提升:无需等待订阅者处理完成,响应更快速

  • 故障隔离:服务没有直接调用,不存在级联失败问题

  • 调用间没有阻塞,不会造成无效的资源占用

  • 耦合度极低,每个服务都可以灵活插拔,可替换

  • 流量削峰:不管发布事件的流量波动多大,都由Broker接收,订阅者可以按照自己的速度去处理事件

缺点:

  • 架构复杂了,业务没有明显的流程线,不好管理

  • 需要依赖于Broker的可靠、安全、性能

好在现在开源软件或云平台上 Broker 的软件是非常成熟的,比较常见的一种就是我们今天要学习的MQ技术。

1.2.技术对比:

MQ,中文是消息队列(MessageQueue),字面来看就是存放消息的队列。也就是事件驱动架构中的Broker。

比较常见的MQ实现:

  • ActiveMQ

  • RabbitMQ

  • RocketMQ

  • Kafka

几种常见MQ的对比:

追求可用性:Kafka、 RocketMQ 、RabbitMQ

追求可靠性:RabbitMQ、RocketMQ

追求吞吐能力:RocketMQ、Kafka

追求消息低延迟:RabbitMQ、Kafka

2.1.安装RabbitMQ

MQ的基本结构:

 

RabbitMQ中的一些角色:

  • publisher:生产者

  • consumer:消费者

  • exchange个:交换机,负责消息路由

  • queue:队列,存储消息

  • virtualHost:虚拟主机,隔离不同租户的exchange、queue、消息的隔离

2.2.RabbitMQ消息模型

RabbitMQ官方提供了5个不同的Demo示例,对应了不同的消息模型:

 

1.基本消息队列

2.工作消息队列

发布订阅模式又分三种

广播:

路由:

主题:

2.4.入门案例

简单队列模式的模型图:

官方的HelloWorld是基于最基础的消息队列模型来实现的,只包括三个角色:

  • publisher:消息发布者,将消息发送到队列queue

  • queue:消息队列,负责接受并缓存消息

  • consumer:订阅队列,处理队列中的消息

2.4.1.publisher实现

思路:

  • 建立连接

  • 创建Channel

  • 声明队列

  • 发送消息

  • 关闭连接和channel

  @Test
    public void testSendMessage() throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数,分别是:主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.150.101");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("itcast");
        factory.setPassword("123321");
        // 1.2.建立连接
        Connection connection = factory.newConnection();

        // 2.创建通道Channel
        Channel channel = connection.createChannel();

        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 4.发送消息
        String message = "hello, rabbitmq!";
        channel.basicPublish("", queueName, null, message.getBytes());
        System.out.println("发送消息成功:【" + message + "】");

        // 5.关闭通道和连接
        channel.close();
        connection.close();

    }

2.4.2.consumer实现

代码思路:

  • 建立连接

  • 创建Channel(这里为什么也创建,因为生产和消费者不一定谁先启动,没有就会报错)

  • 声明队列

  • 订阅消息

 

public static void main(String[] args) throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数,分别是:主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.150.101");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("itcast");
        factory.setPassword("123321");
        // 1.2.建立连接
        Connection connection = factory.newConnection();

        // 2.创建通道Channel
        Channel channel = connection.createChannel();

        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 4.订阅消息
        channel.basicConsume(queueName, true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 5.处理消息
                String message = new String(body);
                System.out.println("接收到消息:【" + message + "】");
            }
        });
        System.out.println("等待接收消息。。。。");
    }

2.5.总结

基本消息队列的消息发送流程:

  1. 建立connection

  2. 创建channel

  3. 利用channel声明队列

  4. 利用channel向队列发送消息

基本消息队列的消息接收流程:

  1. 建立connection

  2. 创建channel

  3. 利用channel声明队列

  4. 定义consumer的消费行为handleDelivery()

  5. 利用channel将消费者与队列绑定

 

标签:订阅,队列,factory,RabbitMQ,消息,channel
From: https://www.cnblogs.com/kisshappyboy/p/16953836.html

相关文章

  • Mac 下erlang及rabbitmq安装
    ---------------------------------------1、首先,安装Homebrew安装Homebrew遇到的问题参考这个链接2、安装erlang环境brewinstallerlang3、安装rabbitmqbrewinstallr......
  • SpringCloud中Rabbitmq
    1.pom配置,添加以来jar包1<dependency>2<groupId>org.springframework.cloud</groupId>3<artifactId>spring-cloud-starter-bus-amqp</artifactId>4</depen......
  • SpringCloud中Rabbitmq的使用
    1.pom配置,添加以来jar包 pom配置2.配置文件配置setting.properties3.rabbitmq消息发送一 发送消息4.rabbitmq消息接收一 接受消息5.rabbitmq发送二 发送mq消......
  • SpringCloud中Rabbitmq的使用
    1#RabbitMq2spring.rabbitmq.host=${dev.spring.rabbitmq.host}3spring.rabbitmq.port=${dev.spring.rabbitmq.port}4spring.rabbitmq.username=${dev.spring.ra......
  • rabbitmqctl set_permissions 报错 invalid command
    执行的命令如下:rabbitmqctlset_permissions-pmyvhostuser1'.*''.*''.*'结果报错:Error:invalidcommand'set_permissions.*.*.*'试了这些命令,都是同样的错......
  • RabbitMQ 使用记录
    RabbitMQ交换机类型RabbitMQ共有四种交换机类型: DirectExchange:直连交换机,根据RoutingKey(路由键)进行投递到不同队列。FanoutExchange:扇形交换机,采用广播模式,根据......
  • SpringCloud(六) - RabbitMQ安装,三种消息发送模式,消息发送确认,消息消费确认(自动,手动)
    1、安装erlang语言环境1.1创建erlang安装目录mkdirerlang1.2上传解压压缩包上传到:/root/解压缩#tar-zxvfotp_src_22.0.tar.gz1.3进入解压缩目录,指定目......
  • RabbitMQ
    RabbitMQ消息队列基础知识消息队列概念MQ(MessageQueue)消息队列,是基础数据结构中先进先出的一种数据机构。指把要传输的数据(消息)放在队列中,用队列机制来实现软件之间的......
  • lavinmq & rabbitmq压测对比
    环境准备docker-compose文件version:'3'services:lavinmq:image:cloudamqp/lavinmq:1.0.0-beta.8volumes:-./mq:/var/li......
  • RabbitMQ 3.7.18安装延时插件rabbitmq-delayed-message-exchange后启动报错
    环境:安装请参考:https://blog.csdn.net/qq_37487520/article/details/126079651https://www.cnblogs.com/myifb/articles/16638251.html启用插件时,不用写插件版本号:r......