0
点赞
收藏
分享

微信扫一扫

Mysql和Elasticsearch的数据同步

科牛 2022-06-30 阅读 104
java

Elasticsearch的数据来自Mysql数据库中,所以当我们的MySQL发生改变时,Elasticsearch也要跟着改变,这时候我们的es的数据就要和mysql同步了


同步实现思路

 方案一:
      

 

 方案二

 

方式三:

 

 在这里使用的是第二种实现方案:使用MQ来写


同步案例代码
 

用来操控ES的代码(负责监听MQ队列)

/**
     * 监听增加和修改的队列
     * 因为我们的ES中可以进行全量修改,当有这个id的数据的时候那么就先删除再新增,没有这个数据那么就直接新增
     * 所以队列过来的id不管是新增还是修改es都可以判断如果有这个数据id那么就先删除再新增,如果没有这个数据就直接新增,所以新增和修改他俩用一个方法就行了
     *
     * @param id 队列中需要进行操作的id
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = MqConstants.HOTEL_INSERT_QUEUE),
            exchange = @Exchange(name = MqConstants.HOTEL_EXCHANGE, type = ExchangeTypes.DIRECT),
            key = MqConstants.HOTEL_INSERT_KEY
    ))
    public void insertAndUpdate(Long id) {
        if (id == null) {
            return;
        }
        log.info("入参:{}", id);
        //监听到以后拿到id去数据库查询整个数据
        Hotel hotel = iHotelService.getById(id);
        //因为查的mysql数据和es的数据有些不一样所以需要做转换
        HotelDoc hotelDoc = new HotelDoc(hotel);
        //转换为json
        String hotelDocJson = JSON.toJSONString(hotelDoc);
        System.out.println("hotelDocJson = " + hotelDocJson);
        //发送到ES中,因为我们的ES中可以进行全量修改,当有这个id的数据的时候那么就先删除再新增,没有这个数据那么就直接新增
        //创建请求语义对象 添加文档数据
        IndexRequest request = new IndexRequest("hotel");
        //这个新增就是PUT在es中
        request.id(hotel.getId().toString()).source(hotelDocJson, XContentType.JSON);
        //发送请求
        try {
            IndexResponse response = client.index(request, RequestOptions.DEFAULT);
            RestStatus status = response.status();
            log.info("响应结果为:{}", status);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 监听删除队列
     *
     * @param id 队列中需要进行操作的id
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = MqConstants.HOTEL_DELETE_QUEUE),
            exchange = @Exchange(name = MqConstants.HOTEL_EXCHANGE, type = ExchangeTypes.DIRECT),
            key = MqConstants.HOTEL_DELETE_KEY
    ))
    public void deleteByMqId(Long id) {
        if (id == null) {
            return;
        }
        log.info("入参:{}", id);
        //先创建语义对象,直接就可以给里面写id的字段
        DeleteRequest request = new DeleteRequest("hotel", id.toString());
        //发送请求
        try {
            DeleteResponse response = client.delete(request, RequestOptions.DEFAULT);
            RestStatus status = response.status();
            log.info("响应结果为:{}", status);
        } catch (IOException e) {
            e.printStackTrace();
        }

 用来操作MySQL的代码:
 

@RestController
@RequestMapping("hotel")
public class HotelController {

    //注入和RabbitMQ链接
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private IHotelService hotelService;

    @GetMapping("/{id}")
    public Hotel queryById(@PathVariable("id") Long id) {
        return hotelService.getById(id);
    }

    @GetMapping("/list")
    public PageResult hotelList(
            @RequestParam(value = "page", defaultValue = "1") Integer page,
            @RequestParam(value = "size", defaultValue = "1") Integer size
    ) {
        Page<Hotel> result = hotelService.page(new Page<>(page, size));

        return new PageResult(result.getTotal(), result.getRecords());
    }

    @PostMapping
    public void saveHotel(@RequestBody Hotel hotel) {
        hotelService.save(hotel);
        rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE, MqConstants.HOTEL_INSERT_KEY, hotel.getId());
    }

    @PutMapping()
    public void updateById(@RequestBody Hotel hotel) {
        if (hotel.getId() == null) {
            throw new InvalidParameterException("id不能为空");
        }
        hotelService.updateById(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);
    }
}

当然也是可以不使用注解来写,直接在配置文件中写队列绑定的交换机
RabbitMQ_小白要变大牛的博客-CSDN博客RabbitMqhttps://blog.csdn.net/hnhroot/article/details/125463948?spm=1001.2014.3001.5502

举报

相关推荐

0 条评论