stm32f107ptp时钟同步

这是stm32f107ptp时钟同步程序&＃xff0c;官方库&＃xff0c;经过移植有用&＃xff0c;精度300ns左右&＃xff0c;建议使用v2版本&＃xff0c;主从机修改一下ip 什么的就可以使用了&＃xff0c;注意你自己使用的是和哪块评估板类似就在mdk里面选择对应版本&＃xff0c;否则会移植不成功

文件&＃xff1a;n459.com/f/25127180-480025005-211882
&＃xff08;访问密码&＃xff1a;551685&＃xff09;

以下内容无关&＃xff1a;

-------------------------------------------分割线---------------------------------------------

大数据领域数据源有业务库的数据&＃xff0c;也有移动端埋点数据、服务器端产生的日志数据。我们在对数据进行采集时根据下游对数据的要求不同&＃xff0c;我们可以使用不同的采集工具来进行。今天老刘给大家讲的是同步mysql增量数据的工具Canal&＃xff0c;本篇文章的大纲如下&＃xff1a;

Canal 的概念
mysql 中主备复制实现原理
Canal 如何从 MySQL 中同步数据
Canal 的 HA 机制设计
各种数据同步解决方法的简单总结
老刘争取用这一篇文章让大家直接上手 Canal 这个工具&＃xff0c;不再花别的时间来学习。

mysql 主备复制实现原理
由于 Canal 是用来同步 mysql 中增量数据的&＃xff0c;所以老刘先讲 mysql 的主备复制原理&＃xff0c;之后再讲 Canal 的核心知识点。

根据这张图&＃xff0c;老刘把 mysql 的主备复制原理分解为如下流程&＃xff1a;

主服务器首先必须启动二进制日志 binlog&＃xff0c;用来记录任何修改了数据库数据的事件。
主服务器将数据的改变记录到二进制 binlog 日志。
从服务器会将主服务器的二进制日志复制到其本地的中继日志&＃xff08;Relaylog&＃xff09;中。这一步细化的说就是首先从服务器会启动一个工作线程 I/O 线程&＃xff0c;I/O 线程会跟主库建立一个普通的客户单连接&＃xff0c;然后在主服务器上启动一个特殊的二进制转储&＃xff08;binlog dump&＃xff09;线程&＃xff0c;这个 binlog dump 线程会读取主服务器上二进制日志中的事件&＃xff0c;然后向 I/O 线程发送二进制事件&＃xff0c;并保存到从服务器上的中继日志中。
从服务器启动 SQL 线程&＃xff0c;从中继日志中读取二进制日志&＃xff0c;并且在从服务器本地会再执行一次数据修改操作&＃xff0c;从而实现从服务器数据的更新。
那么 mysql 主备复制实现原理就讲完了&＃xff0c;大家看完这个流程&＃xff0c;能不能猜到 Canal 的工作原理&＃xff1f;

Canal 核心知识点
Canal 的工作原理
Canal 的工作原理就是它模拟 MySQL slave 的交互协议&＃xff0c;把自己伪装为 MySQL slave&＃xff0c;向 MySQL master 发动 dump 协议。MySQL master 收到 dump 请求后&＃xff0c;就会开始推送 binlog 给 Canal。最后 Canal 就会解析 binlog 对象。

Canal 概念
Canal&＃xff0c;美[kəˈnæl]&＃xff0c;是这样读的&＃xff0c;意思是水道/管道/渠道&＃xff0c;主要用途就是用来同步 MySQL 中的增量数据&＃xff08;可以理解为实时数据&＃xff09;&＃xff0c;是阿里巴巴旗下的一款纯 Java 开发的开源项目。

Canal 架构

server 代表一个 canal 运行实例&＃xff0c;对应于一个 JVM。 instance 对应于一个数据队列&＃xff0c;1 个 canal server 对应 1…n 个 instance instance 下的子模块&＃xff1a;

EventParser:数据源接入&＃xff0c;模拟 salve 协议和 master 进行交互&＃xff0c;协议解析
EventSink&＃xff1a;Parser 和 Store 链接器&＃xff0c;进行数据过滤&＃xff0c;加工&＃xff0c;分发的工作
EventStore&＃xff1a;数据存储
MetaManager: 增量订阅&消费信息管理器
到现在 Canal 的基本概念就讲完了&＃xff0c;那接下来就要讲 Canal 如何同步 mysql 的增量数据。

