Kafka调研（3）：多点通信方案的设计-2

发表于 更新于 分类于 ， 本文字数： 5.5k 阅读时长 ≈ 5 分钟

1. 星形拓扑的通信模型

星形的意思是说，有一个消息接收和转发中心，应用两两之间并不直接通信。所有的应用层消息都通过Kafka集群进行中转。但是对Kafka来说，消息的机制是推拉机制。如果需要接收消息，需要应用层连接到集群，订阅感兴趣的topic，然后poll消息。

为了实现这种模型，
1. 我们利用Kafka本身的ACL控制机制来对topic权限进行控制。

  • 对每一个应用分配唯一的Consumer Group。
  • 对每一个接入的应用分配唯一的账户和密码。
  • 对每一个接入的应用分配SSL证书
  • 对每一个topic按照应用进行配置读写权限。
  • 对每一个接入的应用分配一个唯一的秘钥（该秘钥通过安全方式传递）。

2. 同时我们在应用包装一层API进行消息分派机制

  • 应用层的消息需要进行加密。
  • 应用层需要校验消息的来源和完整性。
  • 判断该消息是否应该被处理，比如消息的接收方是否包含了自己。
  • 应用层需要按照消息的业务ID（或者其他机制）分派合适的处理函数。

3. 星形拓扑通信模型：

2. 应用层框架能力分析

Kafka Client API[1]提供了功能丰富的接口, 同时通过Properties进行不同的配置。 Consumer API不是线程安全的。但是实际在应用过程中，一般来说会固定一套配置，同时希望应用层只需要配置topic行，其他的细节不需要关心。

经过考虑，该框架需要支持的能力如下：
应用层：

  • 提供一套简化的API支持Kafka消息的发送和接收。
  • 和Spring进行集成，服务通过@Service的API进行暴露给应用层。
  • 提供Request-Reply语义的接口（其中请求消息和响应的消息的匹配规则可以自定义）。
  • 自动编译protobuf IDL定义并生成相应的JAR包。
  • 支持消息接收注册和自定义分发机制

包装层：

  • 支持默认的数据序列化和反序列化（POJO对象、protobuf、FastJSON）。
  • 底层框架API支持自定义序列化框架注入。
  • 提供了底层多线程的能力。
  • 提供Offset commit的高级控制机制

2. 应用层消息框架设计

总体设计如下：

模块划分：

于是，经过设计可以初步划分为5个模块：

kafka-message-service-api
kafka-message-service-impl
kafka-protobuf-model-generator
kafka-api-wrapper-api
kafka-api-wrapper-impl

具体模块设计图如下：

详细类图设计：

后面再补充

API设计：

对应用层来说，只需要熟悉和理解这几个API就可以完全的收发消息。至于底层的多线层、偏移量的控制、异常的处理、Kafka的配置等都不需要直接处理。

Message Sender Service API:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
public interface IMessageReceiverService {
    /**
     * 监听Topic application.yml: ${bimgis.kafka.topics.read.froom}，
     * 并按照KafkaFrameMsg格式解析
     * 业务层需要自定义实现是否处理该消息，覆写接口：{IMessageReceiver.canHandle}
     */
    <T extends Message> String subscribe(IMessageReceiver<T> receiver);

    /**
     * 监听某个topic
     *
     * @param topic    主题
     * @param receiver 消息处理函数
     * @return 唯一的ID = receiver.id
     */
    String subscribe(String topic, IMessageReceiver<KafkaFrameMsg> receiver);

    /**
     * 监听某个主题，按照protobuf定义的格式进行预解析
     *
     * @param topic
     * @param receiver
     * @param tClass
     * @param <T>
     * @return 唯一的ID = receiver.id
     */
    <T extends Message> String subscribe(String topic, IMessageReceiver<T> receiver, Class<T> tClass);


    /**
     * 去订阅某个主题，同时移除掉监听的consumer
     *
     * @param recvId 订阅时返回的recvId
     */
    void unsubscribe(String recvId);


    /**
     * 去注册对应的消息接收器
     *
     * @param receiver
     * @param <T>
     */
    <T extends Message> void unsubscribe(IMessageReceiver<T> receiver);

    /**
     * 用于主题消费计数和统计
     *
     * @return
     */
    List<MessageTopicStat> stat();

    @Data
    class MessageTopicStat {
        String consumerId;
        List<MessageReceiverStatItem> handlers = new ArrayList<>();
    }

