深入了解RabbitMQ

OnMyHeart 2020-05-26

AMQP

相关概念介绍

  1. AMQP是一个提供统一消息服务的应用层标准高级消息队列协议, 是应用层协议的一个开放标准，为面向消息的中间件设计.
  2. AMQP是一个二进制协议, 具有多信道, 协商式, 异步, 安全, 多平台, 高效等优点.
  3. RabbitMQ是AMQP协议的Erlang实现.
  4. 相关概念介绍
    • Connection: 连接, 一个网络连接, 如tcp/ip套接字连接.
    • Session: 会话, 端点之间的命名对话. 在一个会话上下文中, 保证“恰好传递一次”.
    • Channel: 信道, 多路复用连接中的一条全双工数据流通道, 为会话提供传输介质.
      • 深入了解RabbitMQ
    • Client: 客户端, AMQP连接或会话的发起者, AMQP是非对称的, 客户端生成和消费消息, 服务器存储和路由这些信息.
    • Broker: 服务节点, 一般情况下可以将一个RabbitMQ Broker看作一RabbitMQ服务器
    • 端点: AMQP对话的任意一方, 如一个AMQP连接包括两个端点(客户端/服务器)
    • Consumer: 消费者, 一个从消息队列里请求消息的客户端程序.
    • Producer: 生产者, 一个向交换机发布消息的客户端应用程序.

RabbitMQ运转流程

  • 在简单模式中
    • 深入了解RabbitMQ
  1. 生产者发送消息
    • 生产者创建连接, 开启一个信道, 连接到RabbitMQ Broker.
    • 声明队列并设置属性: 如是否持久化, 是否独占信道, 是否自动删除.
    • 将路由键(空串)与队列绑定.
    • 发送消息至RabbitMQ Broker
    • 关闭信道, 关闭连接.
  2. 消费者接收消息
    • 消费者创建连接, 开启一个信道, 连接到RabbitMQ Broker.
    • 向Broker请求消费队列中的消息, 设置相应的回调函数.
    • 等待Broker回应闭关投递响应队列中的消息, 消费者接收消息.
    • 自动确认接收到的消息.
    • RabbitMQ从队列中删除已被确认的消息.
    • 关闭信道, 关闭连接.

生产者流转过程

  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这六个命令的交互.
  2. 客户端调用connection.createChannel()方法.
    • 该方法开启信道, 其包装的channel.open命令发给Broker
  3. channel.basicPublish方法发消息, 其对应的AMQP命令为Basic.Publish, 该命令包含了contentHeader和contentBody
      • contentHeader包含消息体的属性, 如投递模式, 优先级.
      • contentBody包含了消息体内容.
  4. 客户端发送完消息要关闭资源的时候, 涉及Channel.Close和Channel.Close-OK, 以及Connection.Close,Connection.Close-OK命令的交互.

深入了解RabbitMQ

 消费者流转过程说明

  1. 消费者客户端与代理服务器Broker建立连接.
    • 过程与生产者建立连接一样.
  2. 消费者客户端调用connection.createChannel方法.
  3. 在真正消费之前, 消费者客户端要向Broker发送Basic.Consume命令, 将Channel置为接受模式, 之后Broker回执Basic.Consume-OK命令告诉消费者已经准备好了消费消息.
  4. Broker向消费者客户端推送消息, 即Basic.Deliver命令, 它一样会携带contentHeader和contentBody.
  5. 消费者接收到消息并正确消费后, 会向Broker发送确认, 即Basic.Ack命令.
  6. 客户端接受完消息关闭资源.

深入了解RabbitMQ

 RabbitMQ的工作模式

Work queues工作队列模式

  1. 工作模式说明
    • 和简单模式相比, 多了一个或一些消费者, 多个消费端共同消费同一个队列中的信息.
    • 深入了解RabbitMQ
  2. 应用场景: 对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度.
  3. 多个消费者之间采取轮询的方式去消费.
  4. 代码和简单模式几乎一样, 只是多了几个消费者客户端.

