elasticsearch与mysql数据同步

news2024/12/24 9:34:30

目录

  • 数据同步
    • 一.思路分析
      • 1.同步调用
      • 2.异步通知
      • 3.监听binlog
      • 4.选择
    • 二.实现数据同步
      • 1.思路
      • 2.导入demo
      • 3.声明交换机、队列
        • 3.1引入依赖
        • 3.2 配置文件
        • 3.3 声明队列交换机名称
        • 3.4 声明队列交换机
      • 4.发送MQ消息
        • 4.1 事务配置类
        • 4.2 service 代码
      • 5.接收MQ消息

数据同步

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

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-HAdv7YDC-1679468296276)(assets/image-20210723214758392.png)]

一.思路分析

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

  • 同步调用
  • 异步通知
  • 监听binlog

1.同步调用

方案一：同步调用

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-2Kr0UnlI-1679466723120)(assets/image-20210723214931869.png)]

基本步骤如下：

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

2.异步通知

方案二：异步通知

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-H3Zr9Vqz-1679466723124)(assets/image-20210723215140735.png)]

流程如下：

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

3.监听binlog

方案三：监听binlog

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-KuKuKj7V-1679466723126)(assets/image-20210723215518541.png)]

流程如下：

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

4.选择

方式一：同步调用

  • 优点：实现简单，粗暴
  • 缺点：业务耦合度高

方式二：异步通知

  • 优点：低耦合，实现难度一般
  • 缺点：依赖mq的可靠性

方式三：监听binlog

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

二.实现数据同步

1.思路

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

步骤：

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

  • 声明exchange、queue、RoutingKey

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

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

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

2.导入demo

资料：资料在这里

导入资料提供的hotel-admin项目：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-NgNS6gN5-1679466723128)(assets/image-20210723220237930.png)]

运行后，访问 http://localhost:8099

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-EHjziqrE-1679466723129)(assets/image-20210723220354464.png)]

其中包含了酒店的CRUD功能：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-EesTGlEK-1679466723129)(assets/image-20210723220511090.png)]

3.声明交换机、队列

MQ结构如图：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-6IzJaY3o-1679466723130)(assets/image-20210723215850307.png)]

3.1引入依赖

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

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

3.2 配置文件

spring:
  rabbitmq:
    host: 192.168.1.100
    username: guest
    password: guest
    virtual-host: /

3.3 声明队列交换机名称

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

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";
}

3.4 声明队列交换机

在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);
    }
}

4.发送MQ消息

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

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-nelp7h1M-1679466723130)(assets/image-20210723221843816.png)]

4.1 事务配置类

保证Rabbitmq在提交事务后执行。事务控制的详情

package cn.itcast.hotel.config;

import com.sun.istack.internal.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

@Component("afterCommitExecutor")
public class AfterCommitExecutor extends TransactionSynchronizationAdapter implements Executor {
    private static final ThreadLocal<List<Runnable>> RUNNABLES = new ThreadLocal<List<Runnable>>();
    private ThreadPoolExecutor threadPool;

    private Logger logger = LoggerFactory.getLogger(AfterCommitExecutor.class);
    
