1.死磕以太坊源码分析之Fetcher同步
2.mysql+canal+adapter+es实现数据同步
3.PyTorch 源码解读之 BN & SyncBN：BN 与 多卡同步 BN 详解
4.Flink mysql-cdc connector 源码解析
5.利用Kettle进行数据同步（下）
6.FlinkCDC数据实时同步Mysql到ES

死磕以太坊源码分析之Fetcher同步

       区块数据同步分为被动同步和主动同步，数据数据Fetcher负责被动同步，同步同步主要任务包括接收新区块广播并进行同步。计划计划新产生的源码源码用区块通过NewBlockHashesMsg 和 NewBlockMsg 进行传播，Fetcher对象通过接收这些消息发现新的数据数据区块信息。Fetcher在内部将同步过程分为几个阶段，同步同步c 软件注册源码并为每个阶段设置状态字段，计划计划用于记录阶段数据。源码源码用首先同步区块哈希，数据数据当接收到哈希时，同步同步会将哈希标记在远程节点上，计划计划并在本地数据库中查找是源码源码用否存在该哈希，若不存在，数据数据则放入unknown列表，同步同步之后通过channel通知本地fetcher模块请求该区块的计划计划header和body。fetcher模块根据接收的header和body状态，在fetching和completing列表中进行管理。当确认fetching和completing列表中不存在指定区块哈希时，将哈希放入到announced列表，并准备拉取header和body。fetcher模块通过fetchTimer周期性地从announced列表中选择区块哈希，进行header的拉取。拉取header时，选择要下载的区块，从announced转移到fetching中，并发送下载请求。header请求由远程节点通过GetBlockHeadersMsg处理，并返回给本地节点。header处理包括过滤和通知downloader对象。header过滤主要步骤涉及校验、过滤与本地数据库的不匹配块以及同步算法的header等。过滤后的header放入complete或incomplete列表。body同步的过程涉及从complete列表中选择哈希，进行同步body。body请求通过p.RequestBodies发送GetBlockBodiesMsg消息，并在downloader对象中处理。body过滤主要涉及过滤和同步逻辑，最终导入完整块到数据库。同步区块哈希和区块的整个流程涉及复杂的机制和逻辑，包括DOS攻击的防范、区块高度的限制、header和body的同步等，最终目标是确保本地区块链与远程节点保持同步状态。

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

       一、版本

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

       二、MySQL开启binlog

       配置MySQL服务器，启用binlog日志功能，有赞小程序源码确保数据变化得以记录。

       三、MySQL配置文件

       调整MySQL配置文件，确保binlog日志开启状态。

       四、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数据同步情况。针对关联表场景，进行新索引构建及数据插入，确保数据完整同步。

       十二、结论

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

PyTorch 源码解读之 BN & SyncBN：BN 与 多卡同步 BN 详解

       BatchNorm原理

       BatchNorm最早在全连接网络中提出，旨在对每个神经元的输入进行归一化操作。在卷积神经网络（CNN）中，这一原理被扩展为对每个卷积核的输入进行归一化，即在channel维度之外的所有维度上进行归一化。BatchNorm带来的优势包括提高网络的收敛速度、稳定训练过程、减少过拟合现象等。

       BatchNorm的数学表达式为公式[1]，引入缩放因子γ和移位因子β，作者在文章中解释了它们的作用。

       PyTorch中与BatchNorm相关的类主要位于torch.nn.modules.batchnorm模块中，包括如下的类：_NormBase、BatchNormNd。

       具体实现细节如下：

       _NormBase类定义了BN相关的一些属性。

       初始化过程。

       模拟BN的forward过程。

       running_mean、running_var的更新逻辑。

       γ、β参数的更新方式。

       BN在eval模式下的行为。

       BatchNormNd类包括BatchNorm1d、BatchNorm2d、BatchNorm3d，它们的区别在于检查输入的合法性，BatchNorm1d接受2D或3D的输入，BatchNorm2d接受4D的输入，BatchNorm3d接受5D的输入。

       接着，介绍SyncBatchNorm的实现。

       BN性能与batch size密切相关。在batch size较小的场景中，如检测任务，内存占用较高，单张显卡难以处理较多，导致BN效果不佳。SyncBatchNorm提供了解决方案，其原理是所有计算设备共享同一组BN参数，从而获得全局统计量。

       SyncBatchNorm在torch/nn/modules/batchnorm.py和torch/nn/modules/_functions.py中实现，前者负责输入合法性检查以及参数设置，后者负责单卡统计量计算和进程间通信。群机器人源码

       SyncBatchNorm的forward过程。

       复习方差计算方式。

       单卡计算均值、方差，进行归一化处理。

       同步所有卡的数据，得到全局均值mean_all和逆标准差invstd_all，计算全局统计量。

       接着，介绍SyncBatchNorm的backward过程。

       在backward过程中，需要在BN前后进行进程间通信。这在_functions.SyncBatchNorm中实现。

       计算weight、bias的梯度以及γ、β，进一步用于计算梯度。

