快捷導(dǎo)航

一文詳解Elasticsearch和MySQL之間的數(shù)據(jù)同步問題

更新時(shí)間：2023年04月18日 09:07:10 作者：小威要向諸佬學(xué)習(xí)呀

Elasticsearch中的數(shù)據(jù)是來自于Mysql數(shù)據(jù)庫的，因此當(dāng)數(shù)據(jù)庫中的數(shù)據(jù)進(jìn)行增刪改后，Elasticsearch中的數(shù)據(jù)，索引也必須跟著做出改變。本文主要來和大家探討一下Elasticsearch和MySQL之間的數(shù)據(jù)同步問題，感興趣的可以了解一下

Elasticsearch中的數(shù)據(jù)是來自于Mysql數(shù)據(jù)庫的，因此當(dāng)數(shù)據(jù)庫中的數(shù)據(jù)進(jìn)行增刪改后，Elasticsearch中的數(shù)據(jù)，索引也必須跟著做出改變。而對(duì)于管理服務(wù)（MySQL）和搜索服務(wù)（Elasticsearch）往往會(huì)在不同的微服務(wù)上。

可以通過微服務(wù)之間的同步調(diào)用來解決數(shù)據(jù)同步問題，雖然實(shí)現(xiàn)起來比較簡(jiǎn)單，但是在搜索服務(wù)中引入管理服務(wù)時(shí)，業(yè)務(wù)的耦合度相對(duì)來說是比較高的。因此可以通過消息隊(duì)列的形式來異步通知管理服務(wù)的改變。這樣做的有優(yōu)點(diǎn)是耦合度較低，但是依賴于消息隊(duì)列的耦合度。

下面根據(jù)一個(gè)例子來介紹使用RabbitMQ來解決Elasticsearch和MySQL之間的數(shù)據(jù)同步問題。當(dāng)酒店數(shù)據(jù)庫中發(fā)送增刪改時(shí)，Elasticsearch中的數(shù)據(jù)也同時(shí)發(fā)生對(duì)應(yīng)的改變。

首先在兩塊微服務(wù)中引入RabbitMQ的依賴：

引入依賴

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

創(chuàng)建常量類，將交換機(jī)和隊(duì)列名稱設(shè)置為常量，使用時(shí)方便取名。

public class MqConstants {
    /**
     * 交換機(jī)名稱
     */
    public final static String HOTEL_EXCHANGE = "hotel.topic";
    /**
     * 監(jiān)聽新增和修改的隊(duì)列
     */
    public final static String HOTEL_INSERT_QUEUE = "hotel.insert.queue";
    /**
     * 監(jiān)聽刪除的隊(duì)列
     */
    public final static String HOTEL_DELETE_QUEUE = "hotel.delete.queue";
    /**
     * 新增或修改的路由鍵
     */
    public final static String HOTEL_INSERT_KEY = "hotel.insert";
    /**
     * 刪除的路由鍵
     */
    public final static String HOTEL_DELETE_KEY = "hotel.delete";
}

創(chuàng)建配置類，聲明交換機(jī)，并將路由鍵，隊(duì)列與交換機(jī)之間互相綁定：

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

編寫controller層，將酒店數(shù)據(jù)保存到數(shù)據(jù)庫中，并將消息發(fā)送到消息隊(duì)列里：

    @Autowired
    private IHotelService hotelService;
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @PutMapping
    public void save(@RequestBody Hotel hotel){
        hotelService.save(hotel);
        rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,MqConstants.HOTEL_INSERT_KEY,hotel.getId());
    }
    @DeleteMapping("/{id}")
    public void deleteById(@PathVariable("id") Long id){
        hotelService.removeById(id);
        rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,MqConstants.HOTEL_DELETE_KEY,id);
    }

在service中加入兩個(gè)方法，根據(jù)id增加和根據(jù)id刪除：

    void insertById(Long id);
    void deleteById(Long id);

利用Ctrl+Alt+B的快捷鍵，在service的實(shí)現(xiàn)類里面重寫方法進(jìn)行實(shí)現(xiàn)：

    @Override
    public void insertById(Long id) {
        try {
            Hotel hotel = getById(id);
            HotelDoc hotelDoc = new HotelDoc(hotel);
            // 1.準(zhǔn)備Request
            IndexRequest request = new IndexRequest("hotel").id(hotelDoc.getId().toString());
            // 2.準(zhǔn)備請(qǐng)求參數(shù)DSL，其實(shí)就是文檔的JSON字符串
            request.source(JSON.toJSONString(hotelDoc), XContentType.JSON);
            // 3.發(fā)送請(qǐng)求
            client.index(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

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

這樣即可完成Elasticsearch和MySQL之間的數(shù)據(jù)同步。

到此這篇關(guān)于一文詳解Elasticsearch和MySQL之間的數(shù)據(jù)同步問題的文章就介紹到這了,更多相關(guān)Elasticsearch MySQL數(shù)據(jù)同步內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: