.NET斗魚直播彈幕客戶端(上)

現在直播平台由於彈幕的存在，主播與觀眾可以更輕松地進行互動，非常受年輕群眾的歡迎。斗魚TV就是一款非常流行的直播平台，彈幕更是非常火爆。看到有不少主播接入彈幕語音播報器、彈幕點歌等模塊，這都需要首先連接斗魚彈幕。

經常看到其它編程語言的開發者，分享了他們斗魚彈幕客戶端的代碼。.NET當然也能做，還能做得更好（只是不知為何很少見人分享😂）。

本文將包含以下內容：

1. 我將使用斗魚TV官方公開的彈幕PDF文檔，使用Socket/TcpClient連續斗魚彈幕；
2. 分析如何利用.NET強大的ValueTask特性，在保持代碼簡潔的同時，輕松享受高性能異步代碼的快樂；
3. 然后將使用Reactive Extensions（RX），演示如何將一系列復雜的彈幕接入操作，就像寫Hello World一般容易；
4. 用我自制的“准游戲引擎”FlysEngine，只需少量代碼，即可將斗魚TV的彈幕顯示左右飛過的效果；

本文內容可能比較多，因此分上、下兩篇闡述，上篇將具體聊聊第1、2點，第3、4點將在下篇進行，整篇完成后，最終效果如下：

斗魚直播API

現在網上可以輕松找到斗魚彈幕服務器第三方接入協議v1.6.2.pdf（網上搜索該關鍵字即可找到）。
文檔提到，第三方接入彈幕服務的服務器為openbarrage.douyutv.com:8601，我們可以使用TcpClient來方便連接：

using (var client = new TcpClient())
{
    client.ConnectAsync("openbarrage.douyutv.com", 8601).Wait();
    Stream stream = client.GetStream();
    // do other works
}

該文檔中提到所有數據包格式如下：

注意前兩個4字節的消息長度是完全一樣的，可以使用Debug.Assert進行斷言。

其中所有數字都為小端整數，剛好.NET的BinaryWriter類默認都以小端整數進行轉換。可以利用起來。

因此，讀取一個消息包的完整代碼如下：

using (var reader = new BinaryReader(stream, Encoding.UTF8, true))
{
    var fullMsgLength = reader.ReadInt32();
    var fullMsgLength2 = reader.ReadInt32();
    Debug.Assert(fullMsgLength == fullMsgLength2);

    var length = fullMsgLength - 1 - 4 - 4;
    var packType = reader.ReadInt16();
    Debug.Assert(packType == ServerSendToClient);
    var encrypted = reader.ReadByte();
    Debug.Assert(encrypted == Encrypted);
    var reserved = reader.ReadByte();
    Debug.Assert(reserved == Reserved);

    var bytes = reader.ReadBytes(length);
    var zero = reader.ReadByte();
    Debug.Assert(zero == ByteZero);
}

其中bytes既是數據部分，根據pdf文檔中的規定，該部分為UTF-8編碼，在C#中使用Encoding.UTF8.GetString()即可獲取其字符串，該字符串長這樣子：

type@=chatmsg/rid@=633019/ct@=1/uid@=124155/nn@=夜科揚羽/txt@=這不壓個蜥蜴/cid@=602c7f1becf2419962a6520300000000/ic@=avatar@S000@S12@S41@S55_avatar/level@=21/sahf@=0/cst@=1570891500125/bnn@=賊開心/bl@=8/brid@=5789561/hc@=21ebd5b2c86c01e0565453e45f14ca5b/el@=/lk@=/urlev@=10/ 

該格式不是JSON/XML等，但仔細分析又確實有邏輯，有層次感，根據文檔，該格式為所謂的STT序列化，該格式包含鍵值對、數組等多種格式。相比JSON可以減少大量的引號"空間開銷。還好協議簡單，我可以通過寥寥幾行代碼，即可轉換為Json.NET的JToken格式：

public static JToken DecodeStringToJObject(string str)
{
    if (str.Contains("//")) // 數組
    {
        var result = new JArray();
        foreach (var field in str.Split(new[] { "//" }, StringSplitOptions.RemoveEmptyEntries))
        {
            result.Add(DecodeStringToJObject(field));
        }
        return result;
    }
    if (str.Contains("@=")) // 對象
    {
        var result = new JObject();
        foreach (var field in str.Split(new[] { '/' }, StringSplitOptions.RemoveEmptyEntries))
        {
            var tokens = field.Split(new[] { "@=" }, StringSplitOptions.None);
            var k = tokens[0];
            var v = UnscapeSlashAt(tokens[1]);
            result[k] = DecodeStringToJObject(v);
        }
        return result;
    }
    else if (str.Contains("@A=")) // 鍵值對
    {
        return DecodeStringToJObject(UnscapeSlashAt(str));
    }
    else
    {
        return UnscapeSlashAt(str); // 值
    }
}

static string EscapeSlashAt(string str)
{
    return str
        .Replace("/", "@S")
        .Replace("@", "@A");
}

static string UnscapeSlashAt(string str)
{
    return str
        .Replace("@S", "/")
        .Replace("@A", "@");
}

這樣一來，即可將STT格式轉換為JSON格式，因此只需像JSON格式取出nn字段和txt字段即可，還有一個col字段，可以用來確定彈幕顏色，我可以將其轉換為RGB的int32值：

