RabbitMQ的基本介绍及与Spring整合

一，场景回顾

? 最近做电商购物项目，在分布式中搜索服务，商品详情服务都是独立的模块。那么有一个问题就是：

  • 商品的原始数据保存在数据库中，增删改查都在数据库中完成。
  • 搜索服务数据来源是索引库，如果数据库商品发生变化，索引库数据不能及时更新。
  • 商品详情做了页面静态化，静态页面数据也不会随着数据库商品发生变化。

? 如果我们在后台修改了商品的价格，搜索页面和商品详情页显示的依然是旧的价格，这样显然不对。该如何解决？

? 通常有两种解决方案：

  • 方案1：每当后台对商品做增删改操作，同时要修改索引库数据及静态页面
  • 方案2：搜索服务和商品页面服务对外提供操作接口，后台在商品增删改后，调用接口

? 以上两种方式都有同一个严重问题：就是代码耦合，后台服务中需要嵌入搜索和商品页面服务，违背了微服务的独立原则。

? 所以，我们会通过另外一种方式来解决这个问题：消息队列

二，消息队列

2.1，什么是消息队列

? 首先我们应该知道队列（Queue）是一种数据结构，一种特殊的线性表。消息队列（Message Queue）是应用程序之间相互通信的方法，我们可以将消息队列比作为一个容器。提供消息的一方将消息存放到消息队列中，而要接收消息的一方同样去消息队列中去获取，因而消息队列是分布式系统中一个重要的中间组件。使用消息队列异步处理系统的性能，降低系统的耦合性，实现高可用。

? 消息队列其实是一种典型的生产者，消费者模型。生产者不断向消息队列中生产消息，消费者不断从消息队列中获取消息，两者之间都是通过异步进行的。因此只关心消息的生产和消费，两者之间没有业务代码的侵入，实现了生产者和消费者之间的解耦。

结合场景回顾所说的问题：

  • 商品服务对商品增删改以后，无需去操作索引库或静态页面，只是发送一条消息，也不关心消息被谁接收。
  • 搜索服务和静态页面服务接收消息，分别去处理索引库和静态页面。

如果以后有其它系统也依赖商品服务的数据，同样监听消息即可，商品服务无需任何代码修改。

2.2，AMQP和JMS

? MQ是消息通信的模型，并不是具体实现。现在实现MQ的有两种主流方式：AMQP、JMS。

2.2.1，AMQP

? AMQP（advanced message queuing protocol）在2003年时被提出，最早用于解决金融领不同平台之间的消息传递交互问题。顾名思义，AMQP是一种协议，更准确的说是一种binary wire-level protocol（链接协议）。这是其和JMS的本质差别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。这使得实现了AMQP的provider天然性就是跨平台的。意味着我们可以使用Java的AMQP provider，同时使用一个python的producer加一个rubby的consumer。从这一点看，AQMP可以用http来进行类比，不关心实现的语言，只要大家都按照相应的数据格式去发送报文请求，不同语言的client均可以和不同语言的server链接。

2.2.2， JMS

? 通常而言提到JMS（Java Message Service）实际上是指JMS API。JMS是由Sun公司早期提出的消息标准，旨在为java应用提供统一的消息操作，包括create，send、receive等。JMS已经成为Java Enterprise Edition的一部分。从使用角度看，JMS和JDBC担任差不多的角色，用户都是根据相应的接口可以和实现了JMS的服务进行通信，进行相关的操作。

? 两者之间的区别：

AMQP JMS
平台 多技术平台 针对JAVA
消息模型 direct，topic，fanout，headers 点对点，发布订阅模型
消息数据类型 二进制消息 StreamMessage，MapMessage，TextMessage，ObjectMessage和BytesMessage
消息路由 routing key 使用客户端的选择过滤器实现路由

2.3，常见的MQ产品

  • ActiveMQ：基于JMS
  • RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好
  • RocketMQ：基于JMS，阿里巴巴产品，目前交由Apache基金会
  • Kafka：分布式消息系统，高吞吐量

三，RabbitMQ

? RabbitMQ是一个开源的AMQP实现的消息管理系统，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面非常优秀。

? 官网地址：https://www.rabbitmq.com/

? 在使用RabbitMQ之前应将环境准备好，该文章中不讲述安装步骤，网上有很多优秀的博客，大家可以去参考以下。

? RabbitMQ提供6种消息模型，但是最后一种为RPC远程调用，所以只有５种。

? 使用之前导入RabbitMQ依赖，后面会介绍与spring boot的整合。

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

