Elasticsearch-數據同步

elasticsearch中的酒店數據來自於mysql數據庫，因此mysql數據發生改變時，elasticsearch也必須跟着改變，這個就是elasticsearch與mysql之間的數據同步。

3.1.思路分析

常見的數據同步方案有三種：

• 同步調用
• 異步通知
• 監聽binlog

3.1.1.同步調用

方案一：同步調用

基本步驟如下：

• hotel-demo對外提供接口，用來修改elasticsearch中的數據
• 酒店管理服務在完成數據庫操作后，直接調用hotel-demo提供的接口，

3.1.2.異步通知

方案二：異步通知

流程如下：

• hotel-admin對mysql數據庫數據完成增、刪、改后，發送MQ消息
• hotel-demo監聽MQ，接收到消息后完成elasticsearch數據修改

3.1.3.監聽binlog

方案三：監聽binlog

流程如下：

• 給mysql開啟binlog功能
• mysql完成增、刪、改操作都會記錄在binlog中
• hotel-demo基於canal監聽binlog變化，實時更新elasticsearch中的內容

3.1.4.選擇

方式一：同步調用

• 優點：實現簡單，粗暴
• 缺點：業務耦合度高

方式二：異步通知

• 優點：低耦合，實現難度一般
• 缺點：依賴mq的可靠性

方式三：監聽binlog

• 優點：完全解除服務間耦合
• 缺點：開啟binlog增加數據庫負擔、實現復雜度高

3.2.實現數據同步

3.2.1.思路

利用課前資料提供的hotel-admin項目作為酒店管理的微服務。當酒店數據發生增、刪、改時，要求對elasticsearch中數據也要完成相同操作。

步驟：

• 導入課前資料提供的hotel-admin項目，啟動並測試酒店數據的CRUD
• 聲明exchange、queue、RoutingKey
• 在hotel-admin中的增、刪、改業務中完成消息發送
• 在hotel-demo中完成消息監聽，並更新elasticsearch中數據
• 啟動並測試數據同步功能

3.2.2.導入demo

導入課前資料提供的hotel-admin項目：
就是上面一直再寫的小案例

鏈接：https://pan.baidu.com/s/1dpTvl0RvfbLOxoJUmKaefQ
提取碼：1211

運行后，訪問 http://localhost:8099
其中包含了酒店的CRUD功能：

3.2.3.聲明交換機、隊列

MQ結構如圖：

1）引入依賴

在hotel-admin、hotel-demo中引入rabbitmq的依賴：

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

在yaml文件中配置rabbitmq地址

server:
  port: 8099
spring:
  datasource:
    url: jdbc:mysql://localhost:3306/heima?useSSL=false
    username: root
    password: 123456
    driver-class-name: com.mysql.jdbc.Driver
  rabbitmq:
    host: 192.168.52.132
    port: 5672
    username: itcast
    password: 123321
    virtual-host: /
logging:
  level:
    cn.itcast: debug
  pattern:
    dateformat: MM-dd HH:mm:ss:SSS
mybatis-plus:
  configuration:
    map-underscore-to-camel-case: true
  type-aliases-package: cn.itcast.hotel.pojo

在docker中啟動mq，具體可以看之前寫的mq安裝

2）聲明隊列交換機名稱

在hotel-admin和hotel-demo中的cn.itcast.hotel.constatnts包下新建一個類MqConstants：

package cn.itcast.hotel.constatnts;

    public class MqConstants {
    /**
     * 交換機
     */
    public final static String HOTEL_EXCHANGE = "hotel.topic";
    /**
     * 監聽新增和修改的隊列
     */
    public final static String HOTEL_INSERT_QUEUE = "hotel.insert.queue";
    /**
     * 監聽刪除的隊列
     */
    public final static String HOTEL_DELETE_QUEUE = "hotel.delete.queue";
    /**
     * 新增或修改的RoutingKey
     */
    public final static String HOTEL_INSERT_KEY = "hotel.insert";
    /**
     * 刪除的RoutingKey
     */
    public final static String HOTEL_DELETE_KEY = "hotel.delete";
}

3）聲明隊列交換機

在hotel-demo中，定義配置類，聲明隊列、交換機：

package cn.itcast.hotel.config;

import cn.itcast.hotel.constants.MqConstants;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MqConfig {
    @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange(MqConstants.HOTEL_EXCHANGE, true, false);
    }

    @Bean
    public Queue insertQueue(){
        return new Queue(MqConstants.HOTEL_INSERT_QUEUE, true);
    }

    @Bean
    public Queue deleteQueue(){
        return new Queue(MqConstants.HOTEL_DELETE_QUEUE, true);
    }

    @Bean
    public Binding insertQueueBinding(){
        return BindingBuilder.bind(insertQueue()).to(topicExchange()).with(MqConstants.HOTEL_INSERT_KEY);
    }

    @Bean
    public Binding deleteQueueBinding(){
        return BindingBuilder.bind(deleteQueue()).to(topicExchange()).with(MqConstants.HOTEL_DELETE_KEY);
    }
}

3.2.4.發送MQ消息

在hotel-admin中的增、刪、改業務中分別發送MQ消息：

3.2.5.接收MQ消息

hotel-demo接收到MQ消息要做的事情包括：

• 新增消息：根據傳遞的hotel的id查詢hotel信息，然后新增一條數據到索引庫
• 刪除消息：根據傳遞的hotel的id刪除索引庫中的一條數據

1）首先在hotel-demo的cn.itcast.hotel.service包下的IHotelService中新增新增、刪除業務

void deleteById(Long id);

void insertById(Long id);

2）給hotel-demo中的cn.itcast.hotel.service.impl包下的HotelService中實現業務：

@Override
public void deleteById(Long id) {
    try {
        // 1.准備Request
        DeleteRequest request = new DeleteRequest("hotel", id.toString());
        // 2.發送請求
        client.delete(request, RequestOptions.DEFAULT);
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}

@Override
public void insertById(Long id) {
    try {
        // 0.根據id查詢酒店數據
        Hotel hotel = getById(id);
        // 轉換為文檔類型
        HotelDoc hotelDoc = new HotelDoc(hotel);

        // 1.准備Request對象
        IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString());
        // 2.准備Json文檔
        request.source(JSON.toJSONString(hotelDoc), XContentType.JSON);
        // 3.發送請求
        client.index(request, RequestOptions.DEFAULT);
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}

3）編寫監聽器

在hotel-demo中的cn.itcast.hotel.mq包新增一個類：

package cn.itcast.hotel.mq;

import cn.itcast.hotel.constants.MqConstants;
import cn.itcast.hotel.service.IHotelService;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class HotelListener {

    @Autowired
    private IHotelService hotelService;

    /**
     * 監聽酒店新增或修改的業務
     * @param id 酒店id
     */
    @RabbitListener(queues = MqConstants.HOTEL_INSERT_QUEUE)
    public void listenHotelInsertOrUpdate(Long id){
        hotelService.insertById(id);
    }

    /**
     * 監聽酒店刪除的業務
     * @param id 酒店id
     */
    @RabbitListener(queues = MqConstants.HOTEL_DELETE_QUEUE)
    public void listenHotelDelete(Long id){
        hotelService.deleteById(id);
    }
}

刪除，傳過來的id，這邊直接刪除

新增或修改，傳過來的id，到了這邊再查詢數據庫，再進行操作，自我感覺這樣應該不適用真是項目吧

這邊自動插入，保證數據一致