MySQL 数据实时推给三方业务？试试用 SeaTunnel 搞定 HTTP 接入

如果外部系统不能直连数据库只有调用 HTTP 服务才能接收数据SeaTunnel 就能搞定这篇教程带你从零搭建 MySQL 到 WEB 系统的 HTTP API 进行数据同步。环境准备一键安装安装必需插件编辑config/plugin_config文件加入两个核心连接器connector-http-baseconnector-cdc-mysql然后执行安装脚本sh bin/install-plugin.sh提示如果你网络不好也可以去 Maven 中央仓库手动下载对应版本的 JAR 包直接放进 connectors/目录。准备 MySQL 驱动mysql-connector-java-8.0.28.jar或其他 8.x 版本复制到 SeaTunnel 的lib/目录。配置 MySQL开启“数据源”实时同步的源头是 MySQL 的​Binlog​需要先打开它。找到 MySQL 的配置文件my.cnf或my.ini加入下面几行然后重启 MySQL 服务[mysqld]server-id 1 # 集群内唯一编号log-bin mysql-bin # 开启二进制日志binlog-format ROW # 必须设为ROW模式**架起接收端一个简单的 HTTP 服务**我们得有个地方接收数据这里用 Go 快速写一个。创建一个server.go文件package mainimport ( fmt io log net/http)// 处理所有HTTP请求打印接收到的数据func handler(w http.ResponseWriter, r *http.Request) { defer r.Body.Close() // 记得关闭请求体 body, err : io.ReadAll(r.Body) if err ! nil { msg : fmt.Sprintf(读取请求体失败: %v, err) http.Error(w, msg, http.StatusInternalServerError) fmt.Println(msg) return } if len(body) 0 { fmt.Printf(请求内容: %s\n, string(body)) fmt.Printf(数据大小: %d 字节\n, len(body)) } else { fmt.Println(请求体为空) } // 给客户端一个成功回复 w.WriteHeader(http.StatusOK) fmt.Fprintf(w, 数据接收成功! 大小: %d 字节。, len(body))}func main() { http.HandleFunc(/, handler) // 所有请求都交给handler处理 fmt.Println(HTTP服务启动端口 9090...) fmt.Println(你可以用curl测试: curl -X POST -d {\test\:123} http://localhost:9090/) if err : http.ListenAndServe(:9090, nil); err ! nil { log.Fatal(启动服务失败: , err) }}启动它go run server.go看到“HTTP 服务启动…”的提示就说明服务已经在本地 9090 端口待命了。准备测试数据在你的 MySQL 数据库里建一张测试表并插入两条初始数据CREATE TABLE post ( id int(11) NOT NULL, content varchar(50) DEFAULT NULL, author varchar(50) DEFAULT NULL, PRIMARY KEY (id)) ENGINEInnoDB DEFAULT CHARSETutf8mb4;INSERT INTO post VALUES (1, Mysql入门, 张三);INSERT INTO post VALUES (2, Http解析, 李四);核心编写 SeaTunnel 任务配置在job/目录下新建mysqlcdc_http.conf内容就是我们的同步脚本env { parallelism 1 job.mode STREAMING # 流式任务模式 checkpoint.interval 10000 # 10秒做一次检查点}source { MySQL-CDC { username root password root table-names [your_database.post] # 改成你的数据库和表名 url jdbc:mysql://你的MySQL地址:3306/your_database # 修改为你的连接信息 schema-changes.enabled true # 允许表结构变更 }}sink { Http { url http://你的服务地址:9090/ # 改成你的HTTP服务地址 headers { Acceptapplication/json, Content-Typeapplication/json;charsetutf-8 } }}注意把里面的数据库连接信息和 HTTP 地址替换成你自己的。启动同步见证时刻在 SeaTunnel 根目录下执行命令启动任务bin/seatunnel.sh --config job/mysqlcdc_http.conf -m local实时效果验证一旦任务启动成功你会立刻在刚才的 Go 服务窗口看到日志E:\go run server.goStarting HTTP server on port 9090...Send POST/PUT requests with body data to test.Example: curl -X POST -d {message:test} http://localhost:9090/Request Body Content: {id:1,content:Mysql入门,author:张三}Body Length: 50 bytesRequest Body Content: {id:2,content:Http解析,author:李四}Body Length: 49 bytes这说明启动时表中已有的两条数据已经被立刻捕获并推送到 HTTP 服务了为了验证实时性现在回到 MySQL插入一条新数据INSERT INTO post VALUES (3, SeaTunnel实时同步测试, 王五);数据就会打印出来Request Body Content: {id:3,content:SeaTunnel实时同步测试,author:王五}Body Length: 66 bytes综上这套从 MySQL CDC 到 SeaTunnel API Sink 再到 自定义 HTTP 服务 的实时数据链路。以后往别的系统实时发送数据不用直连数据库通过 HTTP 服务就搞定在跨业务跨部门之间的数据通讯多了一个选择。来源 | 数仓生态圈