前言

游戏业务发展迅猛，系统繁多，系统间协作效率低下。通过消息队列这个中间件的引入，解耦系统依赖关系以及统一交互方式，本文为详细架构设计文档，用于指导消息队列的开发、运维、测试等

词汇表

Reactor: 网络编程模式

Netty: 开源的网络编程框架

Mysql: 数据存储系统

ZooKeeper:分布式协调服务

Sharding: 数据复制技术

1. 业务背景

随着游戏业务的快速发展，业务的系统数量剧增，协作间无规范、效率低下，由此带来几个明显的系统问题：

• 效率问题：每个子系统提供的接口参数和实现都有一些细微的差别，导致每次都需要重新设计接口和联调接口，开发团队和测试团队花费了许多重复工作量。

• 耦合问题：当新增一个子系统时，例如如果要增加“风控系统”，那么风控系统需要提供对外的服务的接口与参数，同时其他系统需要按需接入。

基于以上背景，我们需要引入消息队列进行系统解耦，加速业务迭代。

2. 约束和限制

1.中间件团队 6 人，研发力量有限，研发周期 3 个月

2.性能上能支撑 2000+ tps

3.能支撑业务方快速对接，可用性高

4.系统需要高可用，保证高优先级业务

5.维护性要高，方便做收发消息情况、权限控制、上下线等

6.现有团队技术上不熟悉 Scala, 对 Kafka 无法进行二次扩展运维系统以及需求研发

3. 总体架构

3.1 架构分析

3.1.1 高性能

游戏新版本发布与 VIP 充值的消息数量不多，对系统高性能不高，满足日常处理能力即可。

3.1.1 高可用

需要做高可用设计，对游戏版本发布与 VIP 充值等业务都属于高优先业务，需要保证服务的可用性。

对于游戏版本发布来说，如果消息丢了, 导致运营无法按期执行对用户的版本更新承诺，以及运营活动无法开展影响产品的口碑以及商业价值。

对于 VIP 充值，若存在消息丢失影响到 VIP 用户特权，会导致用户满意度下降，甚至导致用户流失。

综合来看，消息队列需要高可用性，包括消息写入、消息存储、消息读取都需要保证高可用性。

3.1.2 可维护性

需要支撑运维、运营团队的日常维护需求，例如做到快速上下线，权限管控等

3.2 总体架构

1）采用数据分散集群的架构，集群中的服务器进行分组，每个分组存储一部分消息数据。

2）每个分组包含一台主 MySQL 和一台备 MySQL，分组内主备数据复制，分组间数据不同步。

3）正常情况下，分组内的主服务器对外提供消息写入和消息读取服务，备服务器不对外提供服务；主服务器宕机的情况下，备服务器对外提供消息读取的服务。

4）提供 SDK，客户端采取轮询的策略写入和读取消息。

4. 详细设计

1) 客户端设计：

客户端采用 Java 开发，基于 Netty 实现与服务端交互。

2) 服务端设计：

2.1)基于 Netty 框架开发，用 Reactor 网络模型。

2.2)两台服务器组成一个 Sharding, 每个 sharding 包含一主一从两台服务器，整个系统中可用设计成多个 Sharding。

2.3)服务器基于 Zookeeper 进行主从切换

2.4)主服务器提供消息读写能操作，从服务器只提供消息读操作

3) 客户端与服务端交互设计：

3.1) 采用 TCP 长连接，采用 Json 做为交互数据格式

3.2) 提供 Http 接口能力，支持非 Java 系统应用

4) MySql 设计

4.1）采用 MySql 主从同步机制

4.2) 每个消息队列对应一个表

4.3）消息表最多存储 30 天内的消息，过期的自动清除

4.4）直接用 MySQL 的主从复制来实现数据复制

4.1 核心功能

[必选，描述核心场景或者流程的实现机制，对应 4R 架构中的 Rule，每个核心场景一个小节]

[样例：

4.1.1 消息发送流程

4.1.2 消息消费流程

]

[技巧：使用系统序列图来描述 Rule，跟项目开发中写设计文档一样的写法]

4.1.1 消息发送流程

生产者通过 sdk，轮询本地存放的消息队列服务器列表，从连接着的消息队列服务中选择编号最小的开始，若生产成功则结束，若失败或者超时，则继续下 1 个连接着可用的消息队列服务继续发送。

4.1.2 消息读取流程

4.1.3 服务器主从切换

