灵感来源之前在B站看到一个有趣的视频：【B站】【还有】极致云游戏！5000人一起开一辆车，重现经典的群体智慧实验，非常有趣。up主通过代码实现实时读取直播间的弹幕内容，然后控制自己的电脑，将弹幕翻译成指令来控制《赛博朋克2077》游戏。看的人越来越多，最后连直播间都破了（当然，其实是因为那天整个B站都崩了）。我很好奇它是怎么做到的。外行看热闹，内行看门道。作为半高手，我们模仿UP主的思路，自己做一个。所以我今天的目标是重现一个通过弹幕控制直播间的代码，最后在自己的直播间开播。给大家展示一下我的成品最后的短视频：【B站】模仿UP主做一个弹幕控制的直播间！是不是看起来很像样。第一版的设计思路首先在脑海中规划出一个大概的思路，如下图所示：这个思路看似很简单，但是还是需要说明一下。首先我们要搞清楚弹幕的内容怎么抓取。我们常见的大部分直播平台，在浏览器端，都是通过WebSocket向观众推送弹幕。在手机、平板等客户端（非web）上，可能会有一些比较复杂的TCP来推送弹幕。TCP消息传递有一篇很好的文章，就是美团的这篇：美团终端消息传递服务Pike的进化。归根结底，这些弹幕是通过在客户端和服务器之间建立长链接来实现的。.因此，我们需要做的就是以代码为客户端，与直播平台进行长链接。这样就可以拿到弹幕了。我们只需要实现整个弹幕控制流程即可，所以弹幕的捕获不是本文的重点，还是找个现成的轮子吧！在Github上搜索后发现了一个非常不错的开源库，可以获取很多直播平台的弹幕：https://github.com/wbt5/real-url获取斗鱼&虎牙&哔哩哔哩&抖音&快手等58个真实流媒体地址（直播源）和弹幕。直播源可在PotPlayer、flv.js等播放器中播放。我们克隆代码，运行main函数，输入哔哩哔哩直播间任意地址，获取直播间实时弹幕：代码直接在控制面板塔上打印获取的弹幕（包括用户名）.他是怎么做到的呢？Python核心代码如下（不熟悉Python？没关系，把它当成伪代码就好了，很容易理解）：wss_url='wss://broadcastlv.chat.bilibili.com/sub'心跳=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@staticmethodasyncdefget_ws_info(url):url='https://api.live.bilibili.com/room/v1/Room/room_init?id='+url.split('/')[-1]reg_datas=[]asyncwithaiohttp.ClientSession()assession:asyncwithsession.get(url)asresp:room_json=json.加载（等待resp.text（））room_id=room_json['data']['room_id']data=json.dumps（{'roomid'：room_id，'uid'：int（1e14+2e14*random.random（）),'protover':1},分隔符=(',',':')).encode('ascii')data=(pack('>i',len(data)+16)+b'\x00\x10\x00\x01'+pack('>i',7)+pack('>i',1)+data)reg_datas.append(data)returnBilibili.wss_url,reg_datas连接到bilibili的直播弹幕WSS地址，即WebSocket地址，然后伪装成客户端，接受弹幕推送OK，完成第一条一步到位，下一步就是利用消息队列发送弹幕了。启用单独的消费者接收弹幕。为了实现尽可能简单，没有使用那些专业的消息队列。这里使用redis的列表作为队列，将弹幕内容放入其中。sender的核心代码如下：#linkRedisdefinit_redis():r=redis.Redis(host='localhost',port=6379,decode_responses=True)returnr#messagesenderasyncdefprinter(q,redis):whileTrue:m=awaitq.get()ifm['msg_type']=='danmaku':print(f'{m["name"]}:{m["content"]}')list_str=list(m["content"])print("Bulletchatsplit:",list_str)forcharinlist_str:ifchar.lower()inkey_list:print('Pushqueue:',char.lower())redis.rpush(list_name,char.lower())发送完弹幕内容后，需要写一个consumer来消费弹幕，提取里面的所有指令。以及，消费者收到弹幕后如何消费？我们需要一种方法来使用代码命令来控制计算机。我们继续本着不造轮子的原则，找到了一个Python自动化控制库PyAutoGUIPyAutoGUI是一个跨平台的人机界面自动化Python模块。用于以编程方式控制鼠标和键盘。安装这个库，在代码中引入，就可以通过他的API控制电脑鼠标键盘进行相应的操作。这是完美的！消费者（控制计算机）的核心Python代码如下：#linkRedisdefinit_redis():r=redis.Redis(host='localhost',port=6379,decode_responses=True)returnr#consumerdefcontrol(key_name):print("key_name=",key_name)ifkey_name==None:print("这次没有发出命令")returnkey_name=key_name.lower()#控制计算机命令ifkey_nameinkey_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("开始听弹幕messages,loop_sec=",loop_sec)whileTrue:key_name=r.lpop(list_name)control(key_name)time.sleep(loop_sec)ok，大功告成，我们打开弹幕发送队列和消费者，这样持续循环消费队列开始运行。弹幕中一旦出现wsad按键，这个按键是常用的控制游戏的，电脑就会给自己下达指令。第一个版本运行中的问题我兴冲冲地打开自己的B站直播间开始调试，却发现自己还是太天真了。第一个版本的代码暴露了很多问题。下面一一说说问题，我是怎么解决的。这个命令是不人道的。水友其实很喜欢发wwwdddd这样的重复词（redundantwords），但是第一个版本的实现只支持单字幕。发现不行后，水友们离开了直播间。这个好解决，把弹幕内容拆分成每个词，然后推入队列。解决方案：拆解弹幕，将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("Bulletchatsplit:",list_str)forcharinlist_str:ifchar.lower()inkey_list:print('Pushqueue:',char.lower())redis.rpush(list_name,char.lower())上面两个问题解决后，sender如下是这样工作的：弹幕指令的积累是个大问题。如果把所有水友发来的弹幕指令都处理了，肯定会出现无法消费的问题。解决方法：处理弹幕需要固定时间，其他的全部丢弃。if__name__=='__main__':r=init_redis()print("开始监听弹幕消息,loop_sec=",loop_sec)whileTrue:key_name=r.lpop(list_name)#一次只取出一个命令，并且然后把列表清空，也就是把这个时间窗口内的其他弹幕全部扔掉！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...我是在阿里搬砖的工程师@曼三刀酱持续更新优质文章，离不开您的点赞转发分享！全网唯一技术公众号：浅谈后端技术
