PostgreSQL数据转Docker Kafka遇连接问题，求Spring Boot实现方案

一、Kafka Connect配置错误排查与修正 #

1. 核心逻辑错误：连接器类型搞反 #

你要把PostgreSQL的数据发送到Kafka，应该用JDBC Source Connector（从数据库读取数据到Kafka），但你配置的是JdbcSinkConnector（把Kafka数据写入数据库），这是完全反向的操作，是首要问题。

2. 容器网络无法访问宿主机PostgreSQL #

Docker容器内的localhost指向容器自身，而非宿主机的PostgreSQL服务，因此会出现连接拒绝错误。解决方法分场景：

• Windows/macOS（Docker Desktop）：用host.docker.internal替代localhost，这是Docker内置的宿主机映射地址
• Linux：要么启动Kafka Connect容器时添加--network host参数（让容器共享宿主机网络），要么直接使用宿主机的实际IP（如192.168.x.x）

3. SSL配置不匹配 #

如果你的PostgreSQL未开启SSL，配置里的sslmode=require会强制要求SSL连接导致失败，建议改为sslmode=disable或直接移除该参数。

修正后的连接器配置 #

{
  "name": "pg_source_connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "1",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "connection.url": "jdbc:postgresql://host.docker.internal:5432/Kafka_Example?sslmode=disable",
    "connection.user": "postgres",
    "connection.password": "你的密码",
    "table.whitelist": "messaggio",
    "mode": "timestamp",
    "max.retries": "4",
    "timestamp.column.name": "modified_at,created_at",
    "poll.interval.ms": "2000",
    "topic.prefix": "pg_source_"
  }
}

二、Spring Boot应用中的简便实现方法 #

方法1：Spring Data JPA + Spring Kafka 常规实现 #

适合简单场景，监听数据库实体的增删改事件：

• 引入依赖：spring-boot-starter-data-jpa、spring-boot-starter-kafka、postgresql
• 通过@TransactionalEventListener监听实体的创建/更新/删除事件，触发Kafka消息发送
• 示例代码：

@Service
public class MessageKafkaProducer {
    @Autowired
    private KafkaTemplate<String, Messaggio> kafkaTemplate;

    @TransactionalEventListener
    public void handleMessageCreated(CreationEvent<Messaggio> event) {
        Messaggio msg = event.getSource();
        kafkaTemplate.send("pg_source_messaggio", msg.getId().toString(), msg);
    }
}

方法2：Debezium Embedded CDC（变更数据捕获）实现 #

适合实时捕获数据库所有变更的场景，无需修改业务代码：

• 引入Debezium依赖：io.debezium:debezium-embedded:2.4.0.Final、io.debezium:debezium-connector-postgres:2.4.0.Final
• 配置Debezium引擎监听PostgreSQL的WAL日志，自动将变更事件发送到Kafka
• 核心配置示例：

@Configuration
public class DebeziumConfig {
    @Bean
    public Engine debeziumEngine(Configuration config, KafkaRecordConsumer consumer) {
        return Engine.create(Engine.class.getClassLoader())
                .using(config.asProperties())
                .notifying(consumer)
                .build();
    }

    @Bean
    public Configuration debeziumConfiguration() {
        return Configuration.create()
                .with("name", "pg-cdc-connector")
                .with("connector.class", "io.debezium.connector.postgresql.PostgresConnector")
                .with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")
                .with("offset.storage.file.filename", "/tmp/debezium/offsets.dat")
                .with("database.hostname", "localhost")
                .with("database.port", "5432")
                .with("database.user", "postgres")
                .with("database.password", "你的密码")
                .with("database.dbname", "Kafka_Example")
                .with("database.server.name", "pg-source")
                .with("table.include.list", "public.messaggio")
                .build();
    }
}

内容的提问来源于stack exchange，提问作者Alex