6.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 字段将被忽略。

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

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

与连接器在首次启动后运行的初始快照不同，在连接器停止到从数据库更改事件后，临时阻止快照发生。您可以在任何时间启动临时阻塞快照。

阻塞快照可用于以下 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"}

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

自定义操作允许您扩展 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 文件中声明您的供应商实现。