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

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

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

​ 通常有两种解决方案：

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

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

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

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

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

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

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

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

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

​ 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链接。

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

​ 两者之间的区别：

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

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

​ 基本模型介绍：

​ 文中指出，我们可以将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);
    }
}

​ 接收者结果如图：

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

那么问题来了：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机制改为手动时，只有当消息真正消费结束时，通知消息队列并删除消息。

​ 介绍：

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

​ 这个概念在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）接收到消息以后，通过线程池，异步消费。

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

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

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

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

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

再次运行结果：

详解：

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

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

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

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

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

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

Exchange类型有以下几种：

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

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

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

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

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