在 Kafka 中为消息添加类型？

Question

我们开始在后端重新开发中使用 Kafka，并且有一个关于如何构造我们生成和使用的消息的快速问题。

假设我们有一个用户微服务来处理用户的 CRUD 操作。已提出的两种可能的结构是：

1) 四个kafka主题，每个操作一个。消息值将只包含执行操作所需的数据，即

topic: user_created
message value: {
    firstName: 'john'
    surname: 'smith'
}

topic: user_deleted
message value: c73035d0-6dea-46d2-91b8-d557d708eeb1 // A UUID

and so on

2) 用户相关事件的单个主题，在消息中带有属性描述要采取的操作以及所需的数据，即

// User created
topic: user_events
message value: {
    type: 'user_created',
    payload: {
        firstName: 'john'
        surname: 'smith'
    }
}

// User deleted
topic: user_events
message value: {
    type: 'user_deleted',
    payload: c73035d0-6dea-46d2-91b8-d557d708eeb1 // A UUID
}

我赞成所描述的第一个系统，尽管我对 Kafka 的经验不足使我无法有力地争论为什么。我们将非常重视更有经验的用户的任何意见。

Answer 1

Kafka 消息没有与之关联的类型。

对于每个事件类型的主题，您将不得不担心与从不同主题读取的同一实体相关的事件的排序。仅出于这个原因，我建议将所有事件放在同一个主题中。这样，客户只需使用一个主题即可完全跟踪每个实体的状态。

Answer 2

我最近在研究这种架构。

我们使用了一个 API 网关，它是与我们的前端（在我们的例子中是 ReactJS）通信的 Web 服务。此 API 网关使用 REST 协议。该微服务使用 Spring Boot 开发，在单独的线程中充当生产者和消费者。

1- 生成消息：向 Kafka 代理发送主题为 "producer_topic"

的消息

2-消费消息：收听来自 Kafka 的关于主题 "consumer_topic"

的传入消息

为了消费，有一个线程池处理传入的消息和执行服务，该服务侦听 Kafka 流并将消息处理分配给池中的线程。

最重要的是，有一个 DAO 微服务可以处理 kafka 消息并执行 CRUD 操作。

消息格式看起来很像您的第二种方法。

//content of messages in the consumer_topic
{
    event_type: 'delete'
    message: {
        first_name: 'John Doe'
        user_id: 'c73035d0-6dea-46d2-91b8-d557d708eeb1'
    }
}

这就是为什么我应该推荐你第二种方法。当您只处理一个主题的所有 crud 操作时，复杂性会降低。由于分区并行性，它非常快，您可以添加复制以提高容错能力。

第一种方法在非物质化和关注点分离方面听起来不错，但它并不是真正可扩展的。例如，假设您想添加额外的操作，这是一个要添加的主题。还要看复制。你将有更多的副本要做，我认为这很糟糕。

Answer 3

按照 Tom 的建议，请记住，即使您使用单个主题，您也可以选择多个分区以实现消费者的可扩展性。 Kafka 为您提供分区级别的排序，而不是主题级别的排序。这意味着您应该使用 "key" 来标识您正在创建、删除、更新的资源，以便与此 "key" 相关的消息始终位于同一分区中，因此顺序正确，否则即使单个主题可能会丢失在不同分区上发送消息的消息顺序。

Answer 4

Kafka 0.11 添加了 Message Headers，这是一种为消息 body 指示不同消息类型的简单方法，即使它们都使用相同的序列化程序。

https://cwiki.apache.org/confluence/display/KAFKA/KIP-82+-+Add+Record+Headers

在 Kafka 中为消息添加类型？

Add a type to messages in Kafka?

messaging

message-queue

apache-kafka