微服务—ES数据同步

数据同步

问题分析

方案1. 同步调用

方案2. 异步通知

方案3. 监听binlog编辑

各方案对比

案例——利用MQ实现数据同步

步骤1. 导入hotel-admin项目

步骤2. 声明交换机、队列

步骤3. 发送MQ消息

步骤4. 接收MQ消息

步骤5. 测试同步功能

数据同步

elasticsearch中的数据是由我们通过mysql数据进行导入的，因此mysql数据发生改变时，elasticsearch中的数据也必须跟着改变，这就是elasticsearch与mysql之间的数据同步。

问题分析

在微服务中，负责酒店管理(操作mysql)的业务与负责酒店搜索(操作elasticsearch)的业务可能在两个不同的微服务上，那么数据同步该如何实现呢?

方案1. 同步调用

基本步骤：
hotel-demo对外提供接口，用来修改elasticsearch中的数据，酒店管理服务在完成数据库操作后，直接调用hotel-demo提供的接口。

方案2. 异步通知

基本步骤：
hotel-admin对mysql数据库数据完成增、删、改后，发送MQ消息，hotel-demo监听MQ，接收到消息后完成elasticsearch数据修改。

方案3. 监听binlog

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

各方案对比

方案一：同步调用
优点：实现简单，粗暴
缺点：业务耦合度高

方案二：异步通知
优点：低耦合，实现难度一般
缺点：依赖mq的可靠性

方案三：监听binlog
优点：完全解除服务间耦合
缺点：开启binlog增加数据库负担、实现复杂度高

案例——利用MQ实现数据同步

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

步骤1. 导入hotel-admin项目

导入课前资料提供的hotel-admin项目，注意修改数据库配置为本地数据库，运行并访问http://localhost:8099进入管理界面。

项目中包含了酒店的CRUD功能：

步骤2. 声明交换机、队列

MQ结构如图：

在hotel-admin、hotel-demo项目中做以下4步

1. 引入rabbitmq的依赖：

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

2. 设置rabbitmq配置：

3. 设置队列与交换机名称常量：

package cn.itcast.hotel.constants;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";
}

4. 声明队列与交换机名称：
队列与交换机的声明可以采用注解或者bean的方式，这里采用bean的方式进行声明。

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.TopicExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class MqConfig {@Beanpublic TopicExchange topicExchange(){return new TopicExchange(MqConstants.HOTEL_EXCHANGE, true, false);}@Beanpublic Queue insertQueue(){return new Queue(MqConstants.HOTEL_INSERT_QUEUE, true);}@Beanpublic Queue deleteQueue(){return new Queue(MqConstants.HOTEL_DELETE_QUEUE, true);}@Beanpublic Binding insertQueueBinding(){return BindingBuilder.bind(insertQueue()).to(topicExchange()).with(MqConstants.HOTEL_INSERT_KEY);}@Beanpublic Binding deleteQueueBinding(){return BindingBuilder.bind(deleteQueue()).to(topicExchange()).with(MqConstants.HOTEL_DELETE_KEY);}
}

步骤3. 发送MQ消息

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

步骤4. 接收MQ消息

hotel-demo接收到MQ消息要做的事情包括：
新增消息：根据传递的hotel的id查询hotel信息，然后新增一条数据到索引库
删除消息：根据传递的hotel的id删除索引库中的一条数据

1. 首先在hotel-demo的cn.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.准备RequestDeleteRequest 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包新增一个类

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 {@Autowiredprivate IHotelService hotelService;/*** 监听酒店新增或修改的业务* @param id 酒店id*/@RabbitListener(queues = MqConstants.HOTEL_INSERT_QUEUE)public void listenHotelInsertOrUpdate(Long id){hotelService.insertById(id);}/*** 监听酒店删除的业务* @param id 酒店id*/@RabbitListener(queues = MqConstants.HOTEL_DELETE_QUEUE)public void listenHotelDelete(Long id){hotelService.deleteById(id);}
}