订阅模式类型

  • 模式示例图
    • 深入了解RabbitMQ
  1. 前面两种模式, 只有三个角色, 消费者, 生产者, 消息队列. 
  2. 而在订阅模式中, 多了一个exchange交换机角色, 并且过程略有变化.
    • P: 生产者, 发送消息, 但不再发给消息队列, 而是发给X(交换机)
    • C: 消费者, 消息的接收者, 会一直等待消息到来.
    • Queue: 消息队列, 接收消息, 缓存消息.
    • Exchange: 交换机, 一方面, 会接收生产者发送的消息, 另一方面, 会处理生产者发出的消息, 如递交给某个特定队列, 或是丢弃某消息. 到底如何操作取决于Exchange的类型(常见3种)
      • Fanout: 广播, 把消息交给所有绑定到交换机的队列.
      • Direct: 定向, 把消息交给符合指定routing key的队列.
      • Topic: 通配符, 把消息交给符合routing pattern(路由模式)的队列.
  3. 注意: 交换机只负责转发消息, 不具备储存消息的能力.

Publish/Subscribe发布/订阅模式

  • 模式说明
    1. 每个消费者监听自己的队列
    2. 生产者将消息发给broker, 由交换机将消息转发到绑定此交换机的每个队列, 每个与交换机绑定的队列都能接收到消息.
  • 生产者代码
    • 代码
      /*
发布/订阅使用的交换机类型为: fanout
 */
public class Producer {

    //交换机名称
    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 IOException, TimeoutException {
        //创建连接
        Connection connection = ConnectionUtil.getConnection();

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

        /**
         * 声明交换机
         * 参数1：交换机名称
         * 参数2：交换机类型，fanout、topic、direct、headers(不常用)
         */
        channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);

        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 = 0; i < 10; i++) {
            String message = "RabbitMQ" + Math.random();
            /**
             * 参数1：交换机名称，如果没有指定则使用默认Default Exchage
             * 参数2：路由key,简单模式可以传递队列名称
             * 参数3：消息其它属性
             * 参数4：消息内容
             */
            channel.basicPublish(FANOUT_EXCHAGE, "", null, message.getBytes());
        }
        System.out.println("已发送消息");

        //关闭资源
        channel.close();
        connection.close();
    }
}
    • 如图所示, 两个消息队列均有生产者发出的消息.
    • 深入了解RabbitMQ
  • 消费者代码
    public class Consumer1 {

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtil.getConnection();

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

        //声明交换机
        channel.exchangeDeclare(Producer.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);

        //声明队列1
        channel.queueDeclare(Producer.FANOUT_QUEUE_1, true, false, false, null);

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

        //创建消费者；并设置消息处理
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            /**
             * consumerTag 消息者标签，在channel.basicConsume时候可以指定
             * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                System.out.println("消费者1-接收到的消息为：" + new String(body));
            }
        };

        //监听消息
        /**
         * 参数1：队列名称
         * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
         * 参数3：消息接收到后回调
         */
        channel.basicConsume(Producer.FANOUT_QUEUE_1, true, consumer);
    }
}


public class Consumer2 {

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtil.getConnection();

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

        //声明交换机
        channel.exchangeDeclare(Producer.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);

        //声明队列2
        channel.queueDeclare(Producer.FANOUT_QUEUE_2, true, false, false, null);

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

        //创建消费者；并设置消息处理
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                System.out.println("消费者2-接收到的消息为：" + new String(body));
            }
        };

        //监听消息
        channel.basicConsume(Producer.FANOUT_QUEUE_2, true, consumer);
    }
}
  • 小结
    1. 交换机需要与队列绑定, 绑定后, 一个消息可被多个消费者都收到.
    2. 发布/订阅模式与工作队列模式的区别
      • 工作队列模式不用定义交换机, 而发布/订阅模式需要定义交换机.
      • 发布/订阅模式的生产方是面向交换机发送消息, 工作队列模式的生产方是面向队列发送消息(底层使用默认交换机).
      • 发布/订阅模式需要设置队列和交换机的绑定, 工作队列模式不需要设置, 实际上工作队列模式会将队列绑定到默认的交换机.