Flink mysql-cdc connector 源码解析

       Flink 1. 引入了 CDC功能，用于实时同步数据库变更。Flink CDC Connectors 提供了一组源连接器，支持从MySQL和PostgreSQL直接获取增量数据，如Debezium引擎通过日志抽取实现。以下是Flink CDC源码解析的关键部分：

       首先，MySQLTableSourceFactory是实现的核心，它通过DynamicTableSourceFactory接口构建MySQLTableSource对象，获取数据库和表的信息。MySQLTableSource的getScanRuntimeProvider方法负责创建用于读取数据的运行实例，包括DeserializationSchema转换源记录为Flink的RowData类型，并处理update操作时的前后数据。

       DebeziumSourceFunction是底层实现，继承了RichSourceFunction和checkpoint接口，确保了Exactly Once语义。open方法初始化单线程线程池以进行单线程读取，run方法中配置DebeziumEngine并监控任务状态。值得注意的是，目前只关注insert, update, delete操作，表结构变更暂不被捕捉。

       为了深入了解Flink SQL如何处理列转行、与HiveCatalog的结合、JSON数据解析、DDL属性动态修改以及WindowAssigner源码，可以查阅文章。你的支持是我写作的动力，如果文章对你有帮助，请给予点赞和关注。

       本文由文章同步助手协助完成。地量选股源码

利用Kettle进行数据同步（下）

       上篇内容对基于kettle的数据同步工程的构建进行了介绍，entrypoint.kjb作为工程执行的入口。

       为了减少操作成本，并确保数据同步过程稳定、安全，需要从更高层次进行抽象，创建一个简单易用的系统。

       以下是应用截图：

       除了选择数据源和数据库，还增加了授权码，意味着只有授权范围内的用户才能使用该系统。

       由于是内部使用，授权用户尚未实现后台管理，直接在应用数据库中添加，选择的数据源和数据库都通过配置文件生成。

       文末会提供GitHub上的源码地址，有需要的读者可以进行二次开发。

       一、数据库设计

       数据库名称为kettle，目前包含两张表：

       1、授权用户表。表中记录的用户可以使用数据同步系统。

       2、同步记录表。记录用户的数据同步操作。

       二、程序设计

       系统简单实用，没有特别的设计。以下是重点说明的三点：

       1、数据源及其参数配置。

       在application.yml配置文件中，存在如下配置：

       使用了springboot的@ConfigurationProperties注解。

       其中的DBSetting定义如下：

       通过客户端传递的参数，可以定位到相应的参数设置。

       2、集成kettle的API。

       由于kettle相关jar包放在了自建的nexus私服上，因此如果使用maven管理jar包，需要在settings.xml配置文件中做一些修改：

       其中的mirrorOf节点添加了!pentaho-releases，表示排除pentaho-releases。

       然后，在springboot工程的pom.xml中指定pentaho-releases的url。

       接下来是核心的对接代码，具体可以参考工程源码。

       3、异步执行作业

       由于Job的执行时间可能会很长，主要取决于数据量，因此一个request的来回可能会导致TIMEOUT，需要改为异步模式。

       核心思想是：启动新的线程，客户端定时轮询执行结果。

       三、总结

       本文分两篇文章介绍了如何利用kettle进行数据同步，并实现一个简易的系统，以降低操作成本和出错率。

       介绍到此，如有疑问，请留言。

       欢迎fork我的工程代码。