? 自定义一个连接工具类，方便多程序使用。

public class Rec {
    private static final String QUEUE_NAME = "simple_queue";
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1，获取连接
        Connection connection = ConnectUtil.getConnection();
        // 2，创建通道
        Channel channel = connection.createChannel();
        // 3，声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 4，声明消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {// 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                       byte[] body) {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [x] received : " + msg + "!");
            }
        };

        // 5，监听队列
        channel.basicConsume(QUEUE_NAME,true,consumer);
    }
}

3.1，基本消息模型（simple）

? 基本模型介绍：

? 文中指出，我们可以将RabbitMQ比作一个邮局，发件人将信封放到邮箱，此时邮差先生拿到你的邮件根据上面地址可以邮递到指定的收件人种。在这个过程中RabbitMQ充当邮局，邮差，邮箱三种角色。

? 基本模型如图：

? P（producer/ publisher）：生产者，一个发送消息的用户应用程序。

? C（consumer）：消费者，用于接收并处理逻辑的应用程序。

? 队列（红色区域）：rabbitmq内部类似于邮箱的一个概念。虽然消息流经rabbitmq和你的应用程序，但是它们只能存储在队列中。队列只受主机的内存和磁盘限制，实质上是一个大的消息缓冲区。许多生产者可以发送消息到一个队列，许多消费者可以尝试从一个队列接收数据。

? 接下来通过代码来实现。

? 发送者：

public class send {

    private static final String QUEUE_NAME = "simple_queue";

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

        // 1，获取连接
        Connection connection = ConnectionUtil.getConnection();
        // 2，创建通道
        Channel channel = connection.createChannel();
        // 3，声明队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        // 4，定义消息
        String msg = "Hello world";
        // 5，发送消息
        channel.basicPublish(null,QUEUE_NAME,null,msg.getBytes());
        System.out.println(" [x] Sent '" + msg + "'");

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

? 结果如下：

消费者1：

public class Rec {

    private static final String QUEUE_NAME = "simple_queue";

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

        // 1，获取连接
        Connection connection = ConnectionUtil.getConnection();
        // 2，创建通道
        Channel channel = connection.createChannel();
        // 3，声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 4，声明消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {// 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                       byte[] body) {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [x] received : " + msg + "!");
            }
        };
        // 5，监听队列
        channel.basicConsume(QUEUE_NAME,true,consumer);
    }
}

? 接收者结果如图：

3.1.1，消息确认机制（ACK）--经典面试问题

通过刚才的案例可以看出，消息一旦被消费者接收，队列中的消息就会被删除。

那么问题来了：RabbitMQ怎么知道消息被接收了呢？

如果消费者领取消息后，还没执行操作就挂掉了呢？或者抛出了异常？消息消费失败，但是RabbitMQ无从得知，这样消息就丢失了！

因此，RabbitMQ有一个ACK机制。当消费者获取消息后，会向RabbitMQ发送回执ACK，告知消息已经被接收。不过这种回执ACK分两种情况：

  • 自动ACK：消息一旦被接收，消费者自动发送ACK
  • 手动ACK：消息接收后，不会发送ACK，需要手动调用

那么该如何选择呢？？

这需要看消息的重要性：

  • 如果消息不太重要，丢失也没有影响，那么自动ACK会比较方便
  • 如果消息非常重要，不容丢失。那么最好在消费完成后手动ACK，否则接收消息后就自动ACK，RabbitMQ就会把消息从队列中删除。如果此时消费者宕机，那么消息就丢失了。

上图中有一行代码为：

其中第二个参数就是是否使用ACK机制，true代表使用，false代表不使用。

源码如图：

? 但是需要注意的是，手动ACK存在一定的问题。如果消息接收方已经接收但是在业务处理过程中出现异常，那么该消息并未处理完成。但此时消息队列中消息已经被自动删除，因而就会造成消息丢失，这是个非常严重的问题。

? 解决方法：将autoAck改为false，并且添加如下代码

 // 手动进行ACK
channel.basicAck(envelope.getDeliveryTag(), false);

? 当ACK机制改为手动时，只有当消息真正消费结束时，通知消息队列并删除消息。

3.2 ，Work消息模型

?

? 介绍：

? 工作队列，又称任务队列。主要思想就是避免执行资源密集型任务时，必须等待它执行完成。相反我们稍后完成任务，我们将任务封装为消息并将其发送到队列。 在后台运行的工作进程将获取任务并最终执行作业。当你运行许多工人时，任务将在他们之间共享，但是一个消息只能被一个消费者获取。