    @Data
    class MessageReceiverStatItem {
        String id;
        String topic;
        String className;
        boolean isDelete = false;
        String createdTime;
        String updatedTime;
    }

}

Message Receiver Service API:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
public interface IMessageSenderService {

    /**
     * 给消息添加时间戳、生成auth_token、添加唯一msgId
     *
     * @param req
     * @return
     */
    KafkaFrameMsg.Builder buildHeader(KafkaFrameMsg.Builder req);

    /**
     * 发送消息（指定topic,按照protobuf序列化进行发送）
     *
     * @param topic
     * @param req
     * @return
     */
    Future<Integer> send(String topic,
                         KafkaFrameMsg.Builder req);


    /**
     * 发送固定Topic消息，不需要任何响应也不要回调
     *
     * @param req
     * @return
     */
    Future<Integer> send(KafkaFrameMsg.Builder req);

    /**
     * 发送固定Topic消息，处理响应消息，默认2s超时
     *
     * @param req
     * @param receiver
     * @return
     */
    Future<Integer> send(KafkaFrameMsg.Builder req,
                         IKafkaFrameMessageReplyHandler receiver);


    /**
     * 发送固定Topic消息，处理响应消息，自定义超时时间
     *
     * @param req
     * @param receiver
     * @return
     */
    Future<Integer> send(KafkaFrameMsg.Builder req,
                         IKafkaFrameMessageReplyHandler receiver,
                         Duration timeout);


    interface IMessageReplyHandler<T, R> {
        void onReply(T req, R rsp);

        void onTimeout(T req);

        void onFailed(T req, int errorCode);
    }

    /**
     * Request-Response 回复消息处理回调接口
     */
    abstract class IKafkaFrameMessageReplyHandler<T extends Message>
            implements IMessageReplyHandler<KafkaFrameMsg, KafkaFrameMsg> {

        private static Logger logger = LoggerFactory.getLogger(IKafkaFrameMessageReplyHandler.class);

        /**
         * 自动将消息的Body解析成对应的对象
         *
         * @param req
         * @param rsp
         * @param busiBody
         */
        protected abstract void onReply(KafkaFrameMsg req, KafkaFrameMsg rsp, T busiBody);

        @Override
        public final void onReply(KafkaFrameMsg req, KafkaFrameMsg rsp) {
            //解码先
            logger.debug(String.format("[request-and-reply] Receiving response, req id:%s, req auth:%s" +
                            "rsp id:%s, rsp auth:%s, rsp sender:%s, rsp dest:%s",
                    req.getMsgId(), req.getAuthToken(),
                    rsp.getMsgId(), rsp.getAuthToken(),
                    rsp.getMsgSource().toString(),
                    rsp.getMsgDestList().toString()));

            //先对消息进行解码
            T body = (T) AnyMessageUnpacker.getInstance().unpack(rsp.getBusiData()).get();
            //调用接口
            onReply(req, rsp, body);
        }

        @Override
        public void onTimeout(KafkaFrameMsg req) {
            logger.error(String.format("message timeout for msg id:%s, busiCode:%s.\r\n",
                    req.getMsgId(),
                    req.getBusiCode().toString()));
        }

        public void onFailed(KafkaFrameMsg req, int errorCode) {
            logger.error(String.format("message failed, error code ->%d,msg id:%s, busiCode:%s\r\n",
                    errorCode,
                    req.getMsgId(),
                    req.getBusiCode().toString()));
        }
    }
}

MessageReceiver

1
2
3
4
5
6
7
8
9
10
11
12
13
/**
 * 消息接收处理回调接口定义
 *
 * @param <T>
 */
public interface IMessageReceiver<T extends Message> {

    boolean onReceive(T msg);

    default boolean canHandle(T msg) {
        return true;
    }
}

3. 消息协议制定

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
message MessageEnvelope {
    MessageId message_id = 1;
    MessageType message_type = 2;

    //用来做消息路由
    SystemId receiver_system_id = 16;
    int32 receiver_app_id = 17;
    int32 receiver_domain_id = 18;
    int32 receiver_module_id = 19;
    int32 receiver_function_id = 20;

    SystemId sender_system_id = 21;
    int32 sender_app_id = 22;
    int32 sender_domain_id = 23;
    int32 sender_module_id = 24;
    int32 sender_function_id = 25;

    //消息体
    MessageBody body = 30;
}
  1. 1.https://kafka.apache.org/10/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html ↩