Elasticsearch学习笔记-p8(实现elasticsearch与mysql的数据同步)

Posted LL.LEBRON

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Elasticsearch学习笔记-p8(实现elasticsearch与mysql的数据同步)相关的知识,希望对你有一定的参考价值。

视频指路👉B站黑马微服务超级推荐!

MQ实现elasticsearch与mysql的数据同步

elasticsearch中的酒店数据来自于mysql数据库,因此mysql数据发生改变时,elasticsearch也必须跟着改变,这个就是elasticsearch与mysql之间的数据同步

1.思路分析

常见的数据同步方案有三种:

  • 同步调用
  • 异步通知
  • 监听binlog

1.1 同步调用

方案一:同步调用

基本步骤如下:

  • hotel-demo对外提供接口,用来修改elasticsearch中的数据
  • 酒店管理服务在完成数据库操作后,直接调用hotel-demo提供的接口

1.2 异步通知

方案二:异步通知

流程如下:

  • hotel-adminmysql数据库数据完成增、删、改后,发送MQ消息
  • hotel-demo监听MQ,接收到消息后完成elasticsearch数据修改

1.3 监听binlog

方案三:监听binlog

流程如下:

  • mysql开启binlog功能
  • mysql完成增、删、改操作都会记录在binlog
  • hotel-demo基于canal监听binlog变化,实时更新elasticsearch中的内容

1.4 选择

方式一:同步调用

  • 优点:实现简单,粗暴
  • 缺点:业务耦合度高

方式二:异步通知

  • 优点:低耦合,实现难度一般
  • 缺点:依赖mq的可靠性

方式三:监听binlog

  • 优点:完全解除服务间耦合
  • 缺点:开启binlog增加数据库负担、实现复杂度高

2.MQ实现数据同步

2.1 思路

利用课前资料提供的hotel-admin项目作为酒店管理的微服务。当酒店数据发生增、删、改时,要求对elasticsearch中数据也要完成相同操作。

步骤:

  • 导入课前资料提供的hotel-admin项目,启动并测试酒店数据的CRUD

  • 声明exchangequeueRoutingKey

  • hotel-admin中的增、删、改业务中完成消息发送

  • hotel-demo中完成消息监听,并更新elasticsearch中数据

  • 启动并测试数据同步功能

2.2 声明交换机、队列

MQ结构如图:

(1)引入依赖

hotel-adminhotel-demo中引入rabbitmq的依赖:

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

(2)声明队列交换机名称

hotel-adminhotel-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中,定义配置类MqConfig,声明队列、交换机、绑定队列和交换机:

@Configuration
public class MqConfig 
    /**
     * 定义交换机
     * @return
     */
    @Bean
    public TopicExchange topicExchange() 
        return new TopicExchange(MqConstants.HOTEL_EXCHANGE, true, false);
    

    /**
     * 定义新增或修改的队列
     * @return
     */
    @Bean
    public Queue insertQueue() 
        return new Queue(MqConstants.HOTEL_INSERT_QUEUE, true);
    

    /**
     * 定义删除的队列
     * @return
     */
    @Bean
    public Queue deleteQueue() 
        return new Queue(MqConstants.HOTEL_DELETE_QUEUE, true);
    

    /**
     * 绑定修改队列和交换机
     * @return
     */
    @Bean
    public Binding insertQueueBinding() 
        return BindingBuilder.bind(insertQueue())
                .to(topicExchange())
                .with(MqConstants.HOTEL_INSERT_KEY);
    

    /**
     * 绑定删除队列和交换机
     * @return
     */
    @Bean
    public Binding deleteQueueBinding() 
        return BindingBuilder.bind(deleteQueue())
                .to(topicExchange())
                .with(MqConstants.HOTEL_DELETE_KEY);
    

2.4 发送MQ消息

当在hotel-admin中发生增、删、改业务中分别发送MQ消息:

要传入酒店的id给hotel-demo用于修改

@PostMapping
public void saveHotel(@RequestBody Hotel hotel)
    // 新增酒店
    hotelService.save(hotel);
    // 发送MQ消息
    rabbitTemplate.convertAndSend(HotelMqConstants.EXCHANGE_NAME, HotelMqConstants.INSERT_KEY, hotel.getId());


@PutMapping()
public void updateById(@RequestBody Hotel hotel)
    if (hotel.getId() == null) 
        throw new InvalidParameterException("id不能为空");
    
    hotelService.updateById(hotel);
    // 发送MQ消息
    rabbitTemplate.convertAndSend(HotelMqConstants.EXCHANGE_NAME, HotelMqConstants.INSERT_KEY, hotel.getId());


@DeleteMapping("/id")
public void deleteById(@PathVariable("id") Long id) 
    hotelService.removeById(id);
    // 发送MQ消息
    rabbitTemplate.convertAndSend(HotelMqConstants.EXCHANGE_NAME, HotelMqConstants.DELETE_KEY, id);

2.5 接受MQ消息

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

  • 新增消息:根据传递的hotel的id查询hotel信息,然后新增一条数据到索引库
  • 删除消息:根据传递的hotel的id删除索引库中的一条数据

(1)首先在hotel-democn.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包新增一个类:

@Component
public class Hotelistener 
    @Autowired
    private IHotelService iHotelServicel;
    /**
     * 监听酒店新增或者修改的业务
     * @param id 酒店id
     */
    @RabbitListener(queues = MqConstants.HOTEL_INSERT_QUEUE)
    public void listenHotelInsertOrupdate(Long id)
        iHotelServicel.saveById(id);
    
    /**
     * 监听酒店删除的业务
     * @param id 酒店id
     */
    @RabbitListener(queues = MqConstants.HOTEL_DELETE_QUEUE)
    public void listenHotelDelete(Long id)
        iHotelServicel.deleteById(id);
    


最后喜欢的小伙伴,记得三连哦!😏🍭😘

以上是关于Elasticsearch学习笔记-p8(实现elasticsearch与mysql的数据同步)的主要内容,如果未能解决你的问题,请参考以下文章

阿里P8的《MySQL学习笔记》火了,完整版开放下载!

阿里工作7年,肝到P8就剩这份学习笔记了,已助10个朋友拿到offer

阿里工作8年,肝到P8就剩这份学习笔记了,已助朋友拿到10个Offer

阿里工作6年,肝到P8就剩这份学习笔记了,已助朋友拿到7个Offer

阿里工作8年,肝到P8就剩这份学习笔记了,已助朋友拿到10个Offer

备战金九银十,阿里P8师兄指导完整攻略(附:学习资料+面试宝典+项目实战笔记)