乐优商城（九）数据同步

1. 项目问题分析

现在项目中有三个独立的微服务：

商品微服务：原始数据保存在 MySQL 中，从 MySQL 中增删改查商品数据。
搜索微服务：原始数据保存在 ES 的索引库中，从 ES 中查询商品数据。
商品详情微服务：做了页面静态化，静态页面的商品数据不会随着数据库发生变化。

假如我在商品微服务中，修改了商品的数据，也就是 MySQL 中数据发生了改变。但搜索微服务查询到的数据还是原来的，商品详情微服务的生成的静态页面也没有发生改变，这样显然不对。我们需要实现数据的同步，让搜索微服务和商品详情微服务的商品也发生修改。

我们使用消息队列技术（RabbitMQ）来实现数据同步：

2. 项目改造

接下来，我们就改造项目，实现搜索微服务和商品详情微服务的数据同步。

2.1 改造思路

生产者：商品微服务

什么时候发送消息？

当商品微服务对商品进行增、删、改的操作时候，就发送一条消息。
发送什么内容？

我们发送商品 id，其它微服务可以根据 id 查询自己需要的信息。

消费者：搜索微服务和商品详情微服务

接收消息后如何处理？

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

2.2 商品微服务发送消息

在 leyou-item-service 中实现商品微服务发送消息。

2.2.1 引入依赖

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

2.2.2 配置文件

在 application.yaml 中添加 RabbitMQ 的配置：

spring:
  rabbitmq:
    host: 127.0.0.1
    username: leyou
    password: leyou
    virtual-host: /leyou
    template:
      exchange: leyou.item.exchange
    publisher-confirms: true

template：有关 AmqpTemplate 的配置
- exchange：缺省的交换机名称，此处配置后，发送消息如果不指定交换机就会使用这个
publisher-confirms：生产者确认机制，确保消息会正确发送，如果发送失败会有错误回执，从而触发重试

2.2.3 改造 SpuService

注入 AmpqTemplate 模板

@Autowired
private AmqpTemplate amqpTemplate;

编写发送消息方法

/**
 * 发送消息
 * @param id 商品 id
 * @param type
 */
private void sendMessage(Long id, String type){
    try {
        this.amqpTemplate.convertAndSend("item." + type, id);
    } catch (Exception e) {
        e.printStackTrace();
    }
}

在新增方法中，调用发送消息方法

在修改方法中，调用发送消息方法

2.3 搜索微服务接收消息

在 leyou-search 中实现搜索微服务接收消息。

2.3.1 引入依赖

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

2.3.2 配置文件

spring:
  rabbitmq:
    host: 127.0.0.1
    username: leyou
    password: leyou
    virtual-host: /leyou

2.3.3 编写监听器

@Component
public 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);
    }
}

2.3.4 编写创建和删除索引方法

在 SearchService 中添加创建和删除索引方法：

/**
 * 创建索引
 *
 * @param id
 * @throws IOException
 */
public void createIndex(Long id) throws IOException {
    Spu spu = this.spuClient.querySpuById(id);
    // 构建商品
    Goods goods = this.buildGoods(spu);

    // 保存数据到索引库
    this.goodsRepository.save(goods);
}

/**
 * 删除索引
 *
 * @param id
 */
public void deleteIndex(Long id) {
    this.goodsRepository.deleteById(id);
}

2.4 商品详情微服务接收消息

在 leyou-goods-web 中实现商品详情微服务接收消息。

2.4.1 引入依赖

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

2.4.2 配置文件

spring:
  rabbitmq:
    host: 127.0.0.1
    username: leyou
    password: leyou
    virtual-host: /leyou

2.4.3 编写监听器

@Component
public class GoodsListener {

    @Autowired
    private GoodsHtmlService goodsHtmlService;

    /**
     * 处理 insert 和 update 的消息
     *
     * @param id
     * @throws Exception
     */
    @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);
    }

    /**
     * 处理 delete 的消息
     *
     * @param 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);
    }
}

2.4.4 添加删除页面方法

在 GoodsHtmlService 中添加删除页面方法

/**
 * 删除 HTML 页面
 * @param id
 */
public void deleteHtml(Long id) {
    File file = new File("D:\\nginx-1.14.0\\html\\item\\" + id + ".html");
    file.deleteOnExit();
}