Python 主流棋牌游戏 服务端 框架分析 原创笔记
经测试此产品运营稳定
包含数十款房卡子游戏、俱乐部(五级权限)、比赛场
客户端采用Lua脚本开发 、后端Python
看过一些棋牌产品 很多产品基于此套棋牌框架开发而来 算市面上一个主流框架
但却没有发现一个关于此框架的文档说明 特此个人准备写几篇文档以方便新手和后来的开发维护人员借鉴 以节约队伍开发成本 尽快步入产品开发过程
Web提供的API服务是基于tornado框架实现
网关与业务之间使用了twisted异步通讯库 用于提升性能 socket长链接方式
框架在设计上支持负载均衡 除路由服务外 其他的服务器皆支持动态无限量扩展部署
Web服务 -- web_server.py
--- Http短链接形式 服务更加稳定 且在用户基数比较大的时候可以很有效的节省服务器资源 减轻服务器负担
--- 提供APIs(提供除子游戏 游戏层面的消息通讯外的所有接口 如比较高频的创建房间、加入房间)
---提供游戏系统APIs 方便运营对数据和游戏进行管控
游戏网关 -- gate.py --> base_server.py
--- 服务端除web服务外的 唯一向外开放的网络接口
--- 实现系统和用户(群)通讯 维护每个链接进来客户端的session 将seesion与用户uid对应起来
路由服务 -- router.py --> base_server.py
--- 接受来自web端的请求、游戏服务消息 将消息转发到相应的服务
游戏服务类 -- base_game.py --> base_server.py
--- 发送消息给路由服务 路由将消息转发到网关 网关将消息根据uid(s)发送给相应的客户端 客户端发送消息到服务器则路径相反
网络部分通讯流程关系图:
代码详解
./web-cocos/web_server.py ---> 基于tornado实现的一个web服务处理器
#配置web服务器端口信息
define("port", default=8193, help="run on the given port", type=int)
class Application(tornado.web.Application):
def __init__(self):
#此处仅列出几个代表性的api
handlers = [
#苹果内购iap接口
("/iap", IAPHandler),
#游客登陆
("/guestLogin", GuestLoginHandler),
#微信登陆
("/wechatLogin",WeChatLoginHandler),
#创建房间
("/createRoom", CreateRoomHandler),
#获取钻石信息
("/getDiamondsChange", DiamondsHandler),
#查询房间信息
("/queryServerInfo", QueryRoomHandler),
#...
#...
]
#注册web消息处理接口
tornado.web.Application.__init__(self, handlers, **settings)
./server-cocos/gate.py ---> ./server-cocos/base/init.py
def _do_start(server_class, server_id, service_type, server_name, is_gate):
assert server_id
assert service_type
#如果是网关服务 则从数据库 'servers'表中读取服务器信息
if is_gate:
server_info = servers_model.get(server_id)
else:
#如果是子游戏服务 则从数据库‘server_rooms’表中读取服务器信息
server_info = servers_model.get_idle_room(server_id, service_type)
if not server_info:
text = f"\033[31mError\033[0m {server_name} - \033[31m{service_type}\033[0m - {server_id} isn't idle server!"
print(text)
main_logger.fatal(text)
sys.exit(0)
#启动服务
_run_server(server_class, server_id, service_type, server_name, server_info)
./server-cocos/hall/gateway.py 维护每个链接进来的客户端session 且提供账户验证服务、系统服务
#启动网关服务
def start_service(self):
#链接路由服务 将服务注册到路由表 且设置一个收到网络消息的回调函数
self._start_listen_channel(self.__on_receive_message)
player_agent = Factory()
player_agent.protocol = SessionClient
reactor.listenTCP(self.__port, player_agent)
self.logger.info('Starting listening on port %d', self.__port)
LoopingCall(5 * 60, self.__clean_timeout_sessions)
servers_model.update_status(self.server_id, 1)
#添加系统消息shutdown的处理函数 用于妥善停止网关服务
reactor.addSystemEventTrigger('before', 'shutdown', self.on_signal_stop)
reactor.run()
#装载用户验证服务、系统服务
def setup(self, server_id, service_type, service_name, server_info):
BaseServer.setup(self, server_id, service_type, service_name, server_info)
port = int(server_info.get('port'))
self.__port = port
#设置验证服务处理handler 用户登陆、注册
self.set_service(AuthService())
#设置系统服务处理handler 系统消息广播
self.set_service(SystemService())
#保存用户session 将session与uid关联起来
def __save_session(self, session: SessionClient):
if not session or not session.uid or not session.verified:
return
self.__sessions[session.uid] = session
# 接受到客户端发来的消息
def on_line_received(self, session, line):
if not line or len(line) < 2:
return
#将接受到客户端的消息进行json格式解码
ret = utils.json_decode(line)
if not ret or not ret.get('cmd'):
session.close()
return self.logger.info('receive data error: ' + utils.bytes_to_str(line))
self.distribute(session, ret.get('cmd'), ret.get('msg'))
#消息分发处理
def distribute(self, session, cmd, msg):
service_type, cmd = protocol_utils.unpack_command(cmd)
#将消息分发到相应已注册的handler
service = self.__services.get(service_type)
if not service:
return self.__check_run_client_commands(session, service_type, cmd, msg)
service.service(session, cmd, msg)
#向所有用户广播消息
def send_global_message(self, service_type, cmd, msg):
for s in self.__sessions.values():
self.__send_message_by_session(s, service_type, cmd, msg)
#向指定uid用户发送消息
def __send_message_by_uid(self, uid, service_type, cmd, msg):
session = self.__get_session_by_uid(uid)
if session:
session.send(protocol_utils.pack_client_message(service_type, cmd, msg))
./server-cocos/hall/router.py --- 每一个服务(包括网关、子游戏服务)都会注册到路由表 从web端接受到消息 然后根据服务类型分发到相应的服务处理
from protocol.route_protocol import PubFactory
# 启动路由服务
def start_service(self):
endpoints.serverFromString(reactor, "tcp:{0}".format(self.__port)).listen(PubFactory())
self.logger.info('Starting listening on port %d', self.__port)
#添加系统消息shutdown的处理函数 用于妥善停止路由服务
reactor.addSystemEventTrigger('before', 'shutdown', self.on_signal_stop)
reactor.run()
./protocol/router_protocol.py
#当有服务器链接到路由服务时 将服务注册到路由表clients
def connectionMade(self):
if self.ip not in self.factory.allow_ips:
self.sendLine(b"Access denny.")
self.close()
return
self.factory.clients.add(self)
#接收网络消息
def lineReceived(self, line):
head, body = protocol_utils.unpack_s2s_package(line)
if not head:
return
cid, from_sid, from_service, to_sid, to_service, with_ack, cmd = protocol_utils.unpack_s2s_head(head)
if 1 == with_ack: # 响应ACK消息
self.__response_ack(cid)
if cmd == commands_s2s.S2S_HEART_BEAT:
return self.__on_heart_beat(from_sid, from_service)
if cmd == commands_s2s.S2S_ACK:
return
if cmd == commands_s2s.S2S_SEND:
return self.__try_send_message(to_sid, to_service, line)
#当路由服务接受到消息时 遍历路由表clients 将消息分发给相应的服务器
def __try_send_message(self, to_sid, to_service, line):
for c in self.factory.clients:
if c == self:
continue
if to_sid > 0 and to_sid != c.sid:
continue
if to_service > 0 and to_service != c.service_type:
continue
c.sendLine(line)
./server-cocos/protocol/router_client.py --- 每一个链接到路由服务的 都属于一个router client
#注册链接断开和 消息接受处理的handler
def set_handlers(self, on_connection_lost, on_line_received):
self.__on_connection_lost = on_connection_lost
self.__on_line_received = on_line_received
#接受到路由转发来的消息
def lineReceived(self, line):
self.__last_data_arrive_time = utils.timestamp()
if callable(self.__on_line_received):
#__on_line_received 为一个回调函数 接收到消息后将消息转发到这里处理
self.__on_line_received(self, line)
else:
main_logger.warn("line receive with no handlers!", line)
#给路由服务发送消息
def send(self, obj):
# 发送数据包, obj要可以被json编码
# s = utils.json_encode(obj)
self.sendLine(utils.json_encode(obj).encode("utf8"))
#链接路由服务
def connect_route_server(on_conn_success, on_conn_fail):
host = config.get_item("router_ip") or "127.0.0.1"
port = config.get_item("router_port") or 20000
point = TCP4ClientEndpoint(reactor, host, port)
conn = point.connect(PubClientFactory())
conn.addCallbacks(on_conn_success, on_conn_fail)
./server-cocos/base/base_game.py --- 子游戏服务的基类
#游戏服务类继承于基本服务器BaseServer
class BaseGame(BaseServer):
def __init__(self):
BaseServer.__init__(self)
self.__defer = None
self.__in_stop = False
self.__stop_timer = None
self.__players = {} # 玩家数列列表
self.__judges = {} # 桌子对象列表
#所有子游戏通用 添加处理几个游戏层面的消息处理handler
self._add_handlers({
const.ACK: self.update_ack_time,
const.PLAYER_TABLE_REMOVE: self.remove_player_table,
const.PLAYER_UNREADY: self.unready_player,
})
#启动游戏服务
def start_service(self):
#更新数据库中此游戏服务的状态
servers_model.set_room_start(self.server_id, self.service_type, os.getpid(), utils.read_version())
#链接路由服务 将服务注册到路由表 且设置一个收到网络消息的回调函数
self._start_listen_channel(self.on_receive_message)
try:
#添加系统消息shutdown的处理函数 用于妥善停止游戏服务
reactor.addSystemEventTrigger('before', 'shutdown', self.on_signal_stop)
reactor.run()
except KeyboardInterrupt:
self.close_server()
#发送网络数据给玩家
def send_body_to_player(self, uid, cmd, body, service_type=None, to_all_service=False):
service_type = service_type or self.service_type
#根据协议打包数据
message = protocol_utils.pack_to_player_body(cmd, uid, body)
#如果未指定目的服务器 则发送数据到网关
to_service = 0 if to_all_service else const.SERVICE_GATE
#此函数通过保存的路由器句柄发送数据到路由
return self._s2s_raw_publish(0, self.sid, service_type, 0, to_service, 1,
commands_s2s.S2S_SEND, message)
#注册处理网络消息handlers
self._add_handlers({
const.ACK: self.update_ack_time,
const.PLAYER_TABLE_REMOVE: self.remove_player_table,
const.PLAYER_UNREADY: self.unready_player,
})
#接受到网络数据 根据注册的handlers进行分发处理
def on_receive_message(self, from_sid, from_service, to_sid, to_service, body):
#解包网络数据
cmd, uid, msg = protocol_utils.unpack_to_player_body(body)
if not cmd:
return
if cmd == commands_system.SOCKET_CHANGE:
offline = msg.get("offline")
return self.on_player_connection_change(uid, offline)
if from_service == const.SERVICE_SYSTEM:
return self.__run_system_commands(cmd, uid, msg)
return self.service(cmd, uid, msg)
扩展性
在有需要的情况下也可以部署多个网关服务器
游戏业务负载均衡
1.当一个新桌子被启用时 用redis记录对应服务器id和桌子数量
2.在创建新桌子的时候 更具redis中的记录 选取一个最低负载的游戏服务器提供服务
./server-cocos/base/base_judge.py
#添加玩家到桌子信息中 更新桌子数量信息
def add_player_in_table_info(self, info):
# 玩家第一次加入到一个游戏桌子的时候
if not self.__first_join:
self.__first_join = True
# 在游戏服务中更新桌子数量信息
self.__service.modify_table_count(self.club_id, self.sub_floor_id)
find_player = False
for i in self.__table_info['players']:
if not i:
continue
if i['uid'] == info['uid']:
find_player = True
break
# 添加玩家到桌子信息中
if not find_player:
self.__table_info['players'].append(info)
./server-cocos/base/base_game.py
#更新桌子数量信息
def modify_table_count(self, club_id, sub_floor):
#将redis中字符串key存储的数字值增加一
share_connect().incr(f"table:{self.service_type}:{self.server_id}")
if sub_floor == -1:
return
database.incr_club_sub_floor_count(club_id, sub_floor)
self.club_join_create_room(club_id, sub_floor)
./web-cocos/models/game_room_model.py -- 游戏房间相关模型
#从数据库中查询所有在运行的对应类型的服务器
def _get_running_sid_by_game_type(conn, game_type):
sql = f"SELECT sid FROM `server_rooms` WHERE game_type={game_type} and status=1"
return conn.query(sql)
#从redis中取出服务器id
def _get_server_sid_from_redis(redis_conn, game_type, servers):
rooms = {}
for i in servers:
#在redis中取出字符串Key对应的数字值
count = redis_conn.get(f"table:{game_type}:{i['sid']}")
count = 0 if not count else count.decode('utf-8')
rooms[i['sid']] = int(count)
max_desk = 1000000
default_sid = 1
#获得一个负载最低的服务器 并返回
for key in rooms:
if rooms[key] <= max_desk:
default_sid = key
max_desk = rooms[key]
return default_sid
#在创建游戏房间时 选取负载最低的服务器
def get_best_server_sid(conn, redis_conn, game_type):
#得到对应服务的所有服务器
server = _get_running_sid_by_game_type(conn, game_type)
if not server:
return 0
if len(server) == 1:
return server[0]['sid']
return _get_server_sid_from_redis(redis_conn, game_type, server)
./web-cocos/controllers/room_handler.py --- 游戏房间控制器
#创建房间消息 处理的handler
class CreateRoomHandler(BaseHandler):
#此处可能导致解散一个房间后 创建相同游戏房间 房间号和之前相同的问题
#先尝试从redis中取房间ID 如果为0则 生成随机房间ID
tid = base_redis.spop_table_id() or utils.get_random_num(6)
#校验用户是否具备足够开房钻石
idle_table_diamond = tables_model.get_total_idle_diamonds_by_uid(self.share_db(), user['uid'])
if user.get('diamond', 0) + user.get('yuan_bao') < diamonds + idle_table_diamond:
return self.write_json(error.DIAMONDS_CLUB_NOT_ENOUGH)
#通过redis链接获取当前最低负载房间服务器id
room_sid = game_room_model.get_best_server_sid(self.share_db(), redis_conn(), game_type)
if not room_sid:
return self.write_json(error.DATA_BROKEN)
# 房间创建成功 将房间服务器id等开房数据插入到 'tables'表中
count = tables_model.insert(self.share_db(), room_sid, game_type, int(tid), self.uid, is_agent, total_round,
diamonds, rule_type, club_id, rules, consume_type=consume_type)
#房间创建成功 则返回房间消息给创建者
return self.write_json(error.OK, {'roomID': int(tid), "gameType": game_type,
"ruleDetails": rules, "isAgent": is_agent})
Python 主流棋牌游戏 服务端 框架分析 原创笔记相关推荐
- pomelo + vscode + typescript搭建可约束可调试的游戏服务端框架
说在前面 pomelo: 它是网易开源的一套基于Node.js的游戏服务端框架,详情请戳这里关于pomelo的种种这里不详细说.点击链接查看详情.但是由于pomelo是js项目,使用起来的时候并不是很 ...
- 推广下自己的JAVA开源游戏服务端框架
Carmelo是基于Java的游戏服务端框架,适合于页游和手游.它的主要特点是: 利用Netty实现高效的NIO通信,同时支持TCP/HTTP协议 完善的三层架构模型,易扩展 通用.完善的sessio ...
- 棋牌游戏服务端开发和设计-苏劲-专题视频课程
棋牌游戏服务端开发和设计-279人已学习 课程介绍 本门课程讲解棋牌游戏服务端的架构.数据库的设计.数据库异步存储.帐号管理.房间管理等棋牌游戏服务端的核心技术,有意向从事棋牌研发的同 ...
- 基于skynet设计游戏服务端框架
skynet并不是一个开箱即用的服务端框架,游戏后端在开展业务时,需要根据自身业务特点,合理设计相应的服务端框架.在这里我根据自身的设计目标,写下各方面的选择与取舍.对于小型企业来说,一些商业化的软件 ...
- 从游戏服务端角度分析移动同步（状态同步）
从游戏服务端角度分析移动同步(状态同步) 参考文章: https://www.lfzxb.top/ow-gdc-gameplay-architecture-and-netcode/ https://z ...
- 新一代游戏服务端框架，该是什么样的？
说起游戏服务端引擎,大家会想起Skynet.KbEngine.Photon.Pomelo等等.在探索服务端技术时候,我们不仅仅要了解当代服务端引擎,更要有些前沿眼光,去预测未来的游戏服务端是什么样的. ...
- 基于滴滴云的棋牌游戏服务端架构设计
现在小团队开发的棋牌游戏有很多,棋牌行业的相互攻击是非常普遍的现象,同行之间往往会采取 DDOS.CC 等攻击的手段来打击对手,这是目前棋牌运营商们面临的比较严峻的一个问题,那么在设计棋牌游戏服务端架 ...
- 如何快速开发游戏服务端框架？
快速开发游戏服务端框架的方法如下: 分析游戏需求:首先要明确游戏的功能和玩法,并确定服务端的职责. 选择适当的开发工具:可以选择一些专门用于游戏服务端开发的工具,比如 Unity.Unreal Eng ...
- SGAME：一个简单的go游戏服务端框架
SGame是一个由GO实现的游戏简单服务端框架. 说明 主要是使用GO丰富的库资源和较高的开发效率. 开发简单 可以使用已有的代码框架很方便的构建一个新的进程 方便扩展 基于已有的框架可以动态的扩展进 ...
最新文章
- Flask redirect
- 静态的顺序表（C语言实现）
- 建筑电工模拟考试完整版在线讲解
- JAVA基于UDP的一个聊天程序
- Vue与React的异同
- zedboard u-boot编译的心路历程
- 微商怎么引流学生粉？如何把学生粉变现成精准粉？
- App Links的使用以及坑
- 安卓中自定义view控件代替radiogroup实现颜色渐变效果的写法
- ContentProvider使用Demo
- win10环境向移动固态硬盘安装Ubuntu 18.04.3 LTS系统（即插即用）
- 互联网2B和2C的区别
- python生成快递取件码_快递,顺丰,python,截图,15Seconds
- python字典的用法_python字典dict使用方法大全
- 盛会落幕，精彩延续 | 云扩科技入选《2022中国AI商业落地市场研究报告》
- （图）HOLD住！aiwi最新体感游戏强势来袭！！
- 使用TensorFlow的卷积神经网络识别手写数字（3）-识别篇
- 投资理财这3年 | 其他
- 【海云捷迅云课堂】分布式存储系统纠删码技术分享
- PS 2021最新最全插件滤镜大全一键安装版下载 Photoshop插件合集WIN一键安装版 支持PS 2021