1.借助Canal实现MySQL数据库间链接canal链接mysql
2.canal部署安装使用
3.mysql+canal+adapter+es实现数据同步
4.mysql binlog同步工具binlogportal
5.Canal-adapter1.1.4集成Elasticsearch7.8.0排坑指南及在本地环境运行canal-adapter项目
6.JAVA云HIS医院管理系统源码：可医保对接的云HIS运维平台源码 SaaS模式

借助Canal实现MySQL数据库间链接canal链接mysql

       借助Canal实现MySQL数据库间链接

       在现代化的应用开发中，不同的系统或应用之间可能需要共享同一个数据库，这时候就需要实现数据库间的链接。本文将介绍如何借助Canal实现MySQL数据库间链接。Canal是阿里巴巴开源的一款数据库增量订阅和消费组件，支持MySQL、乳山溯源码Oracle等数据库，它可以将数据库更新的数据通过可靠的方式同步到其他数据存储、NoSQL等系统中。

       一、Canal介绍

       Canal是阿里巴巴开源的一款数据库增量订阅和消费组件，是基于MySQL数据库增量日志构建的，从而实现了与数据源(如MySQL)解耦，达到了异构神异的目的。Canal主要包括三个模块： Canal.Admin、Canal.Server和Canal.Client。

       Canal.Admin: Canal控制台管理界面，用于管理Canal的启停和监控

       Canal.Server: Canal的工作服务端，负责从数据源(如MySQL)订阅增量日志，并把日志传输给客户端

       Canal.Client: Canal的客户端，用于订阅和消费Canal.Server传输的数据

       二、Canal的使用场景

       Canal主要应用于以下场景：

       1、数据实时同步

       提供不同数据存储的数据实时同步，如 MySQL 到 Elasticsearch 的同步，实时更新数据，保持数据一致

       2、数据订阅

       对于需要全量数据同步的场景，结合 snapshot 快照机制，可以实现数据全量订阅

       3、实时数据分析

       对数据实时抓取，进行数据分析计算

       4、缓存更新

       将数据更新到Cache(如Redis)中，提升系统性能

       三、Canal的具体实现

       将A库的数据同步到B库中，具体实现如下：

       1、安装Canal

       Canal的安装需要先下载源码，然后进行编译打包，具体步骤可以参考Canal官网： /alibaba/canal

       2、ffmpeg源码解读配置Canal

       Canal的配置文件位于config文件夹下，通过修改canal.properties实现配置。

       （1）配置MySQL的主从关系

       # mysql主从地址信息

       canal.instance.master.address=.0.0.1:

       canal.instance.dbUsername=canal_test

       canal.instance.dbPassword=canal_test

       # 配置binlog信息，也可以从当前解析到的binlog中获取，

       # 优先从binlog position 获取，找不到才到 GTID_GET中获取, gtid模式推荐打开,

       # 当前的timestamp可以通过show master status或 show binary logs获取

       canal.instance.connectionCharset = UTF-8

       canal.instance.gtidon=on

       canal.instance.position =

       （2）配置Canal连接内容

       # 配置instance连接信息

       canal.instance.filter.regex=canal_test.tb_goods

       canal.instance.rds.accesskey=

       canal.instance.rds.secretkey=

       （3）配置数据输出方式

       # 配置数据输出方式

       canal.mq.topic=test

       # 指定数据传输格式

       canal.mq.flatMessage = false

       （4）配置kafka通常参数和账号信息

       # canal.mq.producerGroup 客户端group组名，同一个topic下的不同group组互不影响

       # canal.mq.servers 指定mq服务器的地址

       # canal.mq.topics 指定MQ topic主题名称

       # authAccount ，配置到应用已指定的账号

       canal.mq.properties.bootstrap.servers=...:

       canal.mq.producer.bootstrap-servers=...:

       canal.mq.producer.topic=myTest can

       （5）启动Canal

       执行bin目录下的startup脚本，即可启动Canal。

       3、配置Canal客户端

       在B库中新建表，同步A库的数据到该表中。

       （1）在B库中创建表

       mysql> create database canal_test2;

       mysql> use canal_test2;

       mysql> create table tb_goods(

        -> id int() not null auto_increment primary key,

        -> name varchar() not null,

        -> price int() not null

        -> )engine=innodb default charset=utf8;

       （2）在Canal服务端中新增instance，即配置同步关系

       mysql> create database canal_client;

       mysql> use canal_client;

       mysql> create table canal_client.tb_goods(

        -> id int() not null,

        -> name varchar() not null,

        -> price int() not null

        -> )engine=innodb default charset=utf8;

       mysql> GRANT ALL PRIVILEGES ON canal_client.* TO canal@’%’ IDENTIFIED BY ‘canal_test’ WITH GRANT OPTION;

       （3）在Canal客户端中启动Canal

       通过Canal客户端，将A库的数据同步到B库中的表中，具体代码实现如下：

       public class SimpleCanalClientExample {

        public static void mn(String[] args) {

        // 从控制台读取参数

        String host = args[0];

        int port = Integer.valueOf(args[1]);

        String destination = args[2];

        String username = args[3];

        String password = args[4];

        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(host,

        port), destination, username, password);

        int batchSize = ;

        try {

        connector.connect();

        connector.subscribe(“canal_test.tb_goods”);

        connector.rollback();

        while (true) {

        Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据

        long batchId = message.getId();

        int size = message.getEntries().size();

        if (batchId == -1 || size == 0) {

        Thread.sleep();

        } else {

        System.out.printf(“batchId: %s, size: %s \n”, batchId, size);

        printEntry(message.getEntries());

        }

        connector.ack(batchId);

        }

        } catch (Exception e) {

        e.printStackTrace();

        } finally {

        connector.disconnect();

        }

        }

        private static void printEntry(List entries) {

        for (CanalEntry.Entry entry : entries) {

        if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN

        || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND

        || entry.getEntryType() == CanalEntry.EntryType.HEARTBEAT) {

        continue;

        }

        RowChange rowChange = null;

        try {

        rowChange = RowChange.parseFrom(entry.getStoreValue());

        } catch (Exception e) {

        continue;

        }

        for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {

        if (rowChange.getEventType() == CanalEntry.EventType.DELETE) {

        printColumn(rowData.getBeforeColumnsList());

        } else if (rowChange.getEventType() == CanalEntry.EventType.INSERT) {

        printColumn(rowData.getAfterColumnsList());

        } else {

        printColumn(rowData.getBeforeColumnsList());

        System.out.println(“=======”);

        printColumn(rowData.getAfterColumnsList());

        }

        }

        }

        }

        private static void printColumn(List columns) {

        for (CanalEntry.Column column : columns) {

        System.out.println(column.getName() + “\t” + column.getValue() + “\t” + column.getUpdated());

        }

        }

       }

       通过以上代码，我们就可以将A库的tb_goods表中的数据实时同步到B库中的canal_client库的tb_goods表中。

       四、总结

       Canal是一款非常优秀的数据库增量订阅和消费组件，它可以很好地解决数据库间链接的问题，实现不同数据存储之间的数据同步。我们可以通过Canal的控制台管理界面，或者通过Canal的客户端代码实现数据库之间的数据同步。当然，Canal也有其缺点，就是在高并发场景下，可能会受到性能的限制，这需要我们在具体的应用场景中进行实际评估。

