文章目录
⛄引言一、思路分析⛅实现方式⚡框架选择
二、实现数据同步⌚需求分析⏰搭建环境⚡核心源码
三、测试四、源码获取⛵小结
⛄引言
本文参考黑马 分布式Elastic search
Elasticsearch是一款非常强大的开源搜索引擎，具备非常多强大功能，可以帮助我们从海量数据中快速找到需要的内容
一、思路分析
⛅实现方式
同步调用
方案一：同步调用
基本步骤如下：
hotel-demo对外提供接口，用来修改elasticsearch中的数据酒店管理服务在完成数据库操作后，直接调用hotel-demo提供的接口，
异步通知
方案二：异步通知
流程如下：
hotel-admin对mysql数据库数据完成增、删、改后，发送MQ消息hotel-demo监听MQ，接收到消息后完成elasticsearch数据修改
监听binlong
方案三：监听binlog
流程如下：
给mysql开启binlog功能mysql完成增、删、改操作都会记录在binlog中hotel-demo基于canal监听binlog变化，实时更新elasticsearch中的内容
⚡框架选择
方式一：同步调用
优点：实现简单，粗暴缺点：业务耦合度高
方式二：异步通知
优点：低耦合，实现难度一般缺点：依赖mq的可靠性
方式三：监听binlog
优点：完全解除服务间耦合缺点：开启binlog增加数据库负担、实现复杂度高
本次实现方式我们选择 以RabbitMQ 异步方式 搭载 SpringCloud Alibaba + Feign 实现。
二、实现数据同步
⌚需求分析
需求
实现酒店管理增删改查业务，已提供页面。 完成其数据发生增删改查操作时 同步 ElasticSearch
分析
我们采用分布式技术的方式来实现
框架采用 SpringCloud Alibaba、Nacos 、OpenFeign 远程调用、RabbitMQ 作为消息承载体承载数据、 Elastic Search 搜索引擎
⏰搭建环境
以下为模块概览
主要分为两大模块
完成酒店模块增删改查业务，引入MQ依赖，完成其向MQ的发送消息 此模块作为生产者完成ES-MQ模块，引入MQ、ES依赖，完成接受MQ的消息以及完成对ES的更新 此模块作为消费者
注意：Nacos需要自行下载，本项目依赖于Nacos注册中心， 运行起来后不影响后面的服务注册进nacos
本次所用到的 RabbitMQ、 ElasticSearch 均部署在 云服务器
MQ结构如图：
⚡核心源码
hotel-service 业务模块
导入hotel-service 核心代码，已完成基础的增删改查工作。 具体源码公众号搜索 程序员Bug终结者 回复 es 获取
ES模块引入依赖
声明队列交换机名称
public class MqConstants {
/**
* 交换机
*/
public static final String HOTEL_EXCHANGE = "hotel.topic";
/**
* 新增或修改的routing_key
*/
public static final String HOTEL_INSERT_KEY = "hotel.insert";
/**
* 删除的 routing_key
*/
public static final String HOTEL_DELETE_KEY = "hotel.delete";
}
hotel-service模块 发送消息
@RestController
@RequestMapping("hotel")
public class HotelController {
@Autowired
private HotelService hotelService;
@Resource
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
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);
}
}
ES模块接受消息
@Component
@Slf4j
public class MqConsumerListener {
@Resource
private HotelService hotelService;
/**
* 监听酒店新增或修改的业务
* @param id
*/
@RabbitListener(bindings = @QueueBinding(value = @Queue(value = MqConstants.HOTEL_INSERT_KEY, durable = "true"),
exchange = @Exchange(name = MqConstants.HOTEL_EXCHANGE, type = ExchangeTypes.DIRECT), key = MqConstants.HOTEL_INSERT_KEY))
public void listenHotelInsertOrUpdate(String id) throws IOException {
hotelService.insertById(id);
}
/**
* 监听酒店删除的业务
* @param id
*/
@RabbitListener(bindings = @QueueBinding(value = @Queue(value = MqConstants.HOTEL_DELETE_KEY, durable = "true"),
exchange = @Exchange(name = MqConstants.HOTEL_EXCHANGE, type = ExchangeTypes.DIRECT), key = MqConstants.HOTEL_DELETE_KEY))
public void listenHotelDelete(String id) throws IOException {
hotelService.deleteById(id);
}
}
核心方法实现
@Service
public class HotelService {
@Resource
private RestHighLevelClient client;
@Resource
private HotelClient hotelClient;
public void insertById(String id) {
try {
//1. 根据id查询酒店数据
Hotel hotel = hotelClient.findById(id);
HotelDoc hotelDoc = new HotelDoc(hotel);
//2. 准备Request
IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString());
//3. 准备DSL
request.source(JSON.toJSONString(hotelDoc), XContentType.JSON);
//4. 发送请求
client.index(request, RequestOptions.DEFAULT);
} catch (IOException e) {
e.printStackTrace();
}
}
public void deleteById(String id) {
try {
//1. 准备 Request
DeleteRequest request = new DeleteRequest("hotel", id);
// 2.发送请求
client.delete(request, RequestOptions.DEFAULT);
} catch (Exception e) {
e.printStackTrace();
}
}
}
三、测试
运行nacos
startup.cmd -m standalone
将hotel-service模块注册到nacos
访问页面，对酒店数据进行增删改查操作
将第一条信息价格修改为399
查看es中数据的变化
成功完成数据同步
四、源码获取
请联系 公众号 程序员Bug终结者 回复 es同步 获取源码及数据库文件
⛵小结
以上就是【Bug 终结者】对 分布式 SpringCloudAlibaba、Feign与RabbitMQ实现MySQL到ES数据同步 的简单介绍，ES搜索引擎无疑是最优秀的分布式搜索引擎，使用它，可大大提高项目的灵活、高效性！ 通过本文已了解 MySQL数据同步ES基本过程以及核心实现 技术改变世界！！！
如果这篇【文章】有帮助到你，希望可以给【Bug 终结者】点个赞，创作不易，如果有对【后端技术】、【前端领域】感兴趣的小可爱，也欢迎关注❤️❤️❤️ 【Bug 终结者】❤️❤️❤️，我将会给你带来巨大的【收获与惊喜】！
好文阅读
您阅读本篇文章共花了：
发表评论