Routing路由模式

  • 路由模式特点
    • 队列与交换机的绑定, 不是任意绑定了, 而要指定一个RoutingKey.
    • 消息的发送方在向Exchange发送消息时, 也必须指定消息的RoutingKey.
    • Exchange不再把消息交给每一个绑定的队列, 而是根据消息的RoutingKey进行判断, 只有队列的RoutingKey与消息的RoutingKey完全一致, 才会接受到消息.
  • 图解
    • 深入了解RabbitMQ
    • P: 生产者, 在向Exchange发消息时, 会指定一个routing key.
    • X: 交换机, 接收生产者的消息, 然后把消息递交给与routing key完全匹配的队列.
    • C1: 消费者, 其所在队列指定了需要routing key为error的消息.
    • C2: 消费者, 其所在队列指定了需要routing key为info, error, warning的消息.
  1. 生产者代码
    • 代码
      /**
 * 路由模式的交换机类型为：direct
 */
public class Producer {

    //交换机名称
    static final String DIRECT_EXCHAGE = "direct_exchange";
    //队列名称
    static final String DIRECT_QUEUE_INSERT = "direct_queue_insert";
    //队列名称
    static final String DIRECT_QUEUE_UPDATE = "direct_queue_update";

    public static void main(String[] args) throws IOException, TimeoutException {
        //创建连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();
        //声明交换机
        channel.exchangeDeclare(DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
        //声明队列
        channel.queueDeclare(DIRECT_QUEUE_INSERT, true, false, false, null);
        channel.queueDeclare(DIRECT_QUEUE_UPDATE, true, false, false, null);
        //队列绑定交换机
        /**
         * 参数1: 队列名
         * 参数2: 交换机名
         * 参数3: Routing Key
         */
        channel.queueBind(DIRECT_QUEUE_INSERT, DIRECT_EXCHAGE, "insert");
        channel.queueBind(DIRECT_QUEUE_UPDATE, DIRECT_EXCHAGE, "update");

        // 发送信息
        String message = "routing key为insert";
        /**
         * 参数1：交换机名称，如果没有指定则使用默认Default Exchage
         * 参数2：路由key,简单模式可以传递队列名称
         * 参数3：消息其它属性
         * 参数4：消息内容
         */
        channel.basicPublish(DIRECT_EXCHAGE, "insert", null, message.getBytes());
        System.out.println("已发送消息: " + message);

        // 发送信息
        message = "routing key为update";
        channel.basicPublish(DIRECT_EXCHAGE, "update", null, message.getBytes());
        System.out.println("已发送消息: " + message);

        // 关闭资源
        channel.close();
        connection.close();
    }
}
    • 如图, 消息队列只会收到特定的消息
    • 深入了解RabbitMQ
  2. 消费者代码
    public class Consumer1 {

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();
        //声明交换机
        channel.exchangeDeclare(Producer.DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);

        //创建队列
        channel.queueDeclare(Producer.DIRECT_QUEUE_INSERT, true, false, false, null);

        //队列绑定交换机
        channel.queueBind(Producer.DIRECT_QUEUE_INSERT, Producer.DIRECT_EXCHAGE, "insert");

        //创建消费者, 并设置消息处理
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println(new String(body, "utf-8"));
            }
        };

        channel.basicConsume(Producer.DIRECT_QUEUE_INSERT, true, consumer);
    }
}


public class Consumer2 {

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(Producer.DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
        channel.queueDeclare(Producer.DIRECT_QUEUE_UPDATE, true, false, false, null);
        channel.queueBind(Producer.DIRECT_QUEUE_UPDATE, Producer.DIRECT_EXCHAGE, "update");
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println(new String(body, "utf-8"));
            }
        };
        channel.basicConsume(Producer.DIRECT_QUEUE_UPDATE, true, consumer);
    }
}
  • 小结: Routing模式要求生产者指定一个RoutingKey给交换机, 消息队列绑定交换机时也指定RoutingKey, 这样消息会转发到符合routingKey的队列.

