[大数据]ES和MYSQL实现数据同步

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

1.思路分析

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

• 同步调用

• 异步通知

• 监听binlog

1.1.同步调用

方案一：同步调用

基本步骤如下：

• hotel-demo对外提供接口，用来修改elasticsearch中的数据

• 酒店管理服务在完成数据库操作后，直接调用hotel-demo提供的接口，

1.2.异步通知

方案二：异步通知

流程如下：

• hotel-admin对mysql数据库数据完成增、删、改后，发送MQ消息

• hotel-demo监听MQ，接收到消息后完成elasticsearch数据修改

?1.3.监听binlog

方案三：监听binlog

流程如下：

• 给mysql开启binlog功能

• mysql完成增、删、改操作都会记录在binlog中

• hotel-demo基于canal监听binlog变化，实时更新elasticsearch中的内容

1.4.选择

方式一：同步调用

• 优点：实现简单，粗暴

• 缺点：业务耦合度高

方式二：异步通知

• 优点：低耦合，实现难度一般

• 缺点：依赖mq的可靠性

方式三：监听binlog

• 优点：完全解除服务间耦合

• 缺点：开启binlog增加数据库负担、实现复杂度高

2.实现数据同步

2.1.思路

以hotel-demo为例，当酒店数据发生增、删、改时，要求对elasticsearch中数据也要完成相同操作。

步骤：

• 导入hotel-admin项目，启动并测试酒店数据的CRUD

• 声明exchange、queue、RoutingKey

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

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

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

?

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

    @Autowired
    private IHotelService hotelService;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @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(Constants.HOTEL_EXCHANGE,Constants.HOTEL_INSERT_KEY,hotel);
    }

    @PutMapping()
    public void updateById(@RequestBody Hotel hotel){
        if (hotel.getId() == null) {
            throw new InvalidParameterException("id不能为空");
        }
        hotelService.updateById(hotel);
        rabbitTemplate.convertAndSend(Constants.HOTEL_EXCHANGE,Constants.HOTEL_INSERT_KEY,hotel);
    }

    @DeleteMapping("/{id}")
    public void deleteById(@PathVariable("id") Long id) {
        hotelService.removeById(id);
        rabbitTemplate.convertAndSend(Constants.HOTEL_EXCHANGE,Constants.HOTEL_DELETE_KEY,id);
    }
}

2.3.声明交换机、队列

MQ结构如图：

1）引入依赖

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

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

2）声明队列交换机名称

在hotel-admin和hotel-demo中的cn.demo.hotel.constatnts包下新建一个类MqConstants：

public class Constants {
    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.insert.queue";

    public final static String HOTEL_INSERT_KEY = "hotel.insert";

    public final static String HOTEL_DELETE_KEY = "hotel.delete";
}

2.4.发送MQ消息

在hotel-admin中的增、删、改业务中分别发送MQ消息：

注意：直接传输对象需要对实体类进行序列化和反序列化，否则报错

2.5.接收MQ消息

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

• 新增消息：根据传递的hotel的id查询hotel信息，然后新增一条数据到索引库

• 删除消息：根据传递的hotel的id删除索引库中的一条数据

1）首先在hotel-demo的cn.demo.hotel.service包下的IHotelService中新增新增、删除业务 ?

public void deleteById(Long id);

public void inseteOrUpdate(Hotel hotel);

2）给hotel-demo中的cn.demo.hotel.service.impl包下的HotelService中实现业务：

@Service
public class HotelService extends ServiceImpl<HotelMapper, Hotel> implements IHotelService {

    @Autowired
    private RestHighLevelClient client;

    @Override
    public PageResult search(HotelVo hotelVo) {
        try {
            SearchRequest request = new SearchRequest("hotel");

            buildBasicQuery(request,hotelVo);

            int voPage = hotelVo.getPage();
            int size = hotelVo.getSize();
            int page = (voPage - 1) * size;
            request.source().from(page).size(size);

            if (hotelVo.getLocation() != null && hotelVo.getLocation() != ""){
                request.source().sort(SortBuilders
                        .geoDistanceSort("location",new GeoPoint(hotelVo.getLocation()))
                        .order(SortOrder.ASC)
                        .unit(DistanceUnit.KILOMETERS));
            }

            SearchResponse response = client.search(request, RequestOptions.DEFAULT);
            return handleResponse(response);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void deleteById(Long id) {
        try {
            DeleteRequest request = new DeleteRequest("hotel",id.toString());

            client.delete(request,RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw  new RuntimeException(e);
        }
    }

    @Override
    public void inseteOrUpdate(Hotel hotel) {

        try {
            HotelDoc hotelDoc = new HotelDoc(hotel);

            IndexRequest request = new IndexRequest("hotel").id(hotelDoc.getId().toString());

            request.source(JSON.toJSONString(hotelDoc), XContentType.JSON);

            client.index(request,RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

    }

3）编写监听器，声明队列交换机，基于纯注解开发

在hotel-demo中的cn.demo.hotel.mq包新增一个类：

@Component
public class HotelListener {
    @Autowired
    private IHotelService service;

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = Constants.HOTEL_INSERT_QUEUE),
            exchange = @Exchange(name = Constants.HOTEL_EXCHANGE, type = ExchangeTypes.TOPIC),
            key = Constants.HOTEL_INSERT_KEY
    ))
    public void  inseteOrUpdateListen(Hotel hotel){

        service.inseteOrUpdate(hotel);

    }


    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = Constants.HOTEL_DELETE_QUEUE),
            exchange = @Exchange(name = Constants.HOTEL_EXCHANGE, type = ExchangeTypes.TOPIC),
            key = Constants.HOTEL_DELETE_KEY
    ))
    public void  deleteListen(Long id){

        service.deleteById(id);

    }

}

这样就实现了ES和Mysql的数据同步了

?

?