使用 Workerman 接入 Bilibili 直播弹幕协议

文章列表 - 她和她的猫

她和她的猫 · 2021-04-01 · via 文章列表 - 她和她的猫

逛 B 站的时候，突然想到可以用 PHP 接入直播弹幕，然后在命令行显示弹幕消息。

经过搜索发现了一篇讲解 Bilibili 直播弹幕协议的文章（链接在文末），通过这篇文章了解到了弹幕的协议格式以及大致的流程，开发过程中遇到的一些问题参考了弹幕姬的解决思路。

本文源码的 GitHub 地址：https://github.com/her-cat/bilibili-barrage

弹幕协议的介绍

弹幕协议由头部和数据组成，头部的长度是固定的 16 字节，数据的长度 = 数据包总长度 - 头部的长度。

协议的字节序均为大端模式。高字节在低地址，低字节在高地址，比如 0x1234，在大端模式下存储是 0x12 0x34，在小端模式下是 0x34 0x12。

弹幕协议图示

下面是弹幕协议的格式。

字段对照表：

字段	含义
packet_len	数据包的总长度
header_len	头部长度（固定为 16 字节）
version	协议版本号（默认为 2）
opcode	操作码，用来标识数据包的类型（详情见下表）
magic_number	魔术数字（默认为 1）
data	携带的数据，长度 = packet_len - header_len

已知的操作码：

操作码	常量	含义
2	Opcode::CLIENT_HEARTBEAT	客户端发送的心跳包
3	Opcode::POPULARITY_VALUE	人气值，数据是 4 字节整数
5	Opcode::CMD	命令，数据中[‘cmd’]表示具体命令（见下表）
7	Opcode::AUTHENTICATION	认证并加入房间
8	Opcode::SERVER_HEARTBEAT	服务器发送的心跳包

已知的命令：

命令	常量	含义
INTERACT_WORD	CMD::INTERACT_WORD	进入直播间
DANMU_MSG	CMD::DANMU_MSG	弹幕消息
SEND_GIFT	CMD::SEND_GIFT	送礼物
COMBO_SEND	CMD::COMBO_SEND	连续送礼物
NOTICE_MSG	CMD::NOTICE_MSG	通知消息
ONLINE_RANK_V2	CMD::ONLINE_RANK_V2	在线 PK

常量列是对应的值在代码中的常量名。

处理弹幕协议

跟协议相关的操作都放在了 Packet 类中，将一些固定的值设置成了类的常量。

/**
 * 头部长度
 */
const HEADER_LEN = 16;

/**
 * 协议版本
 */
const PROTOCOL_VERSION = 2;

/**
 * 魔法数字，设置为 1 即可
 */
const MAGIC_NUMBER = 1;

打包协议

先来看看打包弹幕协议的逻辑，先计算出数据包的总长度，然后将头部信息及数据打包成二进制数据。

public static function pack($opcode, $payload = '')
{
    $packetLen = static::HEADER_LEN;
    if (!empty($payload)) {
        $packetLen += strlen($payload);
    }

    return pack('NnnNN', $packetLen, static::HEADER_LEN, static::PROTOCOL_VERSION, $opcode, static::MAGIC_NUMBER).$payload
}

pack/unpack 函数

这里简单讲下 pack/unpack 函数的使用。

pack 就是将输入参数打包成指定格式的二进制数据，上面的 n、N 就是指定的格式，分别表示无符号短整型(16位，大端字节序)、无符号长整型(32位，大端字节序)。

第一个 N 就是以无符号长整型(32位，大端字节序) 的格式打包数据包总长度。第二个 n 就是以无符号短整型(16位，大端字节序) 的格式打包头部长度。第三个 n 就是以无符号短整型(16位，大端字节序) 的格式打包协议版本号。后面的以此类推…

上面使用的是 PHP 可变参数的方式进行打包，也可以将每个数据单独打包最后再拼在一起，效果也是一样的。

return sprintf(
    '%s%s%s%s%s%s',
    pack('N', $packetLen),
    pack('n', static::HEADER_LEN),
    pack('n', static::PROTOCOL_VERSION),
    pack('N', $opcode),
    pack('N', static::MAGIC_NUMBER),
    $payload
);

更多的介绍可以看 https://www.php.net/manual/zh/function.pack.php

unpack 就是 pack 的反向操作，根据指定的格式将二进制数据解压到数组中。

每条数据以 指定的格式 + key 的方式组成，多条数据用 / 分隔。

举个例子：

$data = pack('Nnn', 2021, 3, 31);

var_dump($data);

$arr = unpack('Nyear/nmonth/nday', $data);

var_dump($arr);

// 输出：

string(8) "\000\000�\000\000"
array(3) {
  'year' => int(2021)
  'month' =>int(3)
  'day' => int(31)
}

打包的时候是按照 Nnn 的格式打包的，所以解压的时候也是按照 Nnn 的格式来的，只不过需要在每个格式的右边指定以这个格式解压出来的数据对应的 key 是什么。