FlinkCDC数据实时同步Mysql到ES

       当需要将数据库数据实时同步到其他系统，如Elasticsearch，一个高效的方法是利用Apache Flink的CDC（Change Data Capture）技术。Flink CDC通过监控数据库日志，捕获数据的增删改操作，并实时将这些变化数据传输到目标系统，满足高实时性的需求。Flink CDC凭借Flink的强大实时处理能力，支持集群部署和高可用性，且与MySQL、Oracle、MongoDB等主流数据库兼容，其Java实现为开发者提供了灵活的开发环境和源码可定制性。

       例如，通过Flink SQL，仅需寥寥几行代码就能实现MySQL数据到Elasticsearch的实时同步。首先，确保安装了相关的Flink和SQL插件，如flink-1..0和flink-sql-connector-组件。启动Flink后，通过窗口功能创建与MySQL的连接表，以及与Elasticsearch同步的表。接着编写SQL任务，任务运行后，MySQL的数据即可实时流入Elasticsearch。此外，Flink CDC还支持其他数据源，如Oracle、MongoDB等，可以灵活地通过Kafka等中间件进行进一步处理和分发。

       想了解更多关于Flink CDC的细节和使用方法，可以参考以下链接：

       Flink CDC官网

       Flink CDC GitHub仓库

       Flink官方文档

       通过以上Flink CDC的介绍，实时同步MySQL到Elasticsearch的任务变得简单而强大。

Linux 内核：RCU机制与使用

       在学习Linux源码时，遇到带有__rcu后缀的数据结构，引发对RCU机制的好奇。RCU（Read-Copy Update）是数据同步机制，主要用于优化链表遍历读取效率，避免锁竞争和内存延迟，适用于读多写少的场景，如文件系统中频繁查找定位目录而目录修改相对较少的情况。

       RCU机制通过在读取数据时不对链表加锁，允许多线程同时读取，但当线程尝试修改数据时，必须加锁以保证数据一致性。这种机制显著提升了性能，尤其在大量读取少量修改的场景中。

       在Linux内核源码中，RCU的详细文档和实现源码可于Documentation/RCU/目录下找到。Paul E. McKenney为主要实现者，并整理了相关文章和链接供参考。

       RCU解决了多个关键问题：如在读取过程中，另一个线程可能删除或插入节点，RCU通过宽限期确保数据的完整性和一致性；发布-订阅机制确保插入的节点在读取时得到完整引用；并保证链表遍历的一致性，避免中间断开。

       RCU基于读-拷贝修改原理，允许读者无锁访问数据，而写者在进行修改时，先复制数据结构副本，修改副本后，通过回调函数在适当时机完成数据结构的更新或释放。这个适当时机由内核自动确定。

       RCU的核心在于允许并行的读取操作，同时对写操作进行延迟处理，通过读者信号和垃圾收集器确保数据的一致性和安全性。与传统锁机制相比，RCU减少了锁竞争和内存延迟，提升了性能。

       RCU通过grace period（宽限期）和quiescent state（静默状态）机制，确保写操作在所有读操作完成后执行，从而避免了数据不一致问题。RCU的实现包括rcu_read_lock和rcu_read_unlock，用于管理读操作的临界区，以及synchronize_rcu用于挂起写操作，直到所有读操作结束。

       在使用RCU保护共享数据结构时，读者可以自由访问，无需加锁；而写者在访问数据时，先复制副本进行修改，最后通过回调函数在适当时机执行真正的修改操作。这种机制确保了数据的一致性和安全性，同时避免了锁竞争的性能开销。

       RCU通过一系列核心API，如rcu_read_lock和rcu_read_unlock，以及synchronize_rcu，实现了读操作的并发性和写操作的延迟处理。读者通过这些API进入读临界区，而写者通过synchronize_rcu挂起，直到所有读操作完成。

       在实际应用中，RCU允许在不牺牲性能的情况下，处理大量读取和少量写入的操作。例如，在系统调用审计、路由表维护等场景中，使用RCU可以显著提升性能，同时减少锁竞争和内存延迟的问题。

       RCU机制虽然提升了性能，但也存在内存占用和写操作开销等问题。在考虑使用RCU时，需要权衡其带来的性能提升与内存使用和写操作的复杂性。

