共计 4411 个字符，预计需要花费 12 分钟才能阅读完成。
灵感起源
之前在 B 站看到一个有意思的视频：
【B 站】【亦】终极云游戏！五千人同开一辆车，复现经典群体智慧试验
大家能够看看，很有意思。
up 主通过代码实现了实时读取直播间里的弹幕内容，进而管制本人的电脑，把弹幕翻译成指令操控《赛博朋克 2077》游戏。
观众也越来越多，最初甚至还把间接间搞崩了（当然，其实是因为那天 B 站全站崩了）。
我非常好奇到底是怎么做到的。
在行看热闹，外行看门道，作为半个外行，咱们就模拟 UP 主的想法，本人做一个。
所以明天我的指标就是复刻一个 通过弹幕管制直播间 的代码，并且最终在本人的直播间开播。
先给大家看看最终我的成品小视频：
【B 站】模拟 UP 主，做一个弹幕管制的直播间！
看起来是不是很像样了。
初版设计思路
首先在脑海里布局一个大抵的思路，如下图：
这个思路看起来很简略，不过还是得解释一下，首先咱们要搞清楚，弹幕的内容是怎么抓到的。
大部分咱们常见的直播平台，在浏览器端，弹幕都是通过 WebSocket 来推送给观众的。在手机平板等客户端（非 Web 端），可能会有一些更加简单的 TCP 进行弹幕的推送。
对于 TCP 的音讯投递，有个很好的文章，就是美团的这个：美团终端音讯投递服务 Pike 的演进之路
归根结底, 这些弹幕都是通过在客户端和服务端建设长链接来实现的。
所以，咱们须要做的就是用代码作为客户端，与直播平台进行长链接。这样就能拿到弹幕。
咱们只是须要实现整个弹幕管制的流程，所以弹幕的抓取也不是本文的重点，咱们来淘一个现成的轮子！在 Github 上一顿找，找到了一个十分不错的开源库，外面可能获取很多直播平台的弹幕：
https://github.com/wbt5/real-url
获取斗鱼 & 虎牙 & 哔哩哔哩 & 抖音 & 快手等 58 个直播平台的实在流媒体地址 (直播源) 和弹幕，直播源可在 PotPlayer、flv.js 等播放器中播放。
咱们把代码 clone 下来，运行 main 函数，轻易输出一个 Bilibili 直播间地址，就能拿到直播间实时的弹幕流：
代码里把获取到的一条条弹幕（包含用户名）间接打印在了控制台。
他是如何做到的呢？外围的 Python 代码如下（不相熟 Python？不要紧，就当做伪代码，很容易看懂）：
wss_url = 'wss://broadcastlv.chat.bilibili.com/sub'
heartbeat = b'\x00\x00\x00\x1f\x00\x10\x00\x01\x00\x00\x00\x02\x00\x00\x00\x01\x5b\x6f\x62\x6a\x65\x63\x74\x20' \
b'\x4f\x62\x6a\x65\x63\x74\x5d'
heartbeatInterval = 60
@staticmethod
async def get_ws_info(url):
url = 'https://api.live.bilibili.com/room/v1/Room/room_init?id=' + url.split('/')[-1]
reg_datas = []
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
room_json = json.loads(await resp.text())
room_id = room_json['data']['room_id']
data = json.dumps({
'roomid': room_id,
'uid': int(1e14 + 2e14 * random.random()),
'protover': 1
}, separators=(',', ':')).encode('ascii')
data = (pack('>i', len(data) + 16) + b'\x00\x10\x00\x01' +
pack('>i', 7) + pack('>i', 1) + data)
reg_datas.append(data)
return Bilibili.wss_url, reg_datas它连上了 Bilibili 的直播弹幕 WSS 地址，也就是 WebSocket 地址，而后伪装成客户端，承受弹幕推送。
OK，做完了第一步，下一步就是用音讯队列将弹幕发送进去。开启独自的消费者接管弹幕。
为了实现上尽量简略，就不上那些业余的音讯队列了，这里用了 redis 的 list 作为队列，将弹幕内容放进去。
发送者外围代码如下：
# 链接 Redis
def init_redis():
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
return r
# 音讯发送者
async def printer(q, redis):
while True:
m = await q.get()
if m['msg_type'] == 'danmaku':
print(f'{m["name"]}：{m["content"]}')
list_str = list(m["content"])
print("弹幕拆分:", list_str)
for char in list_str:
if char.lower() in key_list:
print('推送队列：', char.lower())
redis.rpush(list_name, char.lower())实现了弹幕内容的发送后，须要写一个消费者，生产这些弹幕，把外面的指令都提取进去。
并且，在消费者收到弹幕后，如何生产呢？咱们须要一个可能用代码指令管制电脑的方法。
咱持续本着不造轮子的准则，找到了一个 Python 的自动化管制库 PyAutoGUI
PyAutoGUI is a cross-platform GUI automation Python module for human beings. Used to programmatically control the mouse & keyboard.
装置上这个库，在代码中引入，便能够通过他的 API 管制电脑鼠标和键盘执行对应的操作。几乎是完满啊！
消费者（管制电脑）外围 Python 代码如下：
# 链接 Redis
def init_redis():
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
return r
# 消费者
def control(key_name):
print("key_name =", key_name)
if key_name == None:
print("本次无指令收回")
return
key_name = key_name.lower()
# 管制电脑指令
if key_name in key_list:
print("收回指令", key_name)
pyautogui.keyDown(key_name)
time.sleep(press_sec)
pyautogui.keyUp(key_name)
print("完结指令", key_name)
if __name__ == '__main__':
r = init_redis()
print("开始监听弹幕音讯, loop_sec =", loop_sec)
while True:
key_name = r.lpop(list_name)
control(key_name)
time.sleep(loop_sec)ok，功败垂成，咱们关上弹幕发送队列和消费者，这个一直循环生产的队列就开始运行了。一旦弹幕中有 wsad 这种管制游戏罕用的按键，电脑就会本人给本人收回指令。
初版运行中的问题
我灰溜溜的关上本人的 B 站直播间，开始调试，后果发现我还是太天真了。这个初版代码裸露了十分多的问题。咱们一个个来说下是什么问题，我是如何解决的。
指令不人性化
水友们其实很喜爱发送相似 www dddd 这类反复单词（叠词），但初版的实现只反对单个字幕，水友们发现不得劲，没有作用后，就从直播间走了。
这点很容易解决，把弹幕内容拆分成每个单词，而后再推送给队列。
解决办法：拆解弹幕，把 DDD，拆成 D,D,D，发送个消费者。
危险指令
首先是玩家的指令超出了应该有的范畴。
在我把赛博朋克游戏关上，让弹幕观众管制游戏里的开车时，有个神秘观众进了直播间，默默发了个“F”，而后。。。
而后游戏里的 V（配角名）就从车里下来了，淦，我是让你们开车的，不是让你们下来和警察斗殴的。。。
解决办法：增加弹幕过滤器。
# 将弹幕进行拆分，只发送指定的指令给消费者
key_list = ('w', 's', 'a', 'd', 'j', 'k', 'u', 'i', 'z', 'x', 'f', 'enter', 'shift', 'backspace')
list_str = list(m["content"])
print("弹幕拆分:", list_str)
for char in list_str:
if char.lower() in key_list:
print('推送队列：', char.lower())
redis.rpush(list_name, char.lower())下面两个问题解决后，发送者就像上面这样运行了：
弹幕指令沉积
这是个很大的问题，如果解决所有水友发送的全副弹幕指令，肯定会存在生产不过去的问题。
解决办法：须要固定工夫解决弹幕，其余摈弃。
if __name__ == '__main__':
r = init_redis()
print("开始监听弹幕音讯, loop_sec =", loop_sec)
while True:
key_name = r.lpop(list_name)
# 每次只取出一个指令，而后把 list 清空，也就是这个工夫窗口内其余弹幕都扔掉！r.delete(list_name)
control(key_name)
time.sleep(loop_sec)弹幕从收回到观众看到后果有提早
在最开始的视频里，你们也能感触到了，从观众的指令收回，到最终被观众看到，大略要经验 5 秒的提早。其中，起码有 3 秒，都是网络直播流的提早，这一点，很难去优化。
回炉重造后的版本
通过一系列调优和波及，咱们的版本也算是从 V0.1 到了 V0.2 了。猛虎落泪。
上面是重构后的结构图：
后记
在写完这个我的项目后，我在直播间试了很屡次，体验曾经有限靠近 UP 主过后的视频了。我开播挂在那边良久，然而，人气最高的时候，也只有 20 几个人，寥寥十几条弹幕，还有很多是我发的。我还冀望着观众可能拉更多人进来一起玩呢，大失所望啊。
由此可得出结论，我，先得有粉丝，能力玩得起来啊，呜呜呜呜。大家要是不介意，能够关注下我的 B 站账号，也叫：蛮三刀酱。我会偶然抽风发点乏味的技术视频的。
本文实现的全副代码曾经开源在了 Github 上，大家能够在本人的直播间里试试呀：
https://github.com/qqxx6661/l…
我是在阿里搬砖的工程师 @蛮三刀酱
继续的更新优质文章，离不开你的点赞，转发和分享！
全网惟一技术公众号：后端技术漫谈
