NestJs - Kafka 对有效负载进行过滤

【问题标题】：NestJs - Kafka filtering on payloadNestJs - Kafka 对有效负载进行过滤
【发布时间】：2021-03-14 12:58:08
【问题描述】：

我将 NestJs 与 KafkaJs 一起使用，我有两个微服务订阅相同的 EventPattern，但是其中一个微服务只需要 EventPattern 的单一类型的有效负载，例如：

EventPattern::DoThing

value: {
    jobType: "job1",
    otherStuff: {}
}

value: {
    jobType: "job2",
    otherStuff: {}
}

value: {
    jobType: "job3",
    otherStuff: {}
}

在微服务 1 中，我只关心 DoThing -> job1，但是
DoThing -> 需要job2、job3等等。

有没有一种方法可以从巢中进行过滤，而不是简单的 if 一旦消费完成。

欢迎任何建议。

【问题讨论】：

标签： apache-kafka microservices nestjs kafkajs

【解决方案1】：

看起来所有的 EventPattern 对象都属于一个主题。

为不同的微服务维护不同的消费者组，让微服务决定它必须处理哪些消息（使用KafkaJS，例子就像..）

const consumer = kafka.consumer({ groupId: 'microservice-1' })
await consumer.connect()
await consumer.subscribe({ topic: 'topic-A' })
await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
            if( shouldProcess(message.value) ) {
                // process them
            }
        })
    },
})

您可能需要将消息转换为 JSON 并检查 event.jobType 并进行相应处理。

如果您想避免每种作业类型出现if 条件，请为微服务将感兴趣的作业类型设置一个白名单（或）使用正则表达式来匹配作业类型。 p>

对另一个微服务重复相同的操作，除了 group.id 的变化

【讨论】：