Golang 如何解耦多层级 kafka consumer 比较好?

[Darain]

简单描述一下当前的业务设计:

1.主线程从 kafka 读消息 (message)

2.把 message 根据 id 对 partition 取模转发给 dispatcher (dispatcher 数量 = partiton 大小)

3.dispatcher 转发 message 给 concurrent handler (一个 dispatcher 对应多个 concurrent handler)

4.concurrent handler 内有若干 message handler. 每个 concurrent handler 处理一类表(分表)的消息.

5.message handler 内有一个 handler.

6.handler 是实现了 Handle() 方法的实际业务处理者.(一条 message 被一个 handler 处理)

dispatcher 的设计目的是为了维护 partition offset.

concurrent message handler 的设计目的是简单的内部扩容,

message handler 的设计目的是复用业务判断和错误处理, 让 handler 可以只写业务.

目前的主要问题是, 我都是用 NewXX() 和 Run()方法, 在每一个 Run() 方法调用它的子 handler 的 Run(). 在 NewXX()方法里调用子 handler 的 NewXX() 方法.

最后导致代码结构比较复杂, 而且很难理清. 不知道该怎么优化

或者应该看些什么书 /文章比较好. 感觉思路是对的, 就是代码写的太绕了

[madNeal]

为啥不用现成的轮子

https://github.com/segmentio/kafka-go

[superfat]

我最近也在用这个库做 consumer 相关开发