canal部署安装使用

       本篇文章旨在提供 Canal 部署、安装和使用的指导，适合在虚拟机、Docker 或 Kubernetes（k8s）环境中进行。

       在开始部署前，确保在数据库中创建同步账号，用于读取 binlog，并开启 binlog 模式为行模式，同时确保 server_id 不相同。考核系统 源码创建账号时，需确保其拥有全局的 SELECT, REPLICATION SLAVE, REPLICATION CLIENT 权限。

       下一步是下载 Canal 包，以版本 1.1.7 为例，进行解压。随后，修改配置文件，启动 Canal。通过检查日志，确认服务启动成功。

       对于 Docker 部署，首先安装 Docker 并优化镜像包地址（需确保联网环境）。接着下载 Canal 的源码包，完成 Docker 部署。

       k8s 部署时，编辑 YAML 文件，使用命令 `kubectl apply -f xxx.yaml` 启动 Canal。在部署过程中，需注意，客户端一旦中断，服务端会停止推送消息。此时，可进入容器，备份或删除 `/conf/xxx/meta.dat` 文件（根据实际情况替换文件名），重启 Canal 服务。对于 Docker 或 k8s 环境，可以简单地重启整个镜像。

       在遇到其他故障时，可参考“三方客户端Canal链接RDS获取binlog错位问题-云社区-华为云”进行故障排查。