? 这个概念在Web应用程序中特别有用，因为在短的HTTP请求窗口中无法处理复杂的任务。

? 接下来我们来模拟这个流程：

? P：生产者：任务的发布者

? C1：消费者，领取任务并且完成任务，假设完成速度较快

C2：消费者2：领取任务并完成任务，假设完成速度慢

? 接下来生产者发送50条数据：

public class Send {
    private final static String QUEUE_NAME = "Work_queue";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 循环发布任务
        for (int i = 0; i < 50; i++) {
            // 消息内容
            String message = "task .. " + i;
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");

            Thread.sleep(i * 2);
        }
        // 关闭通道和连接
        channel.close();
        connection.close();
    }
}

? 消费者1：该消费者模拟性能较差的情况下

public class Recv {
    private final static String QUEUE_NAME = "Work_queue";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        final Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 设置每个消费者同时只能处理一条消息
        channel.basicQos(1);
        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者1] received : " + msg + "!");
                try {
                    // 模拟完成任务的耗时：1000ms
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                }
                // 手动ACK
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        // 监听队列。
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

? 消费者2：

public class Recv2 {
    private final static String QUEUE_NAME = "Work_queue";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        final Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 设置每个消费者同时只能处理一条消息
        channel.basicQos(1);
        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println(" [消费者2] received : " + msg + "!");
                // 手动ACK
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        // 监听队列。
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

? 结果如下图：

? 从图中看出发送者50条消息被两个消费者各消费25条，以此可以提高效率，实现任务分配。

? ==经典面试题：如何避免消息堆积？==

? 1） 采用workqueue，多个消费者监听同一队列。

? 2）接收到消息以后，通过线程池，异步消费。

3.2.1，能者多劳

但是刚才的实现有一个问题：

  • 消费者1比消费者2的效率要低，一次任务的耗时较长
  • 然而两人最终消费的消息数量是一样的
  • 消费者2大量时间处于空闲状态，消费者1一直忙碌

现在的状态属于是把任务平均分配，正确的做法应该是消费越快的人，消费的越多。

? 我们可以使用basicQos方法和prefetchCount = 1设置。 这告诉RabbitMQ一次不要向消费者发送多于一条消息。 或者换句话说，不要向消费者发送新消息，直到它处理并确认了前一个消息。 相反，==它会将其分派给不是仍然忙碌的下一个工消费者。==

// 设置每个消费者同时只能处理一条消息
channel.basicQos(1);

再次运行结果：

3.3，订阅者消息模型（Publish/Subscribe）

详解：

1、1个生产者，多个消费者

2、每一个消费者都有自己的一个队列

3、生产者没有将消息直接发送到队列，而是发送到了交换机

4、每个队列都要绑定到交换机

5、生产者发送的消息，经过交换机到达队列，实现一个消息被多个消费者获取的目的

X（Exchanges）：交换机一方面：接收生产者发送的消息。另一方面：知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。

Exchange类型有以下几种：

? Fanout：广播，将消息交给所有绑定到交换机的队列

? Direct：定向，把消息交给符合指定routing key 的队列

? Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列

? 注意：Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

3.3.1，Fanout：广播模式

? 在广播模式下，消息发送流程是这样的：

  • 1） 可以有多个消费者
  • 2） 每个消费者有自己的queue（队列）
  • 3） 每个队列都要绑定到Exchange（交换机）
  • 4） 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。
  • 5） 交换机把消息发送给绑定过的所有队列
  • 6） 队列的消费者都能拿到消息。实现一条消息被多个消费者消费

    创建生产者：

    1） 声明Exchange，不再声明Queue

    2） 发送消息到Exchange，不再发送到Queue

    public class Send {

    private final static String EXCHANGE_NAME = "fanout_exchange";

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();

        // 声明exchange，指定类型为fanout
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

        // 消息内容
        String message = "Hello everyone";
        // 发布消息到Exchange
        channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
        System.out.println(" [生产者] Sent '" + message + "'");

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

    ? 消费者1：

    public class Recv {
    private final static String QUEUE_NAME = "fanout_exchange_queue_1";

    private final static String EXCHANGE_NAME = "fanout_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);

        // 绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");

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

    要注意代码中：队列需要和交换机绑定

    消费者2：

    public class Recv2 {
    private final static String QUEUE_NAME = "fanout_exchange_queue_2";

    private final static String EXCHANGE_NAME = "fanout_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);

        // 绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");

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

    ? 运行结果:

    ? 