4.2 关键设计

1）消息发送可靠性

业务服务器中嵌入消息队列系统提供的 SDK，SDK 支持轮询发送消息，当某个分组的主服务器无法发送消息时，SDK 挑选下一个分组主服务器重发消息，依次尝试所有主服务器直到发送成功；如果全部主服务器都无法发送，SDK 可以缓存消息，也可以直接丢弃消息，具体策略可以在启动 SDK 的时候通过配置指定。

如果 SDK 缓存了一些消息未发送，此时恰好业务服务器又重启，则所有缓存的消息将永久丢失，这种情况 SDK 不做处理，业务方需要针对某些非常关键的消息自己实现永久存储的功能。

2）消息存储可靠性

消息存储在 MySQL 中，每个分组有一主一备两台 MySQL 服务器，MySQL 服务器之间复制消息以保证消息存储高可用。如果主备间出现复制延迟，恰好此时 MySQL 主服务器宕机导致数据无法恢复，则部分消息会永久丢失，这种情况不做针对性设计，DBA 需要对主备间的复制延迟进行监控，当复制延迟超过 30 秒的时候需要及时告警并进行处理。

3）消息如何存储

每个消息队列对应一个 MySQL 表，消息队列名就是表名，表结构设计为

CREATE TABLE `xx_topic` (

`id` bigint NOT NULL AUTO_INCREMENT COMMENT '自增 offset' ,

`create_time` datetime NOT NULL ,

`content` varchar(2048) NULL ,

`state` smallint NULL DEFAULT 1 COMMENT '默认新增记录为 1，无效记录更新为 0' ,

PRIMARY KEY (`id`)

);

CREATE TABLE `xx_topic_relation` (

`id` int(16) not null COMMENT '消费者唯一 id' ,

`update_time` datetime NOT NULL ,

`offset` int(16) NOT NULL ,

PRIMARY KEY (`id`)

);

4.3 设计规范

[必选，描述 Role 和 Relation 相关的开发框架、连接协议、数据包格式等]

[样例：

1）消息队列服务器使用 Spring Boot + Netty 开发

2）MySQL 使用 Innodb 存储引擎

3）TCP 包的结构设计……（此处省略，请自行补充）

]

[技巧：如果某个规范涉及内容比较多，请独立章节描述，例如数据包格式定义]

1）消息队列服务器使用 Spring Boot + Netty 开发

2）MySQL 使用 Innodb 存储引擎

3）TCP 包的结构设计：

3.1) 消息体设计:

整体: [固定协议头][协议内容]

细节: [整体长度 0-31][协议头长度 32-47][协议版本 48-55][消息类型 55-62][序列化方式 63-70][消息 ID 71-85][协议内容]

说明: a. 应用场景比较明确，因此对与协议头可设计成固定的协议头，可不考虑动态扩展的协议头设计

b.增加协议版本号，用于后续系统协议版本改造

c. 消息类型可用于快速对消息分类，比如不同应用场景，也可用通过认为约定消息 ID 划段的形式

d. 序列化方式，目前设计为 json，考虑后续可能的性能提升等扩展，预留序列化格式，比如 protobuf, stream.

3.2) 协议设计:

3.2.1) 客户端与消息管理系统连接:

const int req_svrconfig= 1;

{

"askid":1,

"module_id": 2,

}

module_id: 用于表示具体的业务系统 id，管理系统通过不同的业务系统分不同配置下发权限

const int resp_svrconfig= 2;

{

"askid":1,

"module_id": 2,

"err_code":0,

"servers":[

{ "ip":"127.0.0.1", "port":2001 },

{ "ip":"127.0.0.1", "port":2002 }

]

}

返回 servers 列表，用于客户端 SDK 做轮询策略使用.

3.2.2) 服务端与消息管理系统:

const int req_sync_serverinfo = 1;

{

"askid":1,

"group_id": 2,

"appid":3,

"serverinfo":{

"threads":10,

"packages":100,

"cpu": 30,

"mem":700,

"sockets":1000

}

}

group_id 为分组号，appid 为系统中配置的进程唯一 id, 两者可用来关联消息管理系统上的配置信息

serverinfo 中包含服务进程当前的信息，threads 线程数量，packages 堆包量，cpu 表示进程 cpu 消耗，mem 表示进程占用内存，sockets 表示当前连接者的客户端数量, 用于监控。

const int resp_sync_serverinfo = 2;

