11.5. 向 Debezium 连接器发送信号

信号数据收集或信号表会存储您发送到连接器的信号，以触发指定操作。信号表的结构必须符合以下标准格式：

您可以通过向源数据库提交标准 SQL DDL 查询来创建信号表。

以下示例显示了一个 CREATE TABLE 命令，它会创建一个三列 debezium_signal 表：

CREATE TABLE debezium_signal (id VARCHAR(42) PRIMARY KEY, type VARCHAR(32) NOT NULL, data VARCHAR(2048) NULL);

您可以通过创建带有日志信号类型的信号表条目来请求连接器来向 日志 添加条目。处理信号后，连接器会将指定的消息输出到日志。另外，您可以配置信号，以便生成的消息包含流协调。

{"message": "Signal message at offset {}"}

您可以通过创建一个带有 execute-snapshot 信号类型的信号来请求连接器来启动临时快照。处理信号后，连接器运行请求的快照操作。

与连接器首次启动后运行的初始快照不同，在连接器已经开始流更改事件后会在运行时发生临时快照。您可以随时启动临时快照。

临时快照可用于以下 Debezium 连接器：

{"data-collections": ["public.MyFirstTable", "public.MySecondTable"]}

{"type":"execute-snapshot","data": {"data-collections": ["public.MyFirstTable"], "type": "INCREMENTAL", "additional-conditions":[{"data-collection": "public.MyFirstTable", "filter":"color='blue' AND brand='MyBrand'"]}}

有关临时快照的更多信息，请参阅您的连接器文档中的 Snapshots 主题。

您可以停止以下 Debezium 连接器的临时快照：

{"type":"INCREMENTAL", "data-collections": ["public.MyFirstTable"]}

您必须指定信号 的类型。data-collections 字段是可选的。将 data-collections 字段留空，以请求连接器停止当前快照中的所有活动。如果您希望增量快照继续，但您想要从快照中排除特定的集合，请提供要排除的集合或正则表达式的名称列表。连接器处理信号后，增量快照会进行，但它从您指定的集合中排除数据。

增量快照是特定类型的临时快照。在增量快照中，连接器捕获您指定的表的基准状态，类似于初始快照。但是，与初始快照不同，增量快照会捕获块中的表，而不是一次捕获表。连接器使用水位线方法来跟踪快照的进度。

通过捕获块中指定表的初始状态，而不是在单个单体操作中捕获，与初始快照进程相比，增量快照具有以下优势：

您可以暂停以下 Debezium 连接器的增量快照：

您必须指定信号 的类型。data 字段将被忽略。

您可以为以下 Debezium 连接器恢复增量快照：

您必须指定信号 的类型。data 字段将被忽略。

有关增量快照的更多信息，请参阅您的连接器文档中的 Snapshots 主题。

您可以通过创建一个带有 execute-snapshot 信号类型和 data.type 的 signal 来请求连接器来启动临时 阻塞 快照。处理信号后，连接器运行请求的快照操作。

与连接器首次启动后运行的初始快照不同，在连接器停止以流传输数据库更改事件后，会在运行时发生临时阻止快照。您可以随时启动临时阻止快照。

以下 Debezium 连接器可以使用阻止快照：

  {"type": "blocking", "data-collections": ["schema1.table1", "schema1.table2"], "additional-conditions": [{"data-collection": "schema1.table1", "filter": "SELECT * FROM [schema1].[table1] WHERE column1 = 0 ORDER BY column2 DESC"}, {"data-collection": "schema1.table2", "filter": "SELECT * FROM [schema1].[table2] WHERE column2 > 0"}]}

{"type":"execute-snapshot","data": {"type": "blocking"}

有关阻塞快照的更多信息，请参阅您的连接器文档中的 Snapshots 主题。

自定义操作允许您扩展 Debezium 信号框架，以触发默认实现中不可用的操作。您可以将自定义操作与多个连接器一起使用。

要定义自定义信号操作，您必须定义以下接口：

@FunctionalInterface
public interface SignalAction<P extends Partition> {

    /**
     * @param signalPayload the content of the signal
     * @return true if the signal was processed
     */
    boolean arrived(SignalPayload<P> signalPayload) throws InterruptedException;
}

io.debezium.pipeline.signal.actions.SignalAction 会公开一个参数，它表示通过信号频道发送的消息有效负载。

在定义了自定义信号操作后，使用以下 SPI 接口使自定义操作可供信号机制使用： io.debezium.pipeline.signal.actions.SignalActionProvider.

public interface SignalActionProvider {

    /**
     * Create a map of signal action where the key is the name of the action.
     *
     * @param dispatcher the event dispatcher instance
     * @param connectorConfig the connector config
     * @return a concrete action
     */

    <P extends Partition> Map<String, SignalAction<P>> createActions(EventDispatcher<P, ? extends DataCollectionId> dispatcher, CommonConnectorConfig connectorConfig);
}

您的实施必须返回信号操作的映射。将 map 键设置为操作的名称。密钥用作信号 的类型。

自定义操作 Java 项目对 Debezium 核心模块有编译依赖项。在项目的 pom.xml 文件中包含以下编译依赖项：

<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-core</artifactId>
    <version>${version.debezium}</version> 1
</dependency>

在 META-INF/services/io.debezium.pipeline.signal.actions.SignalActionProvider 文件中声明您的供应商实现。