Topics通配符模式

  • 模式说明: Topic类型和Direct相比, 都可以根据RoutingKey把消息路由到不同的队列, 但Topic类型的交换机可以让队列在绑定RoutingKey的时候使用通配符.
  1. RoutingKey一般都是由一个或多个单词组成, 多个单词之间由"."分隔, 如item.insert
  2. 通配符规则
    • #: 匹配一个或多个词.
    • *: 只能匹配一个词.
      item.* 只能匹配item.insert
item.# 能匹配item.insert, item.insert.abc
  3. 图解
    • 深入了解RabbitMQ
    • 红色Queue, 绑定的是usa.#, 所以凡是以usa.开头的RoutingKey, 都会被匹配到.
    • 黄色的Queue: 绑定的是#.news, 所以凡是以.news结尾的RoutingKey, 都会被匹配的.
  4. 生产者代码
    public class Producer {

    //交换机名称
    static final String TOPIC_EXCHAGE = "topic_exchange";
    //队列名称
    static final String TOPIC_QUEUE_1 = "topic_queue_1";
    //队列名称
    static final String TOPIC_QUEUE_2 = "topic_queue_2";

    public static void main(String[] args) throws IOException, TimeoutException {
        //创建连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();
        //声明交换机
        channel.exchangeDeclare(TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);

        //发送信息
        String message = "Topic: routing key为item.insert";
        channel.basicPublish(TOPIC_EXCHAGE, "item.insert", null, message.getBytes());


        message = "Topic: routing key为item.update";
        channel.basicPublish(TOPIC_EXCHAGE, "item.update", null, message.getBytes());

        message = "Topic: routing key为item.delete";
        channel.basicPublish(TOPIC_EXCHAGE, "item.delete", null, message.getBytes());

        channel.close();
        connection.close();
    }
}
  5. 消费者代码
    public class Consumer1 {

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtil.getConnection();

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

        //声明交换机
        channel.exchangeDeclare(Producer.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);

        channel.queueDeclare(Producer.TOPIC_QUEUE_1, true, false, false, null);

        //队列绑定交换机
        channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.update");
        channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.delete");

        //创建消费者；并设置消息处理
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println(new String(body, "utf-8"));
            }
        };

        //监听消息
        channel.basicConsume(Producer.TOPIC_QUEUE_1, true, consumer);
    }
}

public class Consumer2 {

    public static void main(String[] args) throws IOException, TimeoutException {

        Connection connection = ConnectionUtil.getConnection();

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

        //声明交换机
        channel.exchangeDeclare(Producer.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);

        channel.queueDeclare(Producer.TOPIC_QUEUE_2, true, false, false, null);

        //队列绑定交换机
        channel.queueBind(Producer.TOPIC_QUEUE_2, Producer.TOPIC_EXCHAGE, "item.*");

        //创建消费者；并设置消息处理
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println(envelope.getRoutingKey());
                System.out.println(envelope.getExchange());
                System.out.println(envelope.getDeliveryTag());
                System.out.println(new String(body, "utf-8"));
            }
        };

        //监听消息
        channel.basicConsume(Producer.TOPIC_QUEUE_2, true, consumer);
    }
}
  • 小结: Topic相比起Routing模式, 多了通配符, 使用上更灵活, 若不用通配符, 则和Routing路由模式一样.

模式总结

  1. 简单模式: 一个生产者, 一个消费者, 不用自己设置交换机(用默认交换机)
  2. 工作队列模式: 一个生产者, 多个消费者(轮询, 竞争), 不用自己设置交换机(用默认交换机)
  3. 发布/订阅模式: 需要设置类型为fanout的交换机, 并且交换机和队列进行绑定，当发送消息到交换机后, 交换机会将消息发送到绑定的队列.
  4. 路由模式: 需要设置类型为direct的交换机, 交换机和队列进行绑定, 并且指定routing key, 当发送消息到交换机后, 交换机会根据routing key将消息发送到对应的队列.
  5. 通配符模式: 需要设置类型为topic的交换机, 交换机和队列进行绑定, 并且指定通配符方式的routing key, 当发送消息到交换机后, 交换机会根据routing key将消息发送到对应的队列.