{

"askid":1,

}

消息管理系统管理命令下发

const int req_command = 3；

{

"askid":1,

"cmdid":1,

"content":{

}

}

cmdid 为具体的命令，context 为具体的命令 Json 内容.

cmdid 定义为 {publish_config=1, offline = 2}

const int resp_command = 4;

{

"askid":1,

"cmdid":1,

"err_code":0

}

3.2.3) 客户端与服务端:

连接信息:

const int req_connect = 1;

{

"askid": 10,

"client_type": 1

}

client_type: 1 为生产者，2 为消费者

const int resp_connect = 2;

{

"askid": 10,

"client_type":1,

"err_code": 0,

"client_id": 1,

}

client_id: 服务端给客户端分配的唯一 id, 与 socket 关联起来，能快速找到客户端, 可以理解为鉴权.

同时客户端也能用于映射， map<client_id, server>

心跳消息:

const int req_keepalive = 3;

{

"askid": 10,

"client_id":1

}

const int resp_keepalive = 4;

{

"askid":10,

"weight":10,

}

weight 可以设计为服务的健康值，表示当时服务能力情况，越大表示能力越强。

客户端可辅助来改造 sdk 的均衡策略，提升整体系统的可用性

生产者请求注册 topic:

const int req_add_topic = 5;

{

"askid":10,

"topic": "xx_topic",

"client_id":1

}

const int resp_add_topic = 6;

{

"askid":10,

"topic": "xx_topic",

"err_code": 0,

"topic_message_total":1000,

"topic_nums":10

}

topic_nums 表示当前客户端注册的 topic 数量

topic_message_total 表示该 topic 的消息数量

消费者关注 topic:

const int req_watch_topic = 7;

{

"askid":10,

"topic":"xx_topic",

"client_id":1

}

const int resp_watch_topic = 8;

{

"askid":10,

"topic":"xx_topic",

"err_code":0,

"topic_nums":10,

"topic_message_total":100

}

topic_nums 表示当前客户端注册的 topic 数量

topic_message_total 表示该 topic 的消息数量

const int req_unwatch_topic = 9;

{

"askid":10,

"topic":"xx_topic",

"client_id":1

}

const int resp_unwatch_topic = 10;

{

"askid":10,

"topic":"xx_topic",

"client_id":1，

"err_code":0

}

生产者生产消息:

const int req_producer_message = 11;

{

"askid":1,

"topic":"xx_topic",

"time":12910212,

"client_id":1,

"content":{

}

}

const int resp_producer_message = 12;

{

"askid":1,

"topic":"xx_topic",

"client_id":1,

"err_code":0,

"total":1000

}

消费者消费消息

const int req_consumer_message = 13;

{

"askid":1,

"topic":"xx_topic",

"client_id":1,

"offset":10

}

offset 表示消费者端本地存储的消费偏移量

const int resp_consumer_message = 14;

{

"askid":1,

"topic":"xx_topic",

"client_id":1,

"total":1000,

"offset":1002,

"message_count":2,

"messages":[

{"offset_id":1001, "message:": " {version=\"1.2\"; }"},

{"offset_id":1002, "message:": " {version=\"1.3\"; }"}

]

}

total 表示消息总量， offset 表示当前消费

举例游戏服务场景:

生产者协议:

{

"askid": 10001,

"topic": "version_publish",

"context":{

"game_id": 10,

"game_name": "xx 游戏",

"game_version": "2.1",

"publisher_id": 100,

"package_size(kb)": 109201,

"channel_id": 1000,

"desc_info": "增加 xx 副本"

}

}

消费者协议:

{

"askid": 10001,

"topic": "version_publish",

"total": 10000,

"offset": 1001,

"context":{

"game_id": 10,

"game_name": "xx 游戏",

"game_version": "2.1",

"publisher_id": 100,

"package_size(kb)": 109201,

"channel_id": 1000,

"desc_info": "增加 xx 副本"

}

}

5. 质量设计

[必选，描述和质量相关的设计，包括：可测试性、可维护性、可观测性、成本等设计]

[样例：

5.1 消息队列管理后台

5.2 成本

]

[技巧：如果某个维度不涉及，也请在文档中说明，避免评审的时候被认为考虑不周全]

6. 演进规划

6.1 消息队列一期：支持广播模式，支持消息队列指标监控、支持动态上下线功能

6.2 消息队列二期：支持 partition 分组模式