    @PostConstruct
    public void init() {
        logger.debug("初始化线程池。。。");
        int availableProcessors = Runtime.getRuntime().availableProcessors();
        if (0 >= availableProcessors) {
            availableProcessors = 1;
        }
        int maxPoolSize = (availableProcessors > 5) ? availableProcessors * 2 : 5;
        logger.debug("CPU Processors :%s MaxPoolSize:%s", availableProcessors, maxPoolSize);
        threadPool = new ThreadPoolExecutor(
            availableProcessors,
            maxPoolSize,
            60,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<Runnable>(maxPoolSize * 2),
            Executors.defaultThreadFactory(),
            new RejectedExecutionHandler() {
                @Override
                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                    logger.debug("Task:%s rejected", r.toString());
                    if (!executor.isShutdown()) {
                        executor.getQueue().poll();
                        executor.execute(r);
                    }
                }
            }
        );
    }

    @PreDestroy
    public void destroy() {
        logger.debug("销毁线程池。。。");
        if (null != threadPool && !threadPool.isShutdown()) {
            threadPool.shutdown();
        }
    }

    @Override
    public void execute(@NotNull Runnable runnable) {
        if (!TransactionSynchronizationManager.isSynchronizationActive()) {
            runnable.run();
            return;
        }
        List<Runnable> threadRunnables = RUNNABLES.get();
        if (threadRunnables == null) {
            threadRunnables = new ArrayList<Runnable>();
            RUNNABLES.set(threadRunnables);
            TransactionSynchronizationManager.registerSynchronization(this);
        }
        threadRunnables.add(runnable);
    }

    @Override
    public void afterCommit() {
        logger.debug("事务提交完成处理 ... ");
        List<Runnable> threadRunnables = RUNNABLES.get();
        for (int i = 0; i < threadRunnables.size(); i++) {
            Runnable runnable = threadRunnables.get(i);
            try {
                threadPool.execute(runnable);
            } catch (RuntimeException e) {
                logger.error("", e);
            }
        }
    }

    @Override
    public void afterCompletion(int status) {
        logger.debug("事务处理完毕 .... ");
        RUNNABLES.remove();
    }
}

4.2 service 代码

 @Override
    @Transactional
    public boolean save(Hotel hotel) {
        logger.info("----- into  insert service -----");
        hotel.setId(Long.getLong(UUID.randomUUID().toString()));
        int i = hotelMapper.insert(hotel);
        afterCommitExecutor.execute(new Runnable() {
            @Override
            public void run() {
                rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,MqConstants.HOTEL_INSERT_KEY,hotel.getId().toString());
                logger.info("----- rabbitmq send message -----");
            }
        });


        if (1==i ||"1".equals(1) ){
            return true;

        }else {
            return false;
        }
    }

  
    @Override
    @Transactional
    public boolean updateById(Hotel hotel) {
        logger.info("----- into service -----");

        //修改DB的数据
        int i = hotelMapper.updateById(hotel);

        // 使用AfterCommitExecutor
        afterCommitExecutor.execute(new Runnable() {
            @Override
            public void run() {
                rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE, MqConstants.HOTEL_INSERT_KEY, hotel.getId().toString());
                logger.info("----- rabbitmq send message -----");
            }
        });

        logger.info("--------- out service ----------");

        if (1==i ||"1".equals(1) ){
            return true;

        }else {
            return false;
        }
    }

    @Override
    public void removeById(Long id) {
        logger.info("----- into service -----");

        //修改DB的数据
        int i = hotelMapper.deleteById(id);

        // 使用AfterCommitExecutor
        afterCommitExecutor.execute(new Runnable() {
            @Override
            public void run() {
                rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE, MqConstants.HOTEL_DELETE_KEY, id);
                logger.info("----- rabbitmq send message -----");
            }
        });

        logger.info("--------- out service ----------");


    }

5.接收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.准备Request
        DeleteRequest request = new DeleteRequest("hotel", id.toString());
        // 2.发送请求
        restHighLevelClient.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.发送请求
        restHighLevelClient.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 {

    @Autowired
    private 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);
    }
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/443035.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

数据分析----IQR（Interquartile Range）四分位距的理解与应用及Python实现

数据分析----IQR（Interquartile Range）四分位距的理解与应用及Python实现

【原文链接】数据分析----IQR&#xff08;Interquartile Range&#xff09;四分位距的理解与应用及Python实现 一、IQR&#xff08;Interquartile Range&#xff09;四分位距的含义 1.1 IQR的官方定义 IQR 是用于标记离群值的另一种稳健方法。用于检测离群值的 IQR&#xff…
阅读更多...
解放你的双手：自动化文档整理

解放你的双手：自动化文档整理

目录 引子: 应用场景&#xff1a; 源代码&#xff1a; 源代码说明&#xff1a; 效果如下所示&#xff1a; movefiletofolderbytype.py 引子: 例如&#xff0c;一个人可能会在计算机上存储大量的照片、视频和文档文件&#xff0c;这些文件可能散落在不同的文件夹中&#xf…
阅读更多...
1678_计算机架构黄金时代_文章阅读

1678_计算机架构黄金时代_文章阅读

