当前位置:   article > 正文

黑马es数据同步mq解决方案

黑马es数据同步mq解决方案

方式一:同步调用

        优点:实现简单,粗暴

        缺点:业务耦合度高

方式二:异步通知

        优点:低耦含,实现难度一般
        缺点:依赖mq的可靠性
方式三:监听binlog

        优点:完全解除服务间耦合

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

利用MQ实现mysql与elasticsearch数据同步

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

步骤:

导入课前资料提供的hotel-admin项目,启动并测试酒店数据的CRUD

1、声明exchange、queue、RoutingKey

  1. @Configuration
  2. public class MQConfig {
  3. @Bean
  4. public TopicExchange topicExchange(){
  5. return new TopicExchange(MqConstants.HOTEL_EXCHANGE,true,false);
  6. }
  7. @Bean
  8. public Queue insertQueue(){
  9. return new Queue(MqConstants.HOTEL_INSERT_QUEUE);
  10. }
  11. @Bean
  12. public Queue deleteQueue(){
  13. return new Queue(MqConstants.HOTEL_DELETE_QUEUE);
  14. }
  15. @Bean
  16. public Binding insertBinding(Queue insertQueue, TopicExchange topicExchange){
  17. return BindingBuilder.bind(insertQueue).to(topicExchange).with(MqConstants.HOTEL_INSERT_KEY);
  18. }
  19. @Bean
  20. public Binding deleteBinding(Queue deleteQueue, TopicExchange topicExchange){
  21. return BindingBuilder.bind(deleteQueue).to(topicExchange).with(MqConstants.HOTEL_DELETE_KEY);
  22. }
  23. }

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

  1. @PostMapping
  2. public void saveHotel(@RequestBody Hotel hotel){
  3. hotelService.save(hotel);
  4. rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,MqConstants.HOTEL_INSERT_KEY,hotel.getId());
  5. }
  6. @PutMapping()
  7. public void updateById(@RequestBody Hotel hotel){
  8. if (hotel.getId() == null) {
  9. throw new InvalidParameterException("id不能为空");
  10. }
  11. hotelService.updateById(hotel);
  12. rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,MqConstants.HOTEL_INSERT_KEY,hotel.getId());
  13. }
  14. @DeleteMapping("/{id}")
  15. public void deleteById(@PathVariable("id") Long id) {
  16. hotelService.removeById(id);
  17. rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,MqConstants.HOTEL_DELETE_KEY,id);
  18. }

3、在hotel-demo中完成消息监听,并更新elasticsearch中数据

  1. @Component
  2. public class HotelListener {
  3. @Autowired
  4. private IHotelService iHotelService;
  5. /**
  6. * 监听酒店新增或修改
  7. * @param id
  8. */
  9. @RabbitListener(queues = MqConstants.HOTEL_INSERT_QUEUE)
  10. public void listenHotelInsertOrUpdate(Long id){
  11. iHotelService.insertById(id);
  12. }
  13. /**
  14. * 监听酒店删除
  15. * @param id
  16. */
  17. @RabbitListener(queues = MqConstants.HOTEL_DELETE_QUEUE)
  18. public void listenHotelDelete(Long id){
  19. iHotelService.deleteById(id);
  20. }
  21. }
  1. @Override
  2. public void insertById(Long id) {
  3. try {
  4. //根据id查酒店数据
  5. Hotel hotel = getById(id);
  6. HotelDoc hotelDoc = new HotelDoc(hotel);//转换成文档接收的格式
  7. //1、准备Request对象
  8. IndexRequest request = new IndexRequest("hotel").id(hotelDoc.getId().toString());
  9. //2、准备Json文档
  10. request.source(JSON.toJSONString(hotelDoc),XContentType.JSON);//转换成JSON并发送
  11. //3、发送
  12. client.index(request,RequestOptions.DEFAULT);
  13. } catch (IOException e) {
  14. throw new RuntimeException(e);
  15. }
  16. }
  17. @Override
  18. public void deleteById(Long id) {
  19. try {
  20. DeleteRequest request = new DeleteRequest("hotel");
  21. request.id(id.toString());
  22. log.info("删除!"+id);
  23. client.delete(request,RequestOptions.DEFAULT);
  24. } catch (IOException e) {
  25. throw new RuntimeException(e);
  26. }
  27. }

4、启动并测试数据同步功能

        该项目分成两个部分,一个hotel-demo,一个hotel-admin,分别代表管理端和客户端。具体同步流程即:

        客户端 修改/新增/删除 某酒店时,发送id到mq对应操作的队列中,即上文的第二步

        客户端监听到消息后,便同步操作es,即上文的第三步

        过程收获:es和mysql一样都是数据库管理系统,只是在存储方式、数据类型等方面不一样。把这个过程理解为操作两个不同语言的数据库的同步操作,用mq作为中间件,就很好理解了。

        其中插入操作是相对有点麻烦的:

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

这段代码是用于创建一个索引请求(IndexRequest),用于将数据存储到名为"hotel"的Elasticsearch索引中。具体解释如下:

        1. IndexRequest("hotel"):表示要将数据索引到名为"hotel"的索引中。在Elasticsearch中,索引类似于数据库中的表,用于存储和组织数据。

        2. id(hotelDoc.getId().toString()):设置要存储的文档的唯一标识符,即文档的ID。这里使用hotelDoc对象的ID属性,并将其转换为字符串形式。在Elasticsearch中,每个文档都有一个唯一的ID,用于区分不同的文档。

        综合起来,这段代码的作用是创建一个索引请求,将hotelDoc对象(可能是一个酒店文档)存储到名为"hotel"的Elasticsearch索引中,并指定文档的唯一ID。

        此时如果对应id已存在数据,则覆盖

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/花生_TL007/article/detail/672024
推荐阅读
  

闽ICP备14008679号