如何在单个 Postgres 服务器中为多个数据库使用嵌入式 Debezium？

假设我们有两个微服务：service_A和service_B.
每个人在单个 Postgres 服务器实例中都有自己的数据库（db_a和db_b分别）（这只是一个临时环境，所以我们没有集群）。

也有另一种服务，service_debezium（与嵌入式Debezium v1.6.1Final应监听的变化）db_a和db_b。所以基本上在这个服务中配置了两个 Debezium 引擎。

但不知何故service_debezium不能同时听db_a和db_b。由于某种原因，它只侦听其中一个，并且没有错误日志。

此外，如果我配置service_debezium（即它的 Debezium 引擎）来侦听db_a或db_b，它会按预期工作，因此我确定它们的配置属性是正确的，并且（当只有一个引擎时）一切正常。

1. 那么为什么我们不能使用多个 Debezium 引擎来侦听单个 Postgres 服务器中的多个数据库呢？我在这里缺少什么？
2. 我认为的另一种选择是只使用一个 Debezium 引擎来侦听该 Postgres 服务器实例中的所有数据库，但显然它database.dbname的配置需要它，所以我想首选的方法是为每个数据库定义一个新的 Debezium 引擎。那是对的吗？

以下是 中的 Debezium 配置service_debezium：

• db_a 配置豆：

  @Bean
  public io.debezium.config.Configuration dbAConnector() throws IOException {
    File offsetStorageTempFile = File.createTempFile("offsets_", ".dat");
    File dbHistoryTempFile = File.createTempFile("dbhistory_", ".dat");
    return io.debezium.config.Configuration.create()
        .with("name", "db_a_connector")
        .with("connector.class", "io.debezium.connector.postgresql.PostgresConnector")
        .with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")
        .with("offset.storage.file.filename", offsetStorageTempFile.getAbsolutePath())
        .with("offset.flush.interval.ms", "60000")
        .with("database.hostname", "localhost")
        .with("database.port", 5432)
        .with("database.user", "postgres")
        .with("database.password", "*****")
        .with("database.dbname", "db_a")
        .with("table.whitelist", "public.dummy_table,public.another_dummy_table")
        .with("plugin.name", "pgoutput")
        .with("slot.name", "db_a_connector")
        .with("database.server.name", "db_a_server")
        .with("database.history", "io.debezium.relational.history.FileDatabaseHistory")
        .with("database.history.file.filename", dbHistoryTempFile.getAbsolutePath())
        .build();
  }

• db_b 配置豆：

  @Bean
  public io.debezium.config.Configuration dbBConnector() throws IOException {
    File offsetStorageTempFile = File.createTempFile("offsets_", ".dat");
    File dbHistoryTempFile = File.createTempFile("dbhistory_", ".dat");
    return io.debezium.config.Configuration.create()
        .with("name", "db_b_connector")
        .with("connector.class", "io.debezium.connector.postgresql.PostgresConnector")
        .with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")
        .with("offset.storage.file.filename", offsetStorageTempFile.getAbsolutePath())
        .with("offset.flush.interval.ms", "60000")
        .with("database.hostname", localhost)
        .with("database.port", 5432)
        .with("database.user", "postgres")
        .with("database.password", "*****")
        .with("database.dbname", "db_b")
        .with("table.whitelist", "public.yet_another_table")
        .with("plugin.name", "pgoutput")
        .with("slot.name", "db_b_connector")
        .with("database.server.name", "db_b_server")
        .with("database.history", "io.debezium.relational.history.FileDatabaseHistory")
        .with("database.history.file.filename", dbHistoryTempFile.getAbsolutePath())
        .build();
  }

以下是我在 中创建 Debezium 引擎实例的方法service_debezium：

• db_a 听众：

@Component
public class DBAListener {

  public DBAListener(
      @Qualifier("dbAConnector") Configuration connectorConfiguration /*, ... other services */) {

    this.debeziumEngine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class))
        .using(connectorConfiguration.asProperties())
        .notifying(this::handleChangeEvent)
        .build();

    // ...
  }

  // ...

}

• 和类似的db_b听众：

@Component
public class DBBListener {

  public DBBListener(
      @Qualifier("dbBConnector") Configuration connectorConfiguration /*, ... other services */) {

    this.debeziumEngine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class))
        .using(connectorConfiguration.asProperties())
        .notifying(this::handleChangeEvent)
        .build();

    // ...
  }

  // ...

}

• postgresql.conf：

# ...other conf

wal_level = logical             # minimal, archive, hot_standby, or logical (change requires restart)
# max number of walsender processes (i.e. # of replica databases + # of Debezium engines) (change requires restart)
# Abrupt streaming client disconnection might cause an orphaned connection slot until a timeout is reached, 
# so this parameter should be set slightly higher than the maximum number of expected clients 
# so disconnected clients can immediately reconnect
max_wal_senders = 3
max_replication_slots = 3       # max number of replication slots (change requires restart)

提前致谢！

更新：根据@Yuri Niitsuma 的回答更新了代码片段以提供完整示例。