全部学习汇总&#xff1a; GreyZhang/g_risc_v: Learning notes about RISC V. (github.com) 看了一份几年前的文章&#xff0c;觉得还是挺有收获的&#xff0c;因此做一个简单的整理。 对于架构有很大影响的主要考虑四点&#xff1a;专用硬件的实现、高安全性的要求、开放指令…
阅读更多...
原来这些功能才是blender大受欢迎的原因

原来这些功能才是blender大受欢迎的原因

作为全球最热门的免费开源3D创作套件&#xff0c;好用、易上手是Blender的公认优点。相比许多付费的三维软件来说&#xff0c;Blender在功能上是有过之无不及。除了在建模方面表现出色外&#xff0c;还提供了从模型雕刻、动画、材质、渲染、到音频处理、视频剪辑等一系列制作解…
阅读更多...
Nature子刊:HPV改变阴道菌群的机制

Nature子刊:HPV改变阴道菌群的机制

​ 研究背景 人类乳头瘤病毒(HPV)是最常见的性传播感染&#xff0c;影响全球3亿人以上。 虽然大多数感染像得了场“感冒”一样被免疫系统清除&#xff0c;或保持在无症状或潜伏状态&#xff0c;但致癌(高危)HPV株(如最明显的HPV16和18型)会导致宫颈鳞状上皮内病变[低级别(LSIL…
阅读更多...
MATLAB在逐渐被Python淘汰吗？

MATLAB在逐渐被Python淘汰吗？

Python和MATLAB都是常用的科学计算工具&#xff0c;但是它们有很多不同之处。 Python是一种通用编程语言&#xff0c;而MATLAB主要是用来做数值计算的。Python的基本数据类型和一般的编程语言一样普遍&#xff0c;但是离开了Numpy这个包&#xff0c;就不再有数组或者矩阵的数据…
阅读更多...
网络安全为什么缺人? 缺什么样的人

网络安全为什么缺人? 缺什么样的人

1.网络安全为什么缺人? 缺人的原因是有了新的需求。 以前的时候&#xff0c;所有企业是以产品为核心的&#xff0c;管你有啥漏洞&#xff0c;管你用户信息泄露不泄露&#xff0c;我只要做出来的产品火爆就行。 这一切随着《网络安全法》、《数据安全法》、《网络安全审查办法…
阅读更多...
Blender插件Lazy Viewport

Blender插件Lazy Viewport

目录 1.Lazy Viewport插件1.1 解压Lazy Viewport插件1.2 blender偏好设置1.3 打开插件1.4 安装插件1.5 勾选插件Lazy Viewport1.6 安装插件前1.7 安装插件后 1.Lazy Viewport插件 Blender 的一个简单插件&#xff0c;用于将标准 G、R、S 热键映射到视图工具&#xff0c;因此您…
阅读更多...
哪款无线洗地机最好用？好用的无线洗地机分享

哪款无线洗地机最好用？好用的无线洗地机分享

洗地机是近几年来比较火的家庭清洁工具&#xff0c;我从推出以来就一直在使用&#xff0c;这些年下来也使用过不少品牌的洗地机&#xff0c;其中有好用的&#xff0c;也有体验一般的。今天为大家分享几款个人使用下来感觉还不错的洗地机。希望对于同样在选购洗地机的你有所帮助…
阅读更多...
线程七大状态

线程七大状态

线程生命周期&#xff08;七大状态&#xff09; 新建状态&#xff08;New&#xff09;&#xff1a;当Java线程被创建时&#xff0c;它处于新建状态。此时&#xff0c;线程对象已被创建&#xff0c;但尚未启动。在这个状态下&#xff0c;线程并没有开始执行任何代码&#xff0c;…
阅读更多...
[Java]监听器（Listener）

[Java]监听器（Listener）

过滤器&#xff08;Filter&#xff09;https://blog.csdn.net/m0_71229255/article/details/130246404?spm1001.2014.3001.5501 一 : Listener监听器简述 监听器就是监听某个对象的的状态变化的组件 监听器的相关概念&#xff1a; 事件源&#xff1a; 被监听的对象 ----- 三…
阅读更多...
Jenkins安装maven integration plugin以及jenkins安装allure插件失败的解决方法

Jenkins安装maven integration plugin以及jenkins安装allure插件失败的解决方法