Nyear 就是以无符号长整型(32位，大端字节序) 的格式解压，并将 year 作为该数据的 key。 nmonth 就是以无符号短整型(16位，大端字节序) 的格式解压，并将 month 作为该数据的 key。 …

解压弹幕协议

接下来看看解压弹幕协议的逻辑，其实跟上面说的一样，按照打包的顺序然后指定对应的 key 就可以了。

public static function unpack($data)
{
    if (empty($data)) {
        return [];
    }

    return unpack('Npacket_len/nheader_len/nprotocol_version/Nopcode/Nmagic_number/a*payload', $data);
}

a 表示字符串，* 表示任意长度，更严谨一点应该将 * 改为数据的长度（数据包总长度 - 头部长度）

使用 Node.js 处理协议

这篇文章发出来之后，我试着用 Node.js 来处理弹幕协议，发现写起来是真的舒服。

const PACKET_HEADER_LEN = 16;
const PACKET_PROTOCOL_VERSION = 2;
const PACKET_MAGIC_NUMBER = 1;

class Packet {
    static pack(opcode, payload = '') {
        let packet_len = PACKET_HEADER_LEN;
        if (payload.length > 0) {
            packet_len += payload.length;
        }

        let buffer = Buffer.alloc(packet_len);

        buffer.writeInt32BE(packet_len, 0);
        buffer.writeInt16BE(PACKET_HEADER_LEN, 4);
        buffer.writeInt16BE(PACKET_PROTOCOL_VERSION, 6);
        buffer.writeInt32BE(opcode, 8);
        buffer.writeInt32BE(PACKET_MAGIC_NUMBER, 12);

        if (payload.length > 0) {
            buffer.write(payload, PACKET_HEADER_LEN, payload.length);
        }

        return buffer;
    }

    static unpack(data) {
        let buffer = Buffer.from(data);

        return {
            packet_len: buffer.readInt32BE(0),
            header_len: buffer.readInt16BE(4),
            version: buffer.readInt16BE(6),
            opcode: buffer.readInt32BE(8),
            magic_number: buffer.readInt32BE(12),
            data: buffer.slice(PACKET_HEADER_LEN),
        };
    }
}

与弹幕服务器的交互

接下来看看如何通过弹幕服务器的认证，并在加入房间之后维护在线状态，我将这部分逻辑都放在了 BilibiliBarrage 类中。

获取弹幕服务器信息

在连接弹幕服务器之前，需要通过房间 id 获取到弹幕服务器的地址和端口号，还有认证需要用到的 token。

 const CHAT_CONFIG_URL = 'https://api.live.bilibili.com/room/v1/Danmu/getConf?room_id=%d';

/**
 * 获取直播间配置
 * @param $room_id
 * @return mixed
 * @throws \Exception
 */
public static function getChatConfig($room_id)
{
    if (isset(static::$roomConfigs[$room_id])) {
        return static::$roomConfigs[$room_id];
    }

    $response = file_get_contents(sprintf(self::CHAT_CONFIG_URL, $room_id));
    $response = json_decode($response, true);

    if (empty($response) || $response['code'] != 0) {
        throw new \Exception("Get chat conf failed, reason: {$response['msg']}");
    }

    static::$roomConfigs[$room_id] = $response['data'];

    return $response['data'];
}

接口返回的内容（省略掉了无关的内容）：

{
    "code":0,
    "msg":"ok",
    "message":"ok",
    "data":{
        "refresh_row_factor":0.125,
        "refresh_rate":100,
        "max_delay":5000,
        "port":2243,
        "host":"broadcastlv.chat.bilibili.com",
        "token":"pMF5ippgZMpHIHzTsfKHp9YyqmW3yEfuIuL3pXSMXJ8_9UFN-qSRIPTRxNjpNOr5HQ2-ajI0RcSQkMud2_-lMLoEN92k1glp9fOslshF5SFDqDhlEJRAvUwezoyz72ZdNh-sSqHMPsGOJGMOXZmRNA"
    }
}

认证并加入房间

通过 data 中的 host 和 port 就可以对弹幕服务器发起连接，连接建立后需要发送认证包加入房间。

认证包的内容：

{
  "uid": "0 表示未登录，否则为用户ID",
  "roomid": "房间ID",
  "protover": "协议版本号",
  "platform": "平台",
  "clientver": "客户端版本号",
  "token": "接口返回的 token"
}

认证包的内容就是弹幕协议中携带的数据。

public static function getAuthenticatePacket($room_id, $token = null)
{
    if (empty($token)) {
        $token = static::getChatConfig($room_id)['token'];
    }

    $payload = \json_encode([
        'uid' => 0,
        'roomid' => $room_id,
        'protover' => Packet::PROTOCOL_VERSION,
        'platform' => 'web',
        'token' => $token,
    ]);

    return Packet::pack(Opcode::AUTHENTICATION, $payload);
}

返回的内容：