mysql+canal+adapter+es实现数据同步

       一、版本

       MySQL+canal+adapter+es数据同步方案

       二、MySQL开启binlog

       配置MySQL服务器，启用binlog日志功能，确保数据变化得以记录。

       三、MySQL配置文件

       调整MySQL配置文件，确保binlog日志开启状态。maven打包 源码

       四、MySQL授权canal连接

       为canal连接MySQL服务器的账号分配权限，使其能作为MySQL的从服务器。

       五、下载canal及adapter

       下载并解压canal和adapter相关组件，检查目录结构。

       六、canal配置文件编辑

       在canal配置文件中，仅需调整关键配置项以满足同步需求。

       七、启动canal

       运行canal启动脚本，观察日志输出。若遇到报错，检查startup.sh脚本的-Xss参数设置，必要时调整参数。

       八、adapter配置与启动

       编辑adapter配置文件，确保与canal及es的映射关系正确。启动adapter后，检查日志，若未见输出，尝试调整-Xss虚拟机参数。

       九、解决adapter日志问题

       通过调整-Xss参数至k，解决了adapter启动无日志问题。若依然存在问题，则考虑调整源码。

       十、canal源码修改

       下载canal源码，修改相关配置项，如pom.xml文件，完成重新打包与编译，替换adapter插件中的jar文件。

       十一、测试与验证

       在MySQL中执行数据插入操作，验证adapter日志及ES数据同步情况。针对关联表场景，ce驱动源码进行新索引构建及数据插入，确保数据完整同步。

       十二、结论

       通过以上步骤，实现了MySQL数据通过canal和adapter同步至ES的目标，确保了数据的一致性与实时性。针对关联表的同步，需关注ES索引的创建与数据映射关系的正确性。

mysql binlog同步工具binlogportal

       在重构老系统的过程中，为了实时同步订单修改至重构表，binlog同步工具成为我们的首选方案。相较于一次性埋点，binlog的持久性和灵活性更具优势。

       调研中，我们考察了两个主要的binlog同步工具：canal和mysql-binlog-connector-java。canal以其稳定性和丰富的功能在业界广受好评，但在实际部署时，我们遇到了一些挑战。mysql-binlog-connector-java则是一个基础的binlog接收器，易于与springboot集成，但其功能较为基础，缺乏格式化处理和分布式部署等高级特性。

       鉴于这些需求，我们决定开发一个定制化的binlog同步工具——binlogportal。它的设计初衷是为了弥补现有工具的不足，具备高效、格式化和分布式支持的能力。在项目实施中，我们运用spring boot构建，以保证跨平台的兼容性。

       在分布式部署中，binlogportal利用redis实现高可用性。每个实例都会预先获取redis的分布式锁，在连接binlog前确保一致性。锁的过期时间会定时刷新，即使有实例宕机，其他实例也能通过保存的binlog信息无缝接续。项目源码已开源，地址为：github.com/dothetrick/b...

       这只是个人的学习与实践总结，如有任何改进或建议，欢迎在评论区交流。通过binlogportal，我们成功提升了订单同步的效率和可靠性。

