如何从Quarkus消费Debezium主题中的消息？_python Kafka :如何从我停止的地方继续消费消息_如何从多分区的Kafka主题中按顺序(按时间戳的顺序)消费数据 - 腾讯云开发者社区

从Quarkus消费Debezium主题中的消息，可以使用Quarkus提供的Quarkus Kafka扩展来实现。

Quarkus是一种基于GraalVM和OpenJDK的Java框架，旨在优化Java应用程序的性能和内存占用。Debezium是一个开源的分布式事件捕获平台，可用于监听和捕获数据库更改事件。

要从Quarkus消费Debezium主题中的消息，您需要遵循以下步骤：

添加Quarkus Kafka扩展依赖：在您的Quarkus项目的pom.xml文件中，添加以下依赖：

<dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-kafka</artifactId>
</dependency>

创建一个Kafka消费者：使用Quarkus Kafka扩展提供的注解，在您的消费者类上创建一个Kafka消费者，示例如下：

import io.quarkus.kafka.client.serialization.JsonbDeserializer;
import io.quarkus.runtime.Startup;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.eclipse.microprofile.reactive.messaging.Incoming;

@Startup
public class MyKafkaConsumer {

    @Incoming("my-topic")
    public void consume(ConsumerRecord<String, MyMessage> record) {
        MyMessage message = record.value();
        // 处理消息
    }
}

配置Kafka连接信息：在您的应用配置文件（例如application.properties）中，配置Kafka连接信息，示例如下：

kafka.bootstrap.servers=my-kafka-server:9092

配置Debezium主题消费：如果要消费Debezium主题中的消息，需要添加Debezium连接信息和主题配置，示例如下：

mp.messaging.incoming.my-topic.connector=debezium
mp.messaging.incoming.my-topic.debezium.connector.class=io.debezium.connector.mysql.MySqlConnector
mp.messaging.incoming.my-topic.debezium.offset.storage=org.apache.kafka.connect.storage.FileOffsetBackingStore
mp.messaging.incoming.my-topic.debezium.offset.storage.file.filename=data/offsets.dat
mp.messaging.incoming.my-topic.debezium.offset.flush.interval.ms=60000
mp.messaging.incoming.my-topic.debezium.database.hostname=my-db-host
mp.messaging.incoming.my-topic.debezium.database.port=3306
mp.messaging.incoming.my-topic.debezium.database.user=my-db-username
mp.messaging.incoming.my-topic.debezium.database.password=my-db-password
mp.messaging.incoming.my-topic.debezium.database.dbname=my-db-name
mp.messaging.incoming.my-topic.debezium.database.server.id=1
mp.messaging.incoming.my-topic.debezium.database.server.name=my-server-name
mp.messaging.incoming.my-topic.debezium.table.include.list=my-table-name
mp.messaging.incoming.my-topic.debezium.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
mp.messaging.incoming.my-topic.debezium.transforms.unwrap.drop.tombstones=true
mp.messaging.incoming.my-topic.debezium.transforms.unwrap.delete.handling.mode=drop
mp.messaging.incoming.my-topic.value.deserializer=io.quarkus.kafka.client.serialization.JsonbDeserializer
mp.messaging.incoming.my-topic.key.deserializer=org.apache.kafka.common.serialization.StringDeserializer

在上述配置中，您需要根据实际情况修改Debezium的连接信息和主题配置。