流程

1.聲明exchange、queue、RoutingKey
2. 在hotel-admin中進(jìn)行增刪改（SQL），完成消息發(fā)送
3. 在hotel-demo中完成消息監(jiān)聽(tīng)，并更新elasticsearch數(shù)據(jù)
4. 測(cè)試同步

1.引入依賴(lài)

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

我這里的mq是掛在了docker上，虛擬機(jī)地址是192.168.116.128。到時(shí)候這個(gè)根據(jù)自己的項(xiàng)目改就行

 spring: 
  rabbitmq:
    host: 192.168.116.128 # 主機(jī)名
    port: 5672 # 端口
    virtual-host: / # 虛擬主機(jī)
    username: itcast # 用戶(hù)名
    password: 123321 # 密碼

2.聲明交換機(jī)、隊(duì)列和綁定關(guān)系

package cn.itcast.hotel.constants;

public class MqConstants {

    /**
     * 交換機(jī)
     */
    public final static String HOTEL_EXCHANGE = "hotel.topic";
    /**
     * 監(jiān)聽(tīng)新增和修改的隊(duì)列
     */
    public final static String HOTEL_INSERT_QUEUE = "hotel.insert.queue";
    /**
     * 監(jiān)聽(tīng)刪除的隊(duì)列
     */
    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";
}

在hotel-demo中，定義配置類(lèi)，聲明隊(duì)列、交換機(jī)：

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 {
    // 定義交換機(jī)
    @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange(MqConstants.HOTEL_EXCHANGE,true,false);
    }
    // 定義隊(duì)列
    @Bean
    public Queue insertQueue()
    {
        return new Queue(MqConstants.HOTEL_INSERT_QUEUE,true);
    }
    @Bean
    public Queue deleteQueue()
    {
        return new Queue(MqConstants.HOTEL_DELETE_QUEUE,true);
    }

    // 定義綁定關(guān)系
    @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.發(fā)送MQ消息

在hotel-admin中的增、刪、改業(yè)務(wù)中分別發(fā)送MQ消息,具體怎么添加根據(jù)：

    // 新增
    @PostMapping
    public void saveHotel(){
        //數(shù)據(jù)庫(kù)新增操作
        rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,MqConstants.HOTEL_INSERT_KEY,hotel.getId());
    }

    // 更新
    @PutMapping()
    public void updateById(){
            //數(shù)據(jù)庫(kù)修改操作
            rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,MqConstants.HOTEL_INSERT_KEY,hotel.getId());

    }
    // 刪除
    @DeleteMapping("/{id}")
    public void deleteById() {
		// 數(shù)據(jù)庫(kù)刪除操作
        rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,MqConstants.HOTEL_DELETE_KEY,id);

    }

4.監(jiān)聽(tīng)MQ消息

接收MQ消息

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

• 新增消息：根據(jù)傳遞的hotel的id查詢(xún)hotel信息，然后新增一條數(shù)據(jù)到索引庫(kù)
• 刪除消息：根據(jù)傳遞的hotel的id刪除索引庫(kù)中的一條數(shù)據(jù)

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

void deleteById(Long id);

void insertById(Long id);

2）給hotel-demo中的cn.itcast.hotel.service.impl包下的HotelService中實(shí)現(xiàn)業(yè)務(wù)：文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-707020.html

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

@Override
public void insertById(Long id) {
    try {
        // 0.根據(jù)id查詢(xún)酒店數(shù)據(jù)
        Hotel hotel = getById(id);
        // 轉(zhuǎn)換為文檔類(lèi)型
        HotelDoc hotelDoc = new HotelDoc(hotel);

        // 1.準(zhǔn)備Request對(duì)象
        IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString());
        // 2.準(zhǔn)備Json文檔
        request.source(JSON.toJSONString(hotelDoc), XContentType.JSON);
        // 3.發(fā)送請(qǐng)求
        client.index(request, RequestOptions.DEFAULT);
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}

編寫(xiě)監(jiān)聽(tīng)類(lèi)

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 {
    // 專(zhuān)門(mén)用于消息監(jiān)聽(tīng)的類(lèi)

    @Autowired
    private IHotelService hotelService;

    @RabbitListener(queues = MqConstants.HOTEL_INSERT_QUEUE)
    public void listenHotelInsertOrUpdate(Long id)
    {
        hotelService.insertById(id);
    }

    @RabbitListener(queues = MqConstants.HOTEL_DELETE_QUEUE)
    public void listenHotelDelete(Long id)
    {
        hotelService.deleteById(id);
    }
}

到了這里，關(guān)于利用MQ實(shí)現(xiàn)mysql與elasticsearch數(shù)據(jù)同步的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！