OnMyHeart

OnMyHeart

相关推荐

RabbitMQ如何保证消息的可靠投递？

String message = "this is info message " + i;autoAck=false: RabbitMQ会等待消费者显示回复确认消息后才从内存中移出消息。deliveryTag: 用来标识信道中投递的消息

zhuxue / 0评论 2020-10-14

RabbitMQ七战Kafka，差异立现！

作为一个有丰富经验的微服务系统架构师，经常有人问我，“应该选择RabbitMQ还是Kafka？”。基于某些原因， 许多开发者会把这两种技术当做等价的来看待。的确，在一些案例场景下选择RabbitMQ还是Kafka没什么差别，但是这两种技术在底层实现方面是有

shenzhenzsw / 0评论 2020-10-09

RabbitMQ是如何确定消息是否投递到队列中的

在使用RabbitMQ消息中间件时，因为消息的投递是异步的，默认情况下，RabbitMQ会删除那些无法路由的消息。为了能够检出消息是否顺利投递到队列，我们需要相应的处理机制。今天就来验证一下相关的验证机制。RabbitMQ消息会先到达指定的交换机，然后由交

shyoldboy / 0评论 2020-09-27

c# rabbitmq 简单收发消息的示例代码

//设置prefetchCount : 1来告知RabbitMQ，在未收到消费端的消息确认时，不再分发消息，也就确保了当消费端处于忙碌状态时，不再分配任务。channel.BasicAck;// 消息确认信号，确认后该消息将从队列里移除

leihui00 / 0评论 2020-09-16

Docker 安装Redis、MySql、ELK、RabbitMq

docker run -d --restart always --name kibana --net somenetwork -p 5601:5601 kibana:7.1.1

lishijian / 0评论 2020-08-17

《RabbitMQ》如何保证消息不被重复消费

消息重复的原因有两个：1.生产时消息重复，2.消费时消息重复。由于生产者发送消息给MQ，在MQ确认的时候出现了网络波动，生产者没有收到确认，实际上MQ已经接收到了消息。生产者中如果消息未被确认，或确认失败，我们可以使用定时任务+来进行消息重试。@Sched

程序员伊成 / 0评论 2020-08-06

RabbitMQ都写了，RocketMQ怎么能落下？

整体架构最近看到了我在Github上写的rabbitmq-examples陆续被人star了，就想着写个rocketmq-examples。对rabbitmq感兴趣的小伙伴可以看我之前的文章。下面把RocketMQ的各个特性简单介绍一下，这样在用的时候心里

ljcsdn / 0评论 2020-07-27

RabbitMQ —— 介绍

AMQP，即 AMQP，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开发标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端中间件不同产品，不同的开发语言等条件的限制，Erlang 中的实现有 Rab

waitzkj / 0评论 2020-07-25

mq如何处理消息丢失

　　生产者将数据发送到rabbitmq的时候，可能因为网络问题导致数据就在半路给搞丢了。但是问题是，开始rabbitmq事务机制，基本上吞吐量会下来，因为太耗性能。如果rabbitmq没能处理这个消息，会回调你一个nack接口，告诉你这个消息接收失败，你可

powrexly / 0评论 2020-07-20

ubuntu下快速安装rabbitmq

安装rabbitmqsudo apt-get updatesudo apt-get upgradesudo apt-get install rabbitmq-server. 当我们在创建用户时，会指定用户能访问一个虚拟机，并且该用户只能访问该虚拟机下的队列

liym / 0评论 2020-07-20

深入浅出 RabbitMQ

基于 ErLang 语言开发有高可用高并发的优点，适合集群。有消息确认机制和持久化机制，可靠性高。消息队列提供了 FIFO 的处理机制，具有缓存消息的能力。在 RabbitMQ 中，队列消息可以设置为持久化，临时或者自动删除。如果是临时的队列，Queue

zhoucheng0 / 0评论 2020-07-19

Docker-compose常用软件部署(mysql、redis、rabbitmq)

废话不多说，下面就直接给出部署文件已经启动脚本，这个会持续更新。以下的部署方式均是单例部署。高可用或者其他部署方案我会单独写相应的文章。不指定则会启动失败。filterName=`docker network ls | grep $network_name

shenzhenzsw / 0评论 2020-07-18

Rabbitmq的使用四_Java Client方式使用Topic模式

topic交换机发送给队列的消息，通过routingkey进行匹配。　　* 号只能代替一个单词。　　# 号可以替换零个或多个单词。在这个例子中，我们将发送所有描述动物的信息。消息将与一个routingkey一起发送，routingkey由三个单词(两个点)

woaishanguosha / 0评论 2020-07-18

rabbitmq 延时队列

‘queueName‘ => ‘test_cache_queue‘."_".$time,‘routeKey‘ => ‘test_cache_route‘."_".$time,‘host‘ => ‘1

waitzkj / 0评论 2020-07-18

SpringBoot与RabbitMQ整合

消息代理和目的地。当消息发送者发送消息以后，将由消息代理接管，消息代理保证消息传递到指定目的地。–消息只有唯一的发送者和接受者，但并不是说只能有一个接收者。AMQP定义了wire-level层的协议标准；天然具有跨平台、跨语言特性。绑定，用于消息队列和交换

zhoucheng0 / 0评论 2020-07-08

RabbitMQ使用

那就表示一个工作队列，而工作队列的最大特征在于：若干个消费者一起完成工作。

waitzkj / 0评论 2020-07-05

RabbitMQ 3.8.5版本 windows安装和web管理界面登陆

RabbitMQ默认安装路径为：C:\Program Files\RabbitMQ Server\rabbitmq_server-3.8.5

json0000 / 0评论 2020-07-04

rabbitmq单机入门安装教程(ubuntu环境 亲测有效)

第一步:使用xz -d rabbitmq-server-generic-unix-3.7.26.tar.xz 转换为tar格式的文件。第二步:使用 tar -xvf rabbitmq-server-generic-unix-3.7.26.tar 解压文件。

NVEFLY / 0评论 2020-07-04

springcloud-消息总线Bus(九)

SpringCloud Bus配合Config使用可以实现配置的动态刷新.Bus是用来将分布式系统的节点与轻量级消息系统链接起来的框架, 它整合了Java的事件处理机制和消息中间件的功能.Bus能管理和传播分布式系统间的消息, 就像一个分布式执行器, 可用

OnMyHeart / 0评论 2020-07-04

Rabbitmq安装

RabbitMQ在安装好后，可以访问http://ip地址:15672 ；其自带了guest/guest的用户名和密码；如果需要创建自定义用户；那么也可以登录管理界面后，如下操作：。可登陆管理控制台, 同时可以对policy进行管理。仅可登陆管理控制台，无

OnMyHeart / 0评论 2020-06-28

springboot + rabbitmq 做智能家居，我也没想到会这么简单

前一段有幸参与到一个智能家居项目的开发，由于之前都没有过这方面的开发经验，所以对智能硬件的开发模式和技术栈都颇为好奇。用户还可能向报警器发一些关闭报警、调整音量的指令等。技术选型用rabbitmq 来做核心的组件，主要考虑到运维成本低，组内成员使用的熟练度

lishijian / 0评论 2020-06-22

Rabbitmq应用场景

1） 对于数据量大或者处理耗时长的操作，我们可以引入 MQ 实现异步通信，减少客户端的等待，提升响应速度。2） 对于改动影响大的系统之间，可以引入 MQ 实现解耦，减少系统之间的直接依赖。1）系统可用性降低：原来是两个节点的通信，现在还需要独立运行一个服务

shenzhenzsw / 0评论 2020-06-21

RabbitMQ 第六课 springboot整合rabbitmq

-- 导入配置文件处理器，配置文件进行绑定就会有提示 -->

cj0 / 0评论 2020-06-21

springboot + rabbitmq 做智能家居，我也没想到会这么简单

前一段有幸参与到一个智能家居项目的开发，由于之前都没有过这方面的开发经验，所以对智能硬件的开发模式和技术栈都颇为好奇。用户还可能向报警器发一些关闭报警、调整音量的指令等。技术选型用rabbitmq 来做核心的组件，主要考虑到运维成本低，组内成员使用的熟练度

cj0 / 0评论 2020-06-21

RabbitMQ存储模型

了解一些 RabbitMQ 的实现原理也是很有必要的，它可以让你在遇到问题时能透过现象看本质。比如一个队列的内部存 储其实是由5个子队列来流转运作的，队列中的消息可以有4种不同的状态等，通过这些可以明白在使用 RabbitMQ 时尽量不要有过多的消息堆积，

OnMyHeart / 0评论 2020-06-20

celery异步发送邮箱

它是消费-生产者模型的一个典型的代表，producer往消息队列中不断写入消息，而另一端consumer则可以读取或者订阅队列中的消息。业务上，可以实现服务提供者和消费者之间的数据解耦，提供高可用性的消息传输机制，在实际生产中应用相当广泛。AMQP，即Ad

waitzkj / 0评论 2020-06-20

rabbitmq-3.5.x 安装与运行

## 1. 首先在Linux上进行一些软件的准备工作，yum下来一些基础的软件包yum install build-essential openssl openssl-devel unixODBC unixODBC-devel make gcc gcc-c

OnMyHeart / 0评论 2020-06-17

非常强悍的 RabbitMQ 总结，写得真好！

RabbitMQ是基于AMQP协议的，通过使用通用协议就可以做到在不同语言之间传递。channel：网络信道，几乎所有操作都在channel中进行，channel是消息读写的通道。客户端可以建立多个channel，每个channel表示一个会话任务。mes

OnMyHeart / 0评论 2020-06-16

RabbitMQ常见疑难问题

一.消息队列消息堆积如何处理？当消息产生的速度长时间远远大于消费的速度的时候，就会产生消息的堆积。如果消费成功且ack成功，则删除记录的消息标记。如果ack失败，消息下次被消费消息时候，先去查询消息的消费标识，已经消费则直接ack，未消费则继续消费。

liym / 0评论 2020-06-14

centos上安装RabbitMQ

yum -y install make gcc gcc-c++ kernel-devel m4 ncurses-devel openssl-devel unixODBC-devel. 查看所有用户rabbitmqctl list_users添加一个用户ra

waitzkj / 0评论 2020-06-09

rabbitMQ第四种模型(Routing)

在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某种场景下，我们希望不同的消息被不同的队列消费。这是就要用到Direct类型的Exchange。消息的发送方在 向Exchange发送消息时，也必须指定消息的RoutingKey。Excha

cj0 / 0评论 2020-06-07

RabbitMQ

Kafka追求高吞吐量和性能，但是不支持事务，以及对消息的重复和丢失没有严格要求，一开始主要用于日志收集和传输，适合大量数据收集。RabbitMQ使用Erlang语言开发，基于AMQP协议实现，AMQP主要特征是面向消息、队列、可靠性、安全。

Soongp / 0评论 2020-06-07

RabbitMQ入门研究

将不需要同步处理的并且耗时长的操作由消息队列通知消息接收方进行异步处理。

liym / 0评论 2020-06-07

rabbitMQ的第三种模型(fanout)

public static void main(String[] args) throws IOException {. public static void main(String[] args) throws IOException {. channe

waitzkj / 0评论 2020-06-06

在Linux中安装RabbitMQ并访问

操作系统版本为CentOS7，纯净未安装RabbitMQ，以下是以阿里云服务器为载体。epel-release is needed by erlang-solutions-1.0-1.noarch，依赖检测失败错误，则进行下一步）。将用户放到adminis

OnMyHeart / 0评论 2020-06-06

RabbitMQ 第二课 快速入门

　　RabbitMQ是开源的消息中间件，它是轻量级的，支持多种消息传递协议，可以部署在分布式和联合配置中，以满足高级别、高可用性需求。并且可在许多操作系统和云环境上运行，并为大多数流行语言提供了广泛的开发工具。　　1、异步处理，主要为了较少请求的响应时间和

cj0 / 0评论 2020-06-06

RabbitMQ(二) RabbitMQ高级特性

消费端实现幂等性，就意味着，我们的消息永远不会被消费多次，即时收到多条一样的消息。幂等性，通俗点说，就一个数据，或者一个请求，给你重复来多次，你得确保对应的数据是不会改变的，不能出错。生产者进行接收应该，用来确定这条消息是否正常的发送到broker，这种方

waitzkj / 0评论 2020-06-04

jenkins + gitlab + redis + rabbitmq 安装记录

firewall-cmd --zone=public --add-port=80/tcp --permanent. firewall-cmd --zone=public --add-port=443/tcp --permanent. firewall-cm

shenzhenzsw / 0评论 2020-06-03

RabbitMQ安装管理简要命令

在 RabbitMQ 中这两个是分开的，互换使用是错误的。简单来说，鉴权是验证用户是谁，而授权是决定用户允许和不被允许做的。rabbitmqctl set_permissions [-p <vhostpath>] <user> &l

shenzhenzsw / 0评论 2020-06-03

laravel 下 rabbitmq必看

2,变更了queueTest 需要更新。3, 查看队列执行情况。4,将数据存入队列

OnMyHeart / 0评论 2020-06-02

rabbitmq的基本使用

使用MQ的三大作用：1、同步变异步2、流量削峰3、解耦降低服务间的耦合性要不要使用MQ，需不需要使用MQ依据项目的需要做选择。监控应用中抛出的异常，邮件通知管理员。也就是增加监控功能，且一个消息服务可以监控无数个应用。例如，微信的抢红包，淘宝的订单，铁道部

cj0 / 0评论 2020-06-01

rabbitmq

rabbitmqctl set_permissions -p VHostPath User ConfP WriteP ReadP

waitzkj / 0评论 2020-05-31

二、RabbitMQ客户端开发向导

一旦consume以后，channel会被置为接收模式，直到取消队列的订阅为止，这期间RabbitMQ会不断地推消息给消费者。如果只获取单条消息，建议使用拉的模式，但是持续消费的时候需要使用推的模式来提高吞吐量，用循环拉的模式会影响RabbitMQ的性能。

OnMyHeart / 0评论 2020-05-27

thinkphp使用AMQP库(支持RabbitMq)

// Send a message with the string "quit" to cancel the consumer.

OnMyHeart / 0评论 2020-05-27

[SpringBoot] Spring Boot（14）RabbitMQ延迟队列

　　　　1）通过消息过期后进入死信交换器，再由交换器转发到延迟消费队列，实现延迟功能；　　　　2）使用rabbitmq-delayed-message-exchange插件实现延迟功能；注意： 延迟插件rabbitmq-delayed-message-ex

liym / 0评论 2020-05-17

消息中间件(九)-----RabbitMq消息可靠传递

* @param durable true if we are declaring a durable exchange . * @param autoDelete true if the server should delete the exchange

Soongp / 0评论 2020-05-17

删除rabbitmq的队列和队列中的数据

登录后访问http://{rabbitmq安装IP}:15672/#/queues，这里可以看到你创建的所有的Queue，选中某一个Queue，下方有个Delete/Purge，展开，选择purge即可。

xx0cw / 0评论 2020-05-16

rabbitmq延迟消息

使用RabbitMQ来实现延迟消息必须先了解RabbitMQ的两个概念：消息的TTL和死信Exchange，通过这两者的组合来实现上述需求。对队列设置就是队列没有消费者连着的保留时间，也可以对每一个单独的消息做单独的设置。Dead Letter Excha

liym / 0评论 2020-05-16

CentOS在线安装RabbitMQ3.7

Error: Package: rabbitmq-server-3.6.1-1.noarch Requires: erlang >= R16B-3You could try using --skip-broken to work around the

87447007 / 0评论 2020-05-16