开云体育官方注册网址Debezium引擎

<依赖> < groupId > io.debez开云体育官方注册网址ium < / groupId > < artifactId > debezium-api < / artifactId > <版本> $ {version.debezium} < /版本> < /依赖> <依赖> < groupId > io.debezium < / groupId > < artifactId > debezium-embedded < / artifactId > <版本> $ {version.debezium} < /版本> < /依赖>

<依赖> < groupId > io.debez开云体育官方注册网址ium < / groupId > < artifactId > debezium-connector-mysql < / artifactId > <版本> $ {version.debezium} < /版本> < /依赖>

<依赖> < groupId > io.debez开云体育官方注册网址ium < / groupId > < artifactId > debezium-connector-mongodb < / artifactId > <版本> $ {version.debezium} < /版本> < /依赖>

/ /定义Debezium引擎的配置与MySQL连接器…开云体育官方注册网址最终属性道具= config.asProperties ();道具。setProperty(“名字”,“引擎”);props.setProperty(“抵消。存储”、“org.apache.kafka.connect.storage.FileOffsetBackingStore”);props.setProperty (“offset.storage.file。文件名”、“/ tmp / offsets.dat”);props.setProperty (“offset.flush.interval.ms”、“60000”);/ *开始连接器属性* / props.setProperty(“数据库。开云体育电动老虎机主机名”、“localhost”);props.setProperty(“开云体育电动老虎机数据库。港”,“3306”); props.setProperty("database.user", "mysqluser"); props.setProperty("database.password", "mysqlpw"); props.setProperty("database.server.id", "85744"); props.setProperty("topic.prefix", "my-app-connector"); props.setProperty("schema.history.internal", "io.debezium.storage.file.history.FileSchemaHistory"); props.setProperty("schema.history.internal.file.filename", "/path/to/storage/schemahistory.dat"); // Create the engine with this configuration ... try (DebeziumEngine> engine = DebeziumEngine.create(Json.class) .using(props) .notifying(record -> { System.out.println(record); }).build() ) { // Run the engine asynchronously ... ExecutorService executor = Executors.newSingleThreadExecutor(); executor.execute(engine); // Do something else or wait for a signal or an event } // Engine is stopped when the main code is finished

/ /定义Debezium引擎的配置与MySQL连接器…开云体育官方注册网址最终属性道具= config.asProperties ();道具。setProperty(“名字”,“引擎”);props.setProperty(“连接器。类”、“io.debez开云体育官方注册网址ium.connector.mysql.MySqlConnector”);props.setProperty(“抵消。存储”、“org.apache.kafka.connect.storage.FileOffsetBackingStore”);props.setProperty (“offset.storage.file。文件名”、“/ tmp / offsets.dat”);props.setProperty (“offset.flush.interval.ms ", 60000);

/ *开始连接器属性* / props.setProperty(“数据库。开云体育电动老虎机主机名”、“localhost”) props.setProperty(“数据库。开云体育电动老虎机港”,“3306”)props.setProperty(“数据库。开云体育电动老虎机用户”、“mysqluser”) props.setProperty(“数据库开云体育电动老虎机。密码”、“mysqlpw”) props.setProperty (“databa开云体育电动老虎机se.server。id”,“85744”) props.setProperty(“主题。前缀”、“my-app-connector”) props.setProperty props.setProperty (“schema.history.i开云体育官方注册网址nternal”、“io.debezium.storage.file.history.FileSchemaHistory”) (“schema.history.internal.file.filename”、“/道路/ /存储/ schemahistory.dat”)

/ /创建的引擎配置…试(Deb开云体育官方注册网址eziumEngine < ChangeEvent <字符串,字符串> >引擎= DebeziumEngine.create (Json.class)型(道具).notifying(记录- > {System.out.println(记录);}).build ()) {}

/ /异步运行引擎…ExecutorService执行人= Executors.newSingleThreadExecutor ();executor.execute(引擎);/ /做其他事情或等待一个信号或一个事件

/ /稍后…engine.close ();

尝试{executor.shutdown ();而(!遗嘱执行人。awaitTermination(5, TimeUnit.SECONDS)) { logger.info("Waiting another 5 seconds for the embedded engine to shut down"); } } catch ( InterruptedException e ) { Thread.currentThread().interrupt(); }

最终属性道具= new属性();…props.setProperty (“converter.schemas。启用”、“假”);/ /不包括模式消息……最后Debe开云体育官方注册网址ziumEngine < ChangeEvent <字符串,字符串> >引擎= DebeziumEngine.create (Json.class)型(道具).notifying((记录,提交者)- > {(ChangeEvent <字符串,字符串> r:记录){system . out。println(“关键=”+ r.key () + " " value = ' " + r.value () + " ' ");committer.markProcessed (r);}……

最终属性道具= new属性();…道具。setProperty(“转换”、“过滤路由器”);/ / (1)props.setProperty (“transforms.router。类型”、“org.apache.kafka.connect.transforms.RegexRouter”);/ / (2)props.setProperty (“transforms.router。正则表达式"、" (. *)");/ / (3)props.setProperty (“transforms.router。替换”、“扶轮基金会1美元”);/ / (3)props.setProperty (“transforms.filter。类型”、“io.deb开云体育官方注册网址ezium.embedded.ExampleFilterTransform”); // (4)

最终属性道具= new属性();…道具。setProperty(“转换”、“过滤器”);/ /(1)道具。setProperty(“谓词”、“headerExists”);/ / (2)props.setProperty (“predicates.headerExists。类型”、“org.apache.kafka.connect.transforms.predicates.HasHeaderKey”);/ / (3)props.setProperty (“predicates.headerExists.name”、“header.name”);/ / (4)props.setProperty (“transforms.filter。type", "io.debezium.embedded.ExampleFilterTransform");// (5) props.setProperty("transforms.filter.predicate", "headerExists"); // (6) props.setProperty("transforms.filter.negate", "true"); // (7)

/ * * *处理一批记录,调用{@link RecordCommitter # markProcessed(对象)}*为每一个记录和{@link RecordCommitter # markBatchFinished()}当这批处理完成。* @param记录要处理的记录* @param提交者提交者,指示系统,我们完成* /空白handleBatch (< R >记录列表,RecordCommitter < R >提交者)抛出InterruptedException;

类MyChangeConsumer实现DebeziumEngine。开云体育官方注册网址ChangeConsumer> { public void handleBatch(List> records, RecordCommitter> committer) throws InterruptedException { ... } } // Create the engine with this configuration ... DebeziumEngine> engine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class)) .using(props) .notifying(new MyChangeConsumer()) .build();

类JsonChangeConsumer实现DebeziumEngine。开云体育官方注册网址ChangeConsumer> { public void handleBatch(List> records, RecordCommitter> committer) throws InterruptedException { ... } } // Create the engine with this configuration ... DebeziumEngine> engine = DebeziumEngine.create(Json.class) .using(props) .notifying(new MyChangeConsumer()) .build();