現在直播平臺由於彈幕的存在，主播與觀眾可以更輕鬆地進行互動，非常受年輕群眾的歡迎。鬥魚TV就是一款非常流行的直播平臺，彈幕更是非常火爆。看到有不少主播接入 彈幕語音播報器、 彈幕點歌等模塊，這都需要首先連接鬥魚彈幕。
經常看到其它程式語言的開發者，分享了他們鬥魚彈幕客戶端的代碼。.NET當然也能做，還能做得更好（只是不知為何很少見人分享😂）。
本文將包含以下內容：
我將使用鬥魚TV官方公開的彈幕PDF文檔，使用 Socket/ TcpClient連續鬥魚彈幕；
分析如何利用 .NET強大的 ValueTask特性，在保持代碼簡潔的同時，輕鬆享受高性能異步代碼的快樂；
然後將使用 ReactiveExtensions（ RX），演示如何將一系列複雜的彈幕接入操作，就像寫 HelloWorld一般容易；
用我自製的「準遊戲引擎」 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的 switchexpression功能，可以一個表達式轉成整個顏色轉換，比 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);
}
}
總結最終運行效果如下：
這一篇【DotNet騷操作】文章介紹了如何使用鬥魚tv開放彈幕 API，下篇將會：
敬請期待！「刷一波666🚀🚀🚀」
喜歡的朋友請關注我的微信公眾號：【DotNet騷操作】