Linux下rsync+sersync实现数据实时同步

       防止数据丢失，确保数据有备份，并且实时备份，是实时同步的目的。实时同步通过检测当前目录的变化并触发同步至远程服务器，以保证数据连续性，降低维护成本。

       sersync与rsync是常用的文件同步工具，两者结合实现高效实时数据同步，尤其适合需要实时备份或同步大量数据的环境。当sersync检测到文件变化时，自动调用rsync同步至远程服务器或备份服务器，减少数据传输，提高数据一致性和安全性。

       在配置sersync和rsync实现文件同步时，需要分别在两台服务器上进行设置。首先，确保服务器的防火墙已关闭，然后分别安装sersync和rsync。sersync可能不在官方软件仓库中，需从源代码或预编译的二进制文件安装。查看并修改notify参数，确保其值适当。接下来，编辑sersync的confxml.xml配置文件，设置监控目录、目标服务器信息和同步选项。配置完成后，启动sersync服务。

       对于目标服务器，确保rsync已安装。若未安装，可通过包管理器安装。在目标服务器上配置rsync，编辑/etc/rsyncd.conf或创建新配置文件，定义模块和同步选项。创建密码文件，并确保其权限严格，只有所有者可读写。启动rsync守护进程，完成同步配置。

       测试同步功能时，在源服务器的/home目录下创建用户目录和新建文件，检查目标服务器上是否实时同步显示。确保一切正常工作，以实现有效数据同步。

深度解析sync WaitGroup源码

       waitGroup

       waitGroup 是 Go 语言中并发编程中常用的语法之一，主要用于解决并发和等待问题。它是 sync 包下的一个子组件，特别适用于需要协调多个goroutine执行任务的场景。

       waitGroup 主要用于解决goroutine间的等待关系。例如，goroutineA需要在等待goroutineB和goroutineC这两个子goroutine执行完毕后，才能执行后续的业务逻辑。通过使用waitGroup，goroutineA在执行任务时，会在检查点等待其他goroutine完成，确保所有任务执行完毕后，goroutineA才能继续进行。

       在实现上，waitGroup 通过三个方法来操作：Add、Done 和 Wait。Add方法用于增加计数，Done方法用于减少计数，Wait方法则用于在计数为零时阻塞等待。这些方法通过原子操作实现同步安全。

       waitGroup的源码实现相对简洁，主要涉及数据结构设计和原子操作。数据结构包括了一个 noCopy 的辅助字段以及一个复合意义的 state1 字段。state1 字段的组成根据目标平台的不同（位或位）而有所不同。在位环境下，state1的第一个元素是等待线程数，第二个元素是 waitGroup 计数值，第三个元素是信号量。而在位环境下，如果 state1 的地址不是位对齐的，那么 state1 的第一个元素是信号量，后两个元素分别是等待线程数和计数值。

       waitGroup 的核心方法 Add 和 Wait 的实现原理如下：

       Add方法通过原子操作增加计数值。当执行 Add 方法时，首先将 delta 参数左移位，然后通过原子操作将其添加到计数值上。需要注意的是，delta 的值可正可负，用于在调用 Done 方法时减少计数值。

       Done方法通过调用 Add(-1)来减少计数值。

       Wait方法则持续检查 state 值。当计数值为零时，表示所有子goroutine已完成，调用者无需等待。如果计数值大于零，则调用者会变成等待者，加入等待队列，并阻塞自己，直到所有任务执行完毕。

       通过使用waitGroup，开发者可以轻松地协调和同步并发任务的执行，确保所有任务按预期顺序完成。这在多goroutine协同工作时，尤其重要。掌握waitGroup的使用和源码实现，将有助于提高并发编程的效率和可维护性。

       如果您对并发编程感兴趣，希望持续关注相关技术更新，请通过微信搜索「迈莫coding」，第一时间获取更多深度解析和实战指南。