Canal 同步 MySQL 增量数据
开启 mysql binlog
我们用 Canal 同步 mysql 增量数据的前提是 mysql 的 binlog 是开启的&＃xff0c;阿里云的 mysql 数据库是默认开启 binlog 的&＃xff0c;但是如果我们是自己安装的 mysql 需要手动开启 binlog 日志功能。

先找到 mysql 的配置文件&＃xff1a;

etc/my.cnf

server-id&＃61;1
log-bin&＃61;mysql-bin
binlog-format&＃61;ROW
这里有一个知识点是关于 binlog 的格式&＃xff0c;老刘给大家讲讲。

binlog 的格式有三种&＃xff1a;STATEMENT、ROW、MIXED

ROW 模式&＃xff08;一般就用它&＃xff09;

日志会记录每一行数据被修改的形式&＃xff0c;不会记录执行 SQL 语句的上下文相关信息&＃xff0c;只记录要修改的数据&＃xff0c;哪条数据被修改了&＃xff0c;修改成了什么样子&＃xff0c;只有 value&＃xff0c;不会有 SQL 多表关联的情况。

优点&＃xff1a;它仅仅只需要记录哪条数据被修改了&＃xff0c;修改成什么样子了&＃xff0c;所以它的日志内容会非常清楚地记录下每一行数据修改的细节&＃xff0c;非常容易理解。

缺点&＃xff1a;ROW 模式下&＃xff0c;特别是数据添加的情况下&＃xff0c;所有执行的语句都会记录到日志中&＃xff0c;都将以每行记录的修改来记录&＃xff0c;这样会产生大量的日志内容。

STATEMENT 模式

每条会修改数据的 SQL 语句都会被记录下来。

缺点&＃xff1a;由于它是记录的执行语句&＃xff0c;所以&＃xff0c;为了让这些语句在 slave 端也能正确执行&＃xff0c;那他还必须记录每条语句在执行过程中的一些相关信息&＃xff0c;也就是上下文信息&＃xff0c;以保证所有语句在 slave 端被执行的时候能够得到和在 master 端执行时候相同的结果。

但目前例如 step()函数在有些版本中就不能被正确复制&＃xff0c;在存储过程中使用了 last-insert-id()函数&＃xff0c;可能会使 slave 和 master 上得到不一致的 id&＃xff0c;就是会出现数据不一致的情况&＃xff0c;ROW 模式下就没有。

MIXED 模式

以上两种模式都使用。

Canal 实时同步
首先我们要配置环境&＃xff0c;在 conf/example/instance.properties 下&＃xff1a;

mysql serverId

canal.instance.mysql.slaveId &＃61; 1234
#position info&＃xff0c;需要修改成自己的数据库信息
canal.instance.master.address &＃61; 127.0.0.1:3306
canal.instance.master.journal.name &＃61;
canal.instance.master.position &＃61;
canal.instance.master.timestamp &＃61;
#canal.instance.standby.address &＃61;
#canal.instance.standby.journal.name &＃61;
#canal.instance.standby.position &＃61;
#canal.instance.standby.timestamp &＃61;
#username/password&＃xff0c;需要修改成自己的数据库信息
canal.instance.dbUsername &＃61; canal
canal.instance.dbPassword &＃61; canal
canal.instance.defaultDatabaseName &＃61;
canal.instance.connectionCharset &＃61; UTF-8
#table regex
canal.instance.filter.regex &＃61; .*\\…*
其中&＃xff0c;canal.instance.connectionCharset 代表数据库的编码方式对应到 java 中的编码类型&＃xff0c;比如 UTF-8&＃xff0c;GBK&＃xff0c;ISO-8859-1。

配置完后&＃xff0c;就要启动了
sh bin/startup.sh
关闭使用 bin/stop.sh
观察日志

一般使用 cat 查看 canal/canal.log、example/example.log

启动客户端

在 IDEA 中业务代码&＃xff0c;mysql 中如果有增量数据就拉取过来&＃xff0c;在 IDEA 控制台打印出来

在 pom.xml 文件中添加&＃xff1a;

