本文共 10499 字，大约阅读时间需要 34 分钟。

目前已经完成了商品详情和搜索系统的开发。思考一下，是否存在问题？

商品的原始数据保存在数据库中，增删改查都在数据库中完成。

搜索服务数据来源是索引库，如果数据库商品发生变化，索引库数据不能及时更新。

商品详情做了页面静态化，静态页面数据也不会随着数据库商品发生变化。

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

这里有两种解决方案：

方案1：每当后台对商品做增删改操作，同时要修改索引库数据及静态页面

方案2：搜索服务和商品页面服务对外提供操作接口，后台在商品增删改后，调用接口

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

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

下载和安装

下载

官网下载地址：http://www.rabbitmq.com/download.html

安装

五种消息模型

RabbitMQ提供了6种消息模型，但是第6种其实是RPC，并不是MQ，因此不予学习。那么也就剩下5种。

但是其实3、4、5这三种都属于订阅模型，只不过进行路由的方式不同。

基本消息模型

RabbitMQ怎么知道消息被接收了？

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

• 自动ACK：消息一旦被接收，消费者自动发送ACK
• 手动ACK：消息接收后，不会发送ACK，需要手动调用
• 如果消息不太重要，丢失也没有影响，那么自动ACK会比较方便

public class Recv {
       private final static String QUEUE_NAME = "simple_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);        // 定义队列的消费者        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(" [x] received : " + msg + "!");            }        };        // 监听队列，第二个参数：是否自动进行消息确认。        channel.basicConsume(QUEUE_NAME, true, consumer);    }}

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

public class Recv2 {
       private final static String QUEUE_NAME = "simple_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);        // 定义队列的消费者        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(" [x] received : " + msg + "!");                // 手动进行ACK                channel.basicAck(envelope.getDeliveryTag(), false);            }        };        // 监听队列，第二个参数false，手动进行ACK        channel.basicConsume(QUEUE_NAME, false, consumer);    }}

注意到最后一行代码：

channel.basicConsume(QUEUE_NAME, false, consumer);

如果第二个参数为true，则会自动进行ACK；如果为false，则需要手动ACK。方法的声明：

work消息模型

当你运行许多消费者时，任务将在他们之间共享，但是一个消息只能被一个消费者获取

面试题：避免消息堆积？

1）采用workqueue，多个消费者监听同一队列。2）接收到消息以后，而是通过线程池，异步消费。

能者多劳

订阅模型分类

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

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

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

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

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

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

Exchange类型有以下几种：

Fanout：广播，将消息交给所有绑定到交换机的队列Direct：定向，把消息交给符合指定routing key 的队列 Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列

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

订阅模型-Fanout

有选择性的接收消息

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

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

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

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

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

订阅模型-Direct

订阅模型-Topic

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

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

通配符规则：

`#`：匹配一个或多个词`*`：匹配不多不少恰好1个词

举例：

`audit.#`：能够匹配`audit.irs.corporate` 或者 `audit.irs``audit.*`：只能匹配`audit.irs`

生产者

使用topic类型的Exchange，发送消息的routing key有3种： item.isnert、item.update、item.delete：

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

消费者1

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

public class Recv {
       private final static String QUEUE_NAME = "topic_exchange_queue_1";    private final static String EXCHANGE_NAME = "topic_exchange_test";    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, "item.update");        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.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接收所有类型的消息：新增商品，更新商品和删除商品。

/** * 消费者2 */public class Recv2 {
       private final static String QUEUE_NAME = "topic_exchange_queue_2";    private final static String EXCHANGE_NAME = "topic_exchange_test";    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, "item.*");        // 定义队列的消费者        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);    }}

持久化

如何避免消息丢失？

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

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

是可以将消息进行持久化呢？

要将消息持久化，前提是：队列、Exchange都持久化

交换机持久化

队列持久化

消息持久化

项目改造

接下来，我们就，实现搜索服务、商品静态页的数据同步。

发送方：商品微服务leyou-item-service

• 什么时候发？

  当商品服务对商品进行写操作：增、删、改的时候，需要发送一条消息，通知其它服务。

• 发送什么内容？

  对商品的增删改时其它服务可能需要新的商品数据，但是如果消息内容中包含全部商品信息，数据量太大，而且并不是每个服务都需要全部的信息。因此我们只发送商品id，其它服务可以根据id查询自己需要的信息。

接收方：搜索微服务、静态页微服务

接收消息后如何处理？

• 搜索微服务：
  • 增/改：添加新的数据到索引库
  • 删：删除索引库数据
• 静态页微服务：
  • 增/改：创建新的静态页
  • 删：删除原来的静态页

商品服务发送消息

搜索服务接收消息

@Componentpublic class GoodsListener {
       @Autowired    private SearchService searchService;    /**     * 处理insert和update的消息     *     * @param id     * @throws Exception     */    @RabbitListener(bindings = @QueueBinding(            value = @Queue(value = "leyou.create.index.queue", durable = "true"),            exchange = @Exchange(                    value = "leyou.item.exchange",                    ignoreDeclarationExceptions = "true",                    type = ExchangeTypes.TOPIC),            key = {
   "item.insert", "item.update"}))    public void listenCreate(Long id) throws Exception {
           if (id == null) {
               return;        }        // 创建或更新索引        this.searchService.createIndex(id);    }    /**     * 处理delete的消息     *     * @param id     */    @RabbitListener(bindings = @QueueBinding(            value = @Queue(value = "leyou.delete.index.queue", durable = "true"),            exchange = @Exchange(                    value = "leyou.item.exchange",                    ignoreDeclarationExceptions = "true",                    type = ExchangeTypes.TOPIC),            key = "item.delete"))    public void listenDelete(Long id) {
           if (id == null) {
               return;        }        // 删除索引        this.searchService.deleteIndex(id);    }}

静态页服务接收消息

@Componentpublic class GoodsListener {
       @Autowired    private GoodsHtmlService goodsHtmlService;    @RabbitListener(bindings = @QueueBinding(            value = @Queue(value = "leyou.create.web.queue", durable = "true"),            exchange = @Exchange(                    value = "leyou.item.exchange",                    ignoreDeclarationExceptions = "true",                    type = ExchangeTypes.TOPIC),            key = {
   "item.insert", "item.update"}))    public void listenCreate(Long id) throws Exception {
           if (id == null) {
               return;        }        // 创建页面        goodsHtmlService.createHtml(id);    }    @RabbitListener(bindings = @QueueBinding(            value = @Queue(value = "leyou.delete.web.queue", durable = "true"),            exchange = @Exchange(                    value = "leyou.item.exchange",                    ignoreDeclarationExceptions = "true",                    type = ExchangeTypes.TOPIC),            key = "item.delete"))    public void listenDelete(Long id) {
           if (id == null) {
               return;        }        // 删除页面        goodsHtmlService.deleteHtml(id);    }}

查看RabbitMQ控制台

重新启动项目，并且登录RabbitMQ管理界面：http://192.168.56.101:15672

a.简单模型:一个生产者一个队列一个消费者b.工作模型:一个生产者一个队列多个消费者，一个消息只能被消费一 次c.订阅模型之fanout:一个生产者一个交换机多个队列多个消费者，一个消息可以被多个消费者消费d.订阅模型之direct:一个生产者一个交换机多个队列多个消费者，routingKey, 一个消息发送给符合routingkey的队列e.订阅模型之topic:       通配符     #: 匹配一个或者多个  ,     *:一个词

转载地址：http://sxxab.baihongyu.com/