这里写目录标题 一、Jenkins安装maven integration plugin失败解决方法&#xff08;1&#xff09;修改系统时间&#xff08;2&#xff09;查看当前操作系统时间&#xff08;3&#xff09;防止出错先执行命令&#xff08;4&#xff09;修改系统时间&#xff08;5&#xff09;写入…
阅读更多...
Cocos Creator 源码解读：引擎启动与主循环

Cocos Creator 源码解读：引擎启动与主循环

前言 本文基于 Cocos Creator 2.4.3 撰写。 Ready? 不知道你有没有想过&#xff0c;假如把游戏世界比作一辆汽车&#xff0c;那么这辆“汽车”是如何启动&#xff0c;又是如何持续运转的呢&#xff1f; 如题&#xff0c;本文的内容主要为 Cocos Creator 引擎的启动流程和主…
阅读更多...
C# | 上位机开发新手指南（十一）压缩算法

C# | 上位机开发新手指南（十一）压缩算法

上位机开发新手指南&#xff08;十一&#xff09;压缩算法 文章目录 上位机开发新手指南&#xff08;十一&#xff09;压缩算法前言压缩算法的分类从数据来源角度分类流式压缩块压缩 从是否需要建立字典角度分类字典压缩无字典压缩 流式压缩与块压缩流式压缩的优势与劣势优势劣…
阅读更多...
各种开源协议介绍

各种开源协议介绍

世界上的开源许可证&#xff08;Open Source License&#xff09;大概有上百种&#xff0c;今天我们来介绍下几种我们常见的开源协议。大致有GPL、BSD、MIT、Mozilla、Apache和LGPL等。 Apache License Apache License&#xff08;Apache许可证&#xff09;&#xff0c;是Apac…
阅读更多...
O2OA (翱途) 平台 V8.0 即将亮相

O2OA (翱途) 平台 V8.0 即将亮相

亲爱的小伙伴们&#xff0c;O2OA (翱途) 平台开发团队经过几个月的持续努力&#xff0c;实现功能的新增、优化以及问题的修复。2023 年度 V8.0 版本将于近期正式发布。届时我们将会用文档或者视频的方式详细来介绍新增的功能和优化的亮点&#xff0c;欢迎大家一起来体验&#x…
阅读更多...
在Vue中将单独一张图片设为背景图并充满整个屏幕

在Vue中将单独一张图片设为背景图并充满整个屏幕

将单独一张图片设为背景图并充满整个屏幕 代码如下(在主div中添加样式) background: url("../xx/images/图片名字.jpg");//这里的地址是用你项目中图片所在的路径为准background-repeat: no-repeat;//将图片样式不重复background-size: 100% 100%; //设置图片大小po…
阅读更多...
YOLOv8 更换主干网络之 PP-LCNet

YOLOv8 更换主干网络之 PP-LCNet

《PP-LCNet: A Lightweight CPU Convlutional Neural Network》 论文地址：https://arxiv.org/abs/2109.15099 代码地址：https://github.com/ngnquan/PP-LCNet 我们提出了一种基于MKLDNN加速策略的轻量级CPU网络，名为PP LCNet，它提高了轻量级模型在多个任务上的性能。本文列…
阅读更多...
13、DRF实战总结：重写DRF的to_representation和to_internal_value方法的作用详解（附源码）

13、DRF实战总结：重写DRF的to_representation和to_internal_value方法的作用详解（附源码）

DRF的to_representation和to_internal_value是序列化和反序列化过程中最核心的方法&#xff0c;它们分别用于将数据对象转换成字典&#xff0c;和将字典转换成数据对象。 DRF所有序列化器类都继承了BaseSerializer类&#xff0c;通过重写该类的to_representation()和to_intern…
阅读更多...
Python ----＞＞ PiP 的重要性

Python ----＞＞ PiP 的重要性

我的个人博客主页&#xff1a;如果’真能转义1️⃣说1️⃣的博客主页 关于Python基本语法学习---->可以参考我的这篇博客&#xff1a;《我在VScode学Python》 Python是一种跨平台的计算机程序设计语言&#xff0c;是一个高层次的结合了解释性、编译性、互动性和面向对象的语…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023