      ![](https://img2018.cnblogs.com/blog/1655301/201908/1655301-20190808204357990-1531661533.png)

    3.3.2， Direct：定向模式

    有选择性的接收消息

    在订阅模式中，生产者发布消息，所有消费者都可以获取所有消息。

    在路由模式中，我们将添加一个功能 - 我们将只能订阅一部分消息。 例如，我们只能将重要的错误消息引导到日志文件（以节省磁盘空间），同时仍然能够在控制台上打印所有日志消息。

    但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。

    在Direct模型下，队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）

    消息的发送方在向Exchange发送消息时，也必须指定消息的routing key。

P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。

X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列

C1：消费者，其所在队列指定了需要routing key 为 error 的消息

C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

生产者：

此处我们模拟商品的增删改，发送消息的RoutingKey分别是：insert、update、delete

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

    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();
        // 声明exchange，指定类型为direct
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        // 消息内容
        String message = "商品新增了， id = 1001";
        // 发送消息，并且指定routing key 为：insert ,代表新增商品
        channel.basicPublish(EXCHANGE_NAME, "insert", null, message.getBytes());
        System.out.println(" [商品服务：] Sent '" + message + "'");

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

? 消费者：

? 我们此处假设消费者1只接收两种类型的消息：更新商品和删除商品。

public class Recv {
    private final static String QUEUE_NAME = "direct_exchange_queue_1";
    private final static String EXCHANGE_NAME = "direct_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。假设此处需要update和delete消息
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");

        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, 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：

? 我们此处假设消费者2接收所有类型的消息：新增商品，更新商品和删除商品。

public class Recv2 {
    private final static String QUEUE_NAME = "direct_exchange_queue_2";
    private final static String EXCHANGE_NAME = "direct_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。订阅 insert、update、delete
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");

        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, 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);
    }
}

### 3.3.3，Topic：通配符

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

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

通配符规则：

? #：匹配一个或多个词

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

生产者在路由中指定insert操作：