Color = (x["col"] ?? new JValue(0)).Value<int>() switch
{
    1 => 0xff0000, // 紅
    2 => 0x1e87f0, // 淺藍
    3 => 0x7ac84b, // 淺綠
    4 => 0xff7f00, // 橙色
    5 => 0x9b39f4, // 紫色
    6 => 0xff69b4, // 洋紅
    _ => 0xffffff, // 默認，白色
}

該代碼使用了C# 8.0的switch expression功能，可以一個表達式轉成整個顏色轉換，比if/else和switch/case語句都精簡不少，可謂一氣呵成。

支持異步/ValueTask/Memory<T>優化

C# 5.0提供了強大的異步API——async/await，通過異步API，以前難以用編程實現的操作現在可以像寫串行代碼一樣輕松完成，還能輕松加入取消任務操作。

然后C# 7.0發布了ValueTask，ValueTask是值類型，因此在頻繁調用異步操作（如使用Stream讀取字節）時，不會因為創建過多的Task而分配沒必要的內存。這里，我確實是使用TCP連接流讀取字節，是使用ValueTask的最佳時機。

這里我們將嘗試將代碼切換為ValueTask版本。

首先第一個問題是BinaryReader類，該類提供了便利的字節操作方式，且能確保字節端為小端，但該類不提供異步API，因此需要作一些特殊處理：

public static async Task<string> RecieveAsync(Stream stream, CancellationToken cancellationToken)
{
    int fullMsgLength = await ReadInt32().ConfigureAwait(false);
    int fullMsgLength2 = await ReadInt32().ConfigureAwait(false);
    Debug.Assert(fullMsgLength == fullMsgLength2);

    int length = fullMsgLength - 1 - 4 - 4;
    short packType = await ReadInt16().ConfigureAwait(false);
    Debug.Assert(packType == ServerSendToClient);
    short encrypted = await ReadByte().ConfigureAwait(false);
    Debug.Assert(encrypted == Encrypted);
    short reserved = await ReadByte().ConfigureAwait(false);
    Debug.Assert(reserved == Reserved);

    Memory<byte> bytes = await ReadBytes(length).ConfigureAwait(false);
    byte zero = await ReadByte().ConfigureAwait(false);
    Debug.Assert(zero == ByteZero);

    return Encoding.UTF8.GetString(bytes.Span);
}

如代碼所示，我封裝了ReadInt16()和ReadInt32()兩個方法，

var intBuffer = new byte[4];
var int32Buffer = new Memory<byte>(intBuffer, 0, 4);

async ValueTask<int> ReadInt32()
{
    var memory = int32Buffer;
    int read = 0;
    while (read < 4)
    {
        read += await stream.ReadAsync(memory.Slice(read), cancellationToken).ConfigureAwait(false);
    }
    Debug.Assert(read == memory.Length);
    return 
        (intBuffer[0] << 0) + 
        (intBuffer[1] << 8) +
        (intBuffer[2] << 16) + 
        (intBuffer[3] << 24);
}

如圖，我還使用了一個while語句，因為不像BinaryReader，如果一次無法讀取所需的字節數（4個字節），stream.ReadAsync()並不會堵塞線程。然后需要將int32Buffer轉換為int類型。

注意：此處我沒有使用BitConverter.ToInt32()，也不能使用該方法，因為該方法不像BinaryReader，它在大端/小端的CPU上會有不同的行為。（其中在大端CPU上將有錯誤的行為）涉及二進制序列化需要傳輸的，不能使用BitConverter類。

同樣的，寫TCP流也需要有相應的變化：

static async Task SendAsync(Stream stream, byte[] body, CancellationToken cancellationToken)
{
    var buffer = new byte[4];
    
    await stream.WriteAsync(GetBytesI32(4 + 4 + body.Length + 1), cancellationToken).ConfigureAwait(false);
    await stream.WriteAsync(GetBytesI32(4 + 4 + body.Length + 1), cancellationToken).ConfigureAwait(false);

    await stream.WriteAsync(GetBytesI16(ClientSendToServer), cancellationToken).ConfigureAwait(false);
    await stream.WriteAsync(new byte[] { Encrypted}, cancellationToken).ConfigureAwait(false);
    await stream.WriteAsync(new byte[] { Reserved}, cancellationToken).ConfigureAwait(false);

    await stream.WriteAsync(body, cancellationToken).ConfigureAwait(false);
    await stream.WriteAsync(new byte[] { ByteZero}, cancellationToken).ConfigureAwait(false);
    
    Memory<byte> GetBytesI32(int v)
    {
        buffer[0] = (byte)v;
        buffer[1] = (byte)(v >> 8);
        buffer[2] = (byte)(v >> 16);
        buffer[3] = (byte)(v >> 24);
        return new Memory<byte>(buffer, 0, 4);
    }

    Memory<byte> GetBytesI16(short v)
    {
        buffer[0] = (byte)v;
        buffer[1] = (byte)(v >> 8);;
        return new Memory<byte>(buffer, 0, 2);
    }
}

總結

最終運行效果如下：

這一篇文章介紹了如何使用斗魚tv開放彈幕API，下篇將會：

• 共享本文所使用的所有完整的源代碼；
• 介紹如何使用Reactive Extensions（RX），演示這一系列操作用起來，就像寫Hello World一樣簡單；
• 用我自制的“准游戲引擎”FlysEngine，只需少量代碼，即可實現桌面彈幕的效果；

敬請期待！“刷一波666🚀🚀🚀”

喜歡的朋友請關注我的微信公眾號：【DotNet騷操作】