public class Demo {
public static void main(String[] args) {
//创建连接
CanalConnector connector &＃61; CanalConnectors.newSingleConnector(new InetSocketAddress(“hadoop03”, 11111),
“example”, “”, “”);
connector.connect();
//订阅
connector.subscribe();
connector.rollback();
int batchSize &＃61; 1000;
int emptyCount &＃61; 0;
int totalEmptyCount &＃61; 100;
while (totalEmptyCount > emptyCount) {
Message msg &＃61; connector.getWithoutAck(batchSize);
long id &＃61; msg.getId();
List entries &＃61; msg.getEntries();
if(id &＃61;&＃61; -1 || entries.size() &＃61;&＃61; 0){
emptyCount&＃43;&＃43;;
System.out.println(“emptyCount : " &＃43; emptyCount);
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}else{
emptyCount &＃61; 0;
printEntry(entries);
}
connector.ack(id);
}
}
// batch -> entries -> rowchange - rowdata -> cols
private static void printEntry(List entries) {
for (CanalEntry.Entry entry : entries){
if(entry.getEntryType() &＃61;&＃61; CanalEntry.EntryType.TRANSACTIONBEGIN ||
entry.getEntryType() &＃61;&＃61; CanalEntry.EntryType.TRANSACTIONEND){
continue;
}
CanalEntry.RowChange rowChange &＃61; null;
try {
rowChange &＃61; CanalEntry.RowChange.parseFrom(entry.getStoreValue());
} catch (InvalidProtocolBufferException e) {
e.printStackTrace();
}
CanalEntry.EventType eventType &＃61; rowChange.getEventType();
System.out.println(entry.getHeader().getLogfileName()&＃43;” __ " &＃43;
entry.getHeader().getSchemaName() &＃43; " __ " &＃43; eventType);
List rowDatasList &＃61; rowChange.getRowDatasList();
for(CanalEntry.RowData rowData : rowDatasList){
for(CanalEntry.Column column: rowData.getAfterColumnsList()){
System.out.println(column.getName() &＃43; " - " &＃43;
column.getValue() &＃43; " - " &＃43;
column.getUpdated());
}
}
}
}
}
在mysql中写数据&＃xff0c;客户端就会把增量数据打印到控制台。
Canal 的 HA 机制设计
在大数据领域很多框架都会有 HA 机制&＃xff0c;Canal 的 HA 分为两部分&＃xff0c;Canal server 和 Canal client 分别有对应的 HA 实现&＃xff1a;

canal server&＃xff1a;为了减少对 mysql dump 的请求&＃xff0c;不同 server 上的 instance 要求同一时间只能有一个处于 running&＃xff0c;其他的处于 standby 状态。
canal client&＃xff1a;为了保证有序性&＃xff0c;一份 instance 同一时间只能由一个 canal client 进行 get/ack/rollback 操作&＃xff0c;否则客户端接收无法保证有序。
整个 HA 机制的控制主要是依赖了 ZooKeeper 的几个特性&＃xff0c;ZooKeeper 这里就不讲了。

Canal Server&＃xff1a;

canal server 要启动某个 canal instance 时都先向 ZooKeeper 进行一次尝试启动判断&＃xff08;创建 EPHEMERAL 节点&＃xff0c;谁创建成功就允许谁启动&＃xff09;。
创建 ZooKeeper 节点成功后&＃xff0c;对应的 canal server 就启动对应的 canal instance&＃xff0c;没有创建成功的 canal instance 就会处于 standby 状态。
一旦 ZooKeeper 发现 canal server 创建的节点消失后&＃xff0c;立即通知其他的 canal server 再次进行步骤 1 的操作&＃xff0c;重新选出一个 canal server 启动 instance。
canal client 每次进行 connect 时&＃xff0c;会首先向 ZooKeeper 询问当前是谁启动了 canal instance&＃xff0c;然后和其建立连接&＃xff0c;一旦连接不可用&＃xff0c;会重新尝试 connect。
canal client 的方式和 canal server 方式类似&＃xff0c;也是利用 ZooKeeper 的抢占 EPHEMERAL 节点的方式进行控制。
Canal HA 的配置&＃xff0c;并把数据实时同步到 kafka 中。

修改 conf/canal.properties 文件
canal.zkServers &＃61; hadoop02:2181,hadoop03:2181,hadoop04:2181
canal.serverMode &＃61; kafka
canal.mq.servers &＃61; hadoop02:9092,hadoop03:9092,hadoop04:9092
配置 conf/example/example.instance
canal.instance.mysql.slaveId &＃61; 790 /两台canal server的slaveID唯一
canal.mq.topic &＃61; canal_log //指定将数据发送到kafka的topic
数据同步方案总结
讲完了 Canal 工具&＃xff0c;现在给大家简单总结下目前常见的数据采集工具&＃xff0c;不会涉及架构知识&＃xff0c;只是简单总结&＃xff0c;让大家有个印象。

常见的数据采集工具有&＃xff1a;DataX、Flume、Canal、Sqoop、LogStash 等。

DataX (处理离线数据)
DataX 是阿里巴巴开源的一个异构数据源离线同步工具&＃xff0c;异构数据源离线同步指的是将源端数据同步到目的端&＃xff0c;但是端与端的数据源类型种类繁多&＃xff0c;在没有 DataX 之前&＃xff0c;端与端的链路将组成一个复杂的网状结构&＃xff0c;非常零散无法把同步核心逻辑抽象出来。

为了解决异构数据源同步问题&＃xff0c;DataX 将复杂的网状的同步链路变成了星型数据链路&＃xff0c;DataX 作为中间传输载体负责连接各种数据源。

所以&＃xff0c;当需要接入一个新的数据源的时候&＃xff0c;只需要将此数据源对接到 DataX&＃xff0c;就可以跟已有的数据源做到无缝数据同步。

DataX本身作为离线数据同步框架&＃xff0c;采用Framework&＃43;plugin架构构建。将数据源读取和写入抽象成为Reader/Writer插件&＃xff0c;纳入到整个同步框架中。

Reader: 它为数据采集模块&＃xff0c;负责采集数据源的数据&＃xff0c;将数据发送给Framework。
Writer: 它为数据写入模块&＃xff0c;负责不断向Framework取数据&＃xff0c;并将数据写入到目的端。
Framework&＃xff1a;它用于连接Reader和Writer&＃xff0c;作为两者的数据传输通道&＃xff0c;并处理缓冲、并发、数据转换等问题。
DataX的核心架构如下图&＃xff1a;

核心模块介绍&＃xff1a;

DataX完成单个数据同步的作业&＃xff0c;我们把它称之为Job&＃xff0c;DataX接收到一个Job之后&＃xff0c;将启动一个进程来完成整个作业同步过程。
DataX Job启动后&＃xff0c;会根据不同的源端切分策略&＃xff0c;将Job切分成多个小的Task(子任务)&＃xff0c;以便于并发执行。
切分多个Task之后&＃xff0c;DataX Job会调用Scheduler模块&＃xff0c;根据配置的并发数据量&＃xff0c;将拆分成的Task重新组合&＃xff0c;组装成TaskGroup(任务组)。每一个TaskGroup负责以一定的并发运行完毕分配好的所有Task&＃xff0c;默认单个任务组的并发数量为5。
每一个Task都由TaskGroup负责启动&＃xff0c;Task启动后&＃xff0c;会固定启动Reader->Channel->Writer的线程来完成任务同步工作。
DataX作业运行完成之后&＃xff0c;Job监控并等待多个TaskGroup模块任务完成&＃xff0c;等待所有TaskGroup任务完成后Job成功退出。否则&＃xff0c;异常退出。
Flume&＃xff08;处理实时数据&＃xff09;

Flume主要应用的场景是同步日志数据&＃xff0c;主要包含三个组件&＃xff1a;Source、Channel、Sink。

Flume最大的优点就是官网提供了丰富的Source、Channel、Sink&＃xff0c;根据不同的业务需求&＃xff0c;我们可以在官网查找相关配置。另外&＃xff0c;Flume还提供了自定义这些组件的接口。

Logstash&＃xff08;处理离线数据&＃xff09;

Logstash就是一根具备实时数据传输能力的管道&＃xff0c;负责将数据信息从管道的输入端传输到管道的输出端&＃xff1b;与此同时这根管道还可以让你根据自己的需求在中间加上过滤网&＃xff0c;Logstash提供了很多功能强大的过滤网来满足各种应用场景。

Logstash是由JRuby编写&＃xff0c;使用基于消息的简单架构&＃xff0c;在JVM上运行。在管道内的数据流称之为event&＃xff0c;它分为inputs阶段、filters阶段、outputs阶段。

Sqoop&＃xff08;处理离线数据&＃xff09;

Sqoop是Hadoop和关系型数据库之间传送数据的一种工具&＃xff0c;它是用来从关系型数据库如MySQL到Hadoop的HDFS从Hadoop文件系统导出数据到关系型数据库。Sqoop底层用的还是MapReducer&＃xff0c;用的时候一定要注意数据倾斜。

总结
老刘本篇文章主要讲述了Canal工具的核心知识点及其数据采集工具的对比&＃xff0c;其中数据采集工具只是大致讲了讲概念和应用&＃xff0c;目的也是让大家有个印象。老刘敢做保证看完这篇文章基本等于入门&＃xff0c;剩下的就是练习了。

好啦&＃xff0c;同步mysql增量数据的工具Canal的内容就讲完了&＃xff0c;尽管当前水平可能不及各位大佬&＃xff0c;但老刘会努力变得更加优秀&＃xff0c;让各位小伙伴自学从此不求人&＃xff01;