 // 发送消息，并且指定routing key 为：insert ,代表新增商品
channel.basicPublish(EXCHANGE_NAME, "yx.insert", null, message.getBytes());

消费者1的队列接收更新，删除操作：

 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update");
  channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete");

消费者2

 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.insert");

3.4，持久化

如何避免消息丢失？

1） 消费者的ACK机制。可以防止消费者丢失消息。

2） 但是，如果在消费者消费之前，RabbitMQ就宕机了，消息就没了。

如何解决？

应该将队列、Exchange，消息都进行持久化。

队列持久化：

交换机持久化：

消息持久化：

四，MQ与Spring Boot整合


注意： Spring-amqp是对AMQP协议的抽象实现，而spring-rabbit 是对协议的具体实现，也是目前的唯一实现。底层使用的就是RabbitMQ。

4.1，配置和依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

application.yml中配置信息：

spring:
  rabbitmq:
    host: 192.168.129.130
    username: yx
    password: yx
    virtual-host: /yx

4.2，编写监听者

? 在SpringAmqp中，对消息的消费者进行了封装和抽象，一个普通的JavaBean中的普通方法，只要通过简单的注解，就可以成为一个消费者。

@Component
public class Listener {
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "spring.queue", durable = "true"),
            exchange = @Exchange(
                    value = "spring.exchange",
                    ignoreDeclarationExceptions = "true",
                    type = ExchangeTypes.TOPIC
            ),
            key = {"#.#"}))
    public void listen(String msg){
        System.out.println("接收到消息：" + msg);
    }
}

五，总结

? 以上就是RabbitMQ所提供的消息模型，总体来说大致为三种，一种基本模型，一种work模型，而剩下的订阅模型都大同小异。并没有涉及到一些难点，只是有些需要注意的细节问题。文章中如有不恰当之处，欢迎大牛评论指教。

原文地址：https://www.cnblogs.com/fenjyang/p/11323738.html

时间： 2024-10-11 15:36:47

RabbitMQ的基本介绍及与Spring整合的相关文章

JMS实现-ActiveMQ,介绍,安装,使用,注意点,spring整合

[TOC] 缘由: 最近在用netty开发游戏服务器,目前有这样的一个场景,聊天服务器和逻辑服务器要进行消息交互,比如,某个玩家往某个公会提交了加入申请,这个申请动作是在逻辑服务器上完成的,但是要产生一条申请消息,由聊天服务器推送到对应的公会频道,目前这个申请消息就是通过jms发送到聊天服务器上,聊天服务器监听到后,推送到对应的公会频道. 下面主要介绍以下几点 - JMS简介 - 消息传递模型 - ActiveMQ介绍 - 安装使用 - spring整合JMS - 代码相关 JMS简介 J Ja

Rabbitmq与spring整合之重要组件介绍——AMQP声明式配置&amp;RabbitTemplate组件

上一节是使用rabbitAdmin的管理组件进行声明队列,交换器,绑定等操作,本节则是采用AMQP声明式配置来声明这些东西.AMQP声明主要是通过@Bean注解进行的. 配置: 1 package com.zxy.demo.config; 2 3 import org.springframework.amqp.core.Binding; 4 import org.springframework.amqp.core.BindingBuilder; 5 import org.springframew

Spring整合JMS（三）——MessageConverter介绍

Spring整合JMS(三)——MessageConverter介绍 博客分类: Spring Jms SpringjmsintegratemessageConverter 1.4     消息转换器MessageConverter MessageConverter的作用主要有两方面,一方面它可以把我们的非标准化Message对象转换成我们的目标Message对象,这主要是用在发送消息的时候:另一方面它又可以把我们的Message对象转换成对应的目标对象,这主要是用在接收消息的时候. 下面我们就

spring整合RabbitMQ

今天就来康康spring怎么整合RabbitMQ 注意一点,在发送消息的时候对template进行配置mandatory=true保证监听有效 生产端还可以配置其他属性,比如发送重试,超时时间.次数.间隔等 消费端核心配置 首先配置手工确认模式,用于ACK的手工处理,这样我们可以保证消息的可靠性送达,或者在消费端消费失败的时候可以做到重回队列.根据业务记录日志等处理 可以设置消费端的监听个数和最大个数,用于控制消费端的并发情况 @RabbitListener注解的使用 消费端监听@RabbitL

RabbitMQ之交换机及spring整合

交换机 交换机属性: Name:交换机名称 Type:交换机类型 direct.topic.fanout.headers Durability:是否需要持久化,true为持久化 Auto Delete:当最后一个绑定到Exchange上的队列删除后,自动删除该Exchange Internal:当前Exchange是否用于RabbitMQ内部使用,默认为False Arguments:扩展参数,用于扩展AMQP协议,定制化使用 直流交换机 直连交换机Direct Exchange(完全匹配路由k

Spring整合JMS——基于ActiveMQ实现

1.1     JMS简介 JMS的全称是Java Message Service,即Java消息服务.它主要用于在生产者和消费者之间进行消息传递,生产者负责产生消息,而消费者负责接收消息.把它应用到实际的业务需求中的话我们可以在特定的时候利用生产者生成一消息,并进行发送,对应的消费者在接收到对应的消息后去完成对应的业务逻辑.对于消息的传递有两种类型,一种是点对点的,即一个生产者和一个消费者一一对应:另一种是发布/订阅模式,即一个生产者产生消息并进行发送后,可以由多个消费者进行接收. 1.2  

【jersey】 spring 整合jersey 实现RESTful webservice

Jersey是一个RESTFUL请求服务JAVA框架,与常规的JAVA编程使用的struts框架类似,它主要用于处理业务逻辑层.与Struts类似,它同样可以和hibernate,spring框架整合. 由于Struts2+hibernate+spring整合在市场的占有率太高,所以很少一部分人去关注Jersey.所以网上有关于Jersey的介绍很少.但是它确实是一个非常不错的框架.对于请求式服务,对于GET,DELETE请求,你甚至只需要给出一个URI即可完成操. jar 文件依赖: <pro

Spring整合Mybatis应用

1.整合步骤介绍 基于SpringMVC和Mybatis技术开发的主要步骤如下 1>创建工程,搭建SpringMVC和Mybatis技术环境 2>基于MapperScannerConfigurer方式整合Mybatis的Mapper接口(推荐) 3>编写和配置SpringMVC的主要组件,例如Controller,HandlerMapping,ViewResolver等 4>编写JSP视图组件,利用标签和表达式显示模型数据 5>测试程序 2.如何搭建SpringMVC和Myb

spring源码剖析（八）spring整合mybatis原理

前言 MyBatis相信很多人都会使用,但是当MyBatis整合到了Spring中,我们发现在Spring中使用更加方便了.例如获取Dao的实例,在Spring的我们只需要使用注入的方式就可以了使用Dao了,完全不需要调用SqlSession的getMapper方法去获取Dao的实例,更不需要我们去管理SqlSessionFactory,也不需要去创建SqlSession之类的了,对于插入操作也不需要我们commit. 既然那么方便,Spring到底为我们做了哪些工作呢,它如何将MyBatis整