1、数据同步思路分析elasticsearch中的酒店数据来自于mysql数据库，因此mysql数据发生改变时，elasticsearch也必须跟着改变，这个就是elasticsearch与mysql之间的数据同步。
方案一：同步调用
基本步骤如下：
- hotel-demo对外提供接口，用来修改elasticsearch中的数据
- 酒店管理服务在完成数据库操作后，直接调用hotel-demo提供的接口
方案二：异步通知
流程如下：
- hotel-admin对mysql数据库数据完成增、删、改后，发送MQ消息
- hotel-demo监听MQ，接收到消息后完成elasticsearch数据修改
方案三：监听binlog
流程如下：
- 给mysql开启binlog功能
- mysql完成增、删、改操作都会记录在binlog中
- hotel-demo基于canal监听binlog变化，实时更新elasticsearch中的内容
方式一：同步调用
- 优点：实现简单，粗暴
- 缺点：业务耦合度高
方式二：异步通知
- 优点：低耦合，实现难度一般
- 缺点：依赖mq的可靠性
方式三：监听binlog
- 优点：完全解除服务间耦合
- 缺点：开启binlog增加数据库负担、实现复杂度高
导入项目完成对酒店数据的【增删改查】实现数据同步操作
hotel-admin项目作为酒店管理的微服务。当酒店数据发生增、删、改时，要求对elasticsearch中数据也要完成相同操作。
步骤：
-
创建hotel-admin项目，启动并测试酒店数据的CRUD
-
声明exchange、queue、RoutingKey
-
在hotel-admin中的增、删、改业务中完成消息发送
-
在hotel-demo中完成消息监听，并更新elasticsearch中数据
-
启动并测试数据同步功能
配置springboot核心配置文件ymlorg.springframework.boot spring-boot-starter-amqp
spring:
rabbitmq:
host: 192.168.26.131
port: 5672
username: itcast
password: 123321
virtual-host: /
在hotel-demo项目中定义一个类记录交换机和队列的名称
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";
public final static String HOTEL_INSERT_KEY = "hotel.insert";
public final static String HOTEL_DELETE_KEY = "hotel.delete";
}
在hotel-demo中，定义配置类，声明队列、交换机之间的绑定关系：
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 {
@Bean
public TopicExchange topicExchange() {
return new TopicExchange(MqConstants.HOTEL_EXCHANGE,true,false);
}
@Bean
public Queue insertQueue(){
return new Queue(MqConstants.HOTEL_INSERT_QUEUE, true);
}
@Bean
public Queue deleteQueue(){
return new Queue(MqConstants.HOTEL_DELETE_QUEUE, true);
}
@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);
}
}
2.2、hotel_admin声明交换机、队列发送更新信息
- 引入依赖
- 添加yml配置文件，连接队列
- 复制MqConstants类到constants包中
- 因为交换机和队列已经在消费者hotel_demo中创建了，接下来就在提供者中发送消息
因为队列是占内存的，所以我们直接传递一个id过去就可以了
@Autowired private RabbitTemplate rabbitTemplate; // 添加、修改数据中加入 rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,MqConstants.HOTEL_INSERT_KEY,hotel.getId()); // 删除类中加入 rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,MqConstants.HOTEL_DELETE_KEY,id);2.3、在hotel_demo消费者中监听队列消息
当我们监听到队列的信息后就需要对文档库等做一个操作，所以这里注入了IHotelService业务层的接口，下面就要创建这些接口的实现类
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 {
@Autowired
IHotelService hotelService;
@RabbitListener(priority = MqConstants.HOTEL_INSERT_QUEUE)
public void listenerHotelInsertOrUpdate(Long id){
// 调用修改
hotelService.insertById(id);
}
@RabbitListener(priority = MqConstants.HOTEL_DELETE_QUEUE)
public void listenerHotelInsertOrDelete(Long id){
// 调用删除
hotelService.deleteById(id);
}
}
2.4、创建监听调用的实现类
但提供者有消息发送过来的时候就会执行下面的两段代码
@Override
public void deleteById(Long id) {
try {
// 1、获取request
DeleteRequest request = new DeleteRequest("hotel").id(id.toString());
// 2、发送请求
client.delete(request,RequestOptions.DEFAULT);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
@Override
public void insertById(Long id) {
try {
// 0、根据id查询数据库数据【实现了MP接口可以直接调用】
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);
}
}
2.6、测试
-
重启服务消费者hotel_demo
查看管理页面队列和交换机绑定关系
交换机绑定
-
重启服务提供者hotel_admin
-
修改酒店消息然后返回刷新页面的消息，查看酒店信息是否修改了