Canal-adapter1.1.4集成Elasticsearch7.8.0排坑指南及在本地环境运行canal-adapter项目

       在集成canal的过程中，我遇到了众多问题，尽管网上有诸多解答，但质量不尽如人意。于是，我下载源码进行本地编译，逐一排查，总结出以下要点：

       以下是常见问题：

       1、如何使canal-adapter1.1.4支持ES7系列？

       2、常见错误信息

       3、canal-adapter1.1.4支持的具体版本号范围

       问题一：让canal-adapter支持ES7系列

       首先，下载canal对应版本的源码到本地，使用编码工具打开。由于canal1.1.4最高支持的版本是6.4.3，在canal-adapter的elasticsearch模块中，引用的ES版本号为6.4.3，因此需要将ES的依赖版本号升起来。

       修改完毕后，重新编译项目，会发现有几处代码编译报错。因为不同版本的ES的代码语法有所不同，只需要稍作改动即可。

       代码编译通过后，修改canal-adapter下的launcher模块中的application.yml文件，修改后的示例如下：

       修改完配置文件后，接下来配置数据库与ES索引的对应关系。位于elasticsearch模块下的资源文件目录下的es文件夹下，默认有3个文件。为了方便演示，先删除了两个文件。

       然后在ES中创建相应的mapping结构，用于将数据库数据同步到ES中。

       完成上述步骤后，即可启动canal-adapter本地项目。

       问题二：关于常见的报错信息

       canal-adapter在使用过程中，通常会遇到很多报错。以下逐一为大家解答：

       采坑点之一：在本地运行前一定先在maven的root模块下安装，安装完毕后再运行CanalAdapterApplication启动类。

       如果没有先安装直接运行，会出现报错，提示找不到OuterAdapter类的实现类。

       通过报错信息可以发现，当前提示是ESAdapter这个类找不到。根据抛出异常代码所在行通过源码打断点进一步排查，发现找不到target目录下的plugin目录下面的jar包。

       有两种方式可以解决这个问题，第一种是在canal-adapter项目的launcher模块下的main方法下面新建文件夹canal-adapter/plugin，将编译后的es的jar包放进去，然后修改源码中关于本地文件加载的路径。

       另外一种方法就是，运行前还是先使用maven的install安装一下。

       采坑点之二：报错信息Config dir not found

       在本地调试过程中，发现有报错Config dir not found。通过报错行打断点进一步排查，发现是项目启动完毕后在执行数据初始化阶段没有找到配置文件所导致的异常。

       这个问题也比较好解决，我们可以在canal-adapter的launcher模块的配置文件中新建一个叫es的文件夹，把elasticsearch模块下的es文件夹拷贝过来，即可解决这个问题。

       采坑点之三：报错Elasticsearch exception [type=index_not_found_exception, reason=no such index [XXXX]]

       这个问题，大家可以检查一下ES里面对应的索引名称是否存在，索引的mapping结构是否已经创建；当然，可能还有其他情况下导致出现这个问题，暂时没有遇到。

       采坑点之四：报错Not found the mapping info of index: XXX

       这个问题从报错信息来看，总感觉像是ES中索引的Mapping结构没有创建好。我用多种方式进行mapping结构的创建，可一直报错。

       根据报错堆栈信息，通过打断点的方式进一步排查，我们会看到在ESConnection类的行有这样一些被注释了的代码。

       这也正是canal-adapter1.1.4为什么不支持ES7以上的版本了。我们只需要将这些被注释的代码打开即可解决这个问题。

       通过上述代码的改造，我们可以对改完后的内容进行测试，全量同步数据和增量同步数据。

       canal-adapter为我们提供了全量同步数据的接口，我们在canal-adapter的launcher模块的com.alibaba.otter.canal.adapter.launcher.rest目录下可以看到有一个类叫做CommonRest，其里面提供全量同步数据的方法和条件同步数据的方法。

       直接使用postman发送如下请求即可完成数据的全量同步，效果如下，同时，如果数据库当前表的数据发生变更，canal-adapter也能及时监听到并同步到ES中。

       关于canal-adapter配置文件的，大家可以参考一下官网文档：github.com/alibaba/cana...

       另外还有一个网上经常提到的name: es6和es7，通过观察源码，在adapter1.1.4版本中，直接使用es即可。

       如上，canal-adapter1.1.4在本地运行起来了，并且全量同步数据和增量同步数据都已触发并生效。

       通过kibana也可以查询到对应的数据了。

       最后，这个项目在本地编译后在target目录下会生成一个canal-adapter的文件夹，这个文件夹可以拷贝出来直接运行。

       在windos和linux都可以运行。我这边编译后，在本地直接运行bat文件，程序正常并且可以正常全量同步数据和增量同步数据。

       不过遇到很奇怪的一个问题，将编译后的文件放在linux系统运行，则会不同的刷错误日志如下。

       暂时还未解决当前问题。不过我这边在目前的实际应用场景中，使用不到adapter，因为它的使用场景比较有效，对数据有较高的要求。

       这个问题在github上提了issues。

       地址：canal-adapter在本地环境可正常运行，编译后在服务器上运行出错；· Issue # · alibaba/canal

JAVA云HIS医院管理系统源码：可医保对接的云HIS运维平台源码 SaaS模式

       云HIS是专门为中小型医疗健康机构设计的云端诊所服务平台，提供内部管理、临床辅助决策、体检、客户管理、健康管理等全面解决方案。系统集成了多个大系统和子模块，助力诊所和家庭医生在销售、管理和服务等方面提升效率。

       基于SaaS模式的Java版云HIS系统，在公立二甲医院应用三年，经过多轮优化，运行稳定、功能丰富，界面布局合理，操作简单。

       系统融合B/S版电子病历系统，支持电子病历四级，拥有自主知识产权。

       技术细节方面，前端采用Angular+Nginx，后台使用Java+Spring、SpringBoot、SpringMVC、SpringSecurity、MyBatisPlus等技术。数据库为MySQL + MyCat，缓存为Redis+J2Cache，消息队列采用RabbitMQ，任务调度中心为XxlJob。接口技术包括RESTful API、WebSocket和WebService，报表组件为itext、POI和ureport2，数据库监控组件为Canal。

       云HIS系统对接医保流程包括准备阶段、技术对接阶段、业务协同阶段和后续维护与优化阶段。在准备阶段，需了解医保政策和要求，准备系统环境。在技术对接阶段，确定接口规范，开发医保接口，并进行测试和验证。在业务协同阶段，实现业务流程对接和数据同步。在后续维护与优化阶段，监控与故障处理，政策更新与适配，安全与保密工作。

       云HIS系统具有成本节约、高效运维、安全可靠和政策支持等优势，为医疗机构提供便捷、高效的医保服务。无论是大型三甲医院、连锁医疗集团还是中小型医疗机构，云HIS都是实现高效低成本云计算的最佳选择。