\000\000\000�\000\000\000\000\000\000\000\000{"uid":0,"roomid":22590309,"protover":2,"platform":"web","token":"pMF5ippgZMpHIHzTsfKHp9YyqmW3yEfuIuL3pXSMXJ8_9UFN-qSRIPTRxNjpNOr5HQ2-ajI0RcSQkMud2_-lMLoEN92k1glp9fOslshF5SFDqDhlEJRAvUwezoyz72ZdNh-sSqHMPsGOJGMOXZmRNA"}

弹幕服务器收到认证包后，会回复我们加入成功的消息，Packet::unpack 后得到消息内容：

array(6) {
  'packet_len' => int(26)
  'header_len' => int(16)
  'protocol_version' => int(2)
  'opcode' => int(8)
  'magic_number' => int(1)
  'payload' => string(10) "{"code":0}"
}

opcode 为 8 表示是服务器发送的心跳包，payload 是一个 JSON 字符串，code 为 0 表示连接成功。

这一步完成之后就可以收到弹幕消息了，但是还差最后一步。

维持在线状态

弹幕服务器要求每隔 30 秒发送一次心跳包，以确定客户端还处于活跃状态。

心跳包没有数据，只需要发送 opcode 为 2 的数据包就可以了。

public static function getHeartBeatPacket()
{
    return Packet::pack(Opcode::CLIENT_HEARTBEAT);
}

考虑到网络传输的因素，心跳包间隔时间一般设置小于 30 秒，防止一些原因导致心跳包没有及时发送。

实现弹幕客户端

可以使用 Workerman、Swoole 甚至 PHP 原生 socket 来实现弹幕客户端，那为啥要用 Workerman 呢？

简单、方便，最重要的是写起来快，不用装扩展也没有原生 socket 那么繁杂，三两下就写完了。

一句话：就是通透

由于篇幅的原因，我会摘取重要的部分来讲，完整的代码可以去 GitHub 获取完整代码。

话不多说，干就完了。

连接弹幕服务器

Worker 进程启动后，通过 AsyncTcpConnection 创建异步 TCP 连接对象。

在 onConnect 回调中发送认证包、开启定时任务，每隔 20 秒发送一次心跳包。

$room_id = 22590309;
/* 获取直播间配置 */
$config = BilibiliBarrage::getChatConfig($room_id);

/* 创建异步 TCP 连接对象 */
$conn = new AsyncTcpConnection("tcp://{$config['host']}:{$config['port']}");

$conn->onConnect = function(TcpConnection $conn) use ($room_id, $config) {
    $packet = BilibiliBarrage::getAuthenticatePacket($room_id, $config['token']);
    /* 发送认证包 */
    $result = $conn->send($packet, true);
    if (!$result) {
        Worker::safeEcho("发送认证包失败\n");
        return;
    }

    /* 开启定时任务 */
    Timer::add(BilibiliBarrage::HEART_BEAT_INTERVAL, function (TcpConnection $conn) {
        /* 发送心跳包 */
        $conn->send(BilibiliBarrage::getHeartBeatPacket(), true);
    }, [$conn]);
};

处理弹幕消息

在 onMessage 回调中，先 unpack 数据，通过 opcode 判断本次消息是做什么的，不同的消息做不同的处理。如果 opcode 为 CMD，需要通过 Packet::parsePayload 解析数据才能得到真正的消息内容。

$conn->onMessage = function($conn, $data) {
    $packet = Packet::unpack($data);
    /* 通过 opcode 判断消息类型 */
    switch ($packet['opcode']) {
        case Opcode::POPULARITY_VALUE:
            Worker::safeEcho(sprintf("人气值: %d\n", Packet::parsePayload($packet['opcode'], $packet['payload'])));
            break;
        case Opcode::CMD:
            /* 解析数据 */
            $payload = Packet::parsePayload($packet['opcode'], $packet['payload']);
            if (empty($payload)) {
                break;
            }

            switch ($payload['cmd']) {
                case 'INTERACT_WORD':
                    Worker::safeEcho("{$payload['data']['uname']} 进入直播间\n");
                    break;
                case 'DANMU_MSG':
                    Worker::safeEcho("{$payload['info'][2][1]}: {$payload['info'][1]}\n");
                    break;
                case 'SEND_GIFT':
                    Worker::safeEcho("{$payload['data']['uname']} {$payload['data']['action']} {$payload['data']['giftName']}\n");
                    break;
                case 'COMBO_SEND':
                    Worker::safeEcho("{$payload['data']['uname']} {$payload['data']['action']} {$payload['data']['gift_name']} [combo]\n");
                    break;
                /* 更多命令查看 \App\CMD.php 文件 */
            }
            break;
        case Opcode::SERVER_HEARTBEAT:
            Worker::safeEcho("加入房间成功\n");
            break;
        default:
            /* 未知的 opcode 可以打印 packet */
            // var_dump($packet);
            break;
    }
};

总结

最后附上一张运行图：

⚠️ 注意！！！本文及源码仅用于学习研究！请勿用于商业或非法目的，否则后果自负。