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

 io.开云体育官方注册网址debezium debezium-api ${version.debezium}   io.debezium debezium-embedded ${version.debezium} 

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

 io.开云体育官方注册网址debezium  debezu -connector-mongodb ${version.debezium}  .debezium . io.debezium

//定义Debezium引擎与MySQL连接器的配置…开云体育官方注册网址final Properties props = 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("database. 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连接器的配置…开云体育官方注册网址final Properties props = config.asProperties();道具。setProperty(“名字”,“引擎”);props.setProperty(“connector.class”、“io.d开云体育官方注册网址ebezium.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("database. setproperty ")开云体育电动老虎机hostname"， "localhost") props.setProperty("开云体育电动老虎机数据库。port"， "3306") props.setProperty("开云体育电动老虎机数据库。props.setProperty("database. user")开云体育电动老虎机props.setProperty("database.server. passw开云体育电动老虎机ord"， "mysqlpw")props.setProperty("topic. id"， "85744")prop . setproperty ("schema.history.internal"， "io.debezium.storage.file.histo开云体育官方注册网址ry.FileSchemaHistory") props.setProperty("schema.history.internal.file.filename"， "/path/to/storage/schemahistory.dat")

//创建引擎try (开云体育官方注册网址DebeziumEngine>引擎= DebeziumEngine.create(Json.class) .using(props) .notify (record -> {System.out.println(record);}) .build()) {}

//异步运行引擎…ExecutorService executor = 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(); }

final Properties props = new Properties();．.．props.setProperty(“converter.schemas。启用”、“假”);// message中不包含schema…final 开云体育官方注册网址DebeziumEngine> engine = DebeziumEngine.create(Json.class) .using(props) .notifying((records, committer) -> {for (ChangeEvent r: records) {System.out. String>> engine = DebeziumEngine.create(Json.class) .using(props) .notifying((records, committer) -> {println(“关键= " + r.key () + " ' value = " + r.value () + "'");committer.markProcessed (r);}……

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

final Properties props = new Properties();．.．道具。setProperty(“转换”、“过滤器”);// (1)setProperty(“谓词”、“headerExists”);// (2) props.setProperty("predicates.headerExists. "类型”、“org.apache.kafka.connect.transforms.predicates.HasHeaderKey”);//(3) props.setProperty(" predicments . 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(Object)} *，当这批记录完成时调用{@link RecordCommitter#markBatchFinished()}。*/ void handleBatch(List records, RecordCommitter committer)抛出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();