Ene Kafka：Rust 的意见驱动 Kafka 框架

介绍

Ene kafka 是一个旨在简化 Rust 程序中与 Apache Kafka 交互的 Kafka 客户端。Ene kafka 的主要目标是使构建使用 Kafka 作为其分布式消息系统的驱动微服务的流程更容易、更快。

Ene Kafka 使用云事件来定义消费和生产的消息结构。

示例

创建一个事件

#[derive(KafkaMessage, Serialize, CloudEvent, Debug, Deserialize, DeserializeFrom)]
#[kafka(topic = "test", serde = Json, key = entity_id, headers = CloudEvent)]
#[cloud_event(
    content_type = "application/json",
    version = "1.0",
    event_type = "com.ene.entity.created.v1",
    event_source = "https://ene-kafka.com/docs/cloudevents/entity/created"
)]
struct EntityCreated {
    pub entity_id: i64,
    pub organisation_id: i64,
}

生产一个事件

let producer: KafkaProducer = kafka_producer!(bootstrap_servers = bootstrap_servers.clone());
    let event = EntityCreated {
        entity_id: 1755,
        organisation_id: 42,
    };

producer.send(event).await?;

消费一个事件

// Create a handler to process the event
#[derive(EventHandler)]
#[event_handler(event = EntityCreated, handler = handle_entity_created_event)]
struct EntityCreatedEventHandler {}

impl EntityCreatedEventHandler {
    async fn handle_entity_created_event(&self, event: &EntityCreated) -> ene_kafka::KafkaResult<()> {
        println!("EntityCreatedEventHandler: {:?}", event);
        // Do something with the event
        Ok(())
    }
}

// Create a consumer that listens to the topic and registers the handler
let consumer = kafka_consumer!(
    topic = KafkaTopic {
        name: "test".to_string(),
        content_type: ContentType::Json
    },
    // Dead letter queueing is included in the consumer!
    dlq_topic = KafkaTopic {
        name: "test-dlq".to_string(),
        content_type: ContentType::Json
    },
    consumer_group_id = "test-group",
    bootstrap_servers = bootstrap_servers,
    handlers = [
        EntityCreatedEventHandler -> EntityCreatedEventHandler {}
    ]
);

更多示例，请查看仓库中的 examples 目录。

特性

• 借助 Rust 宏的易用性：Ene Kafka 使用 Rust 宏来简化 Kafka 消息的定义以及它们的产生和消费。宏帮助抽象大量模板代码，使您能够专注于最重要的业务逻辑。

• 云事件：Ene Kafka 支持 CloudEvents 规范用于事件消息。

• 死信队列：Ene Kafka 支持死信队列，用于处理失败的消息。

• 自动（反）序列化：Ene Kafka 自动将消息序列化到指定的事件类型。

• 可扩展性：Ene Kafka 考虑到可扩展性进行设计（尽管这还是一个进行中的工作）。应该可以使用不同的底层 Kafka 客户端，或者使用其他序列化库代替 serde。

限制

• 仅支持 JSON 格式：Ene Kafka 目前仅支持 JSON 序列化和反序列化。目前没有支持 Avro 或 Protobuf。

• 仅支持 rdKafka Kafka 客户端

• Ene Kafka 基于云事件构建，可能不适合所有用例。

• 消费者和生产者配置有限：Ene Kafka 的一个目标是将 Kafka 的所有技术细节抽象化，以便快速、可靠地实现事件驱动微服务。不过，如果消费者和生产者配置能够更加灵活，那就更好了。这实际上很容易实现。

欢迎贡献

Ene Kafka 是一个开源项目，我们欢迎社区的贡献。如果您有任何改进或新功能的想法，请随时打开问题或拉取请求。