kafka 拉取数据到数据库 Java】

### 使用 Java 编写的 Kafka Consumer 将数据写入数据库 要实现从 Kafka 主题中消费数据并将数据存储到关系型数据库（如 PostgreSQL），可以按照以下方式编写代码。此过程涉及创建一个 Kafka 消费者实例，订阅主题，并将接收到的消息解析后插入到数据库。 以下是完整的解决方案： #### 1. 添加 Maven 依赖项 首先，在项目的 `pom.xml` 文件中添加必要的库： ```xml <dependencies> <!-- Kafka Client --> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>3.0.0</version> </dependency> <!-- PostgreSQL JDBC Driver --> <dependency> <groupId>org.postgresql</groupId> <artifactId>postgresql</artifactId> <version>42.5.0</version> </dependency> </dependencies> ``` #### 2. 创建 KafkaConsumer 和 DatabaseWriter 类 下面是一个示例程序，展示如何使用 Java 从 Kafka 中提取数据并将其保存到 PostgreSQL 数据库中。 ```java import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import java.sql.Connection; import java.sql.DriverManager; import java.sql.PreparedStatement; import java.time.Duration; import java.util.Collections; import java.util.Properties; public class KafkaToDatabase { public static void main(String[] args) { Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", "test-group"); props.put("enable.auto.commit", "true"); props.put("auto.offset.reset", "earliest"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) { consumer.subscribe(Collections.singletonList("your-topic-name")); while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { writeToDatabase(record.value()); } } } catch (Exception e) { System.out.println("Error occurred during consumption or writing to DB."); e.printStackTrace(); } } private static void writeToDatabase(String message) throws Exception { Connection conn = null; PreparedStatement pstmt = null; try { Class.forName("org.postgresql.Driver"); conn = DriverManager.getConnection( "jdbc:postgresql://localhost:5432/your_database", "username", "password" ); pstmt = conn.prepareStatement("INSERT INTO your_table (data_column) VALUES (?)"); pstmt.setString(1, message); pstmt.executeUpdate(); } finally { if (pstmt != null) { pstmt.close(); } if (conn != null) { conn.close(); } } } } ``` --- #### 关键点说明 1. **Kafka 配置属性** 上述代码设置了 Kafka 客户端的一些基本参数，例如服务器地址、组 ID 和序列化器类型[^1]。 2. **消费者轮询逻辑** 调用 `consumer.poll()` 方法定期获取来自指定主题的新记录。每次调用都会返回一批新的消息集合。 3. **数据库连接管理** 使用标准 JDBC API 来建立与 PostgreSQL 的连接，并执行 SQL 插入操作。确保关闭资源以释放内存。 4. **事务一致性** 如果需要更高的可靠性，则应采用两阶段提交协议或将偏移量存储在同一位置作为输出数据的一部分[^2]。 5. **实时流处理框架替代方案** 对于大规模生产环境下的复杂需求，建议利用 Apache Flink 或 Spark Streaming 等工具完成类似任务[^3]。 --- ###