StackExchange.Redis 是一個高性能的 Redis 客戶端庫,主要用於 .NET 環境下與 Redis 伺服器進行通信,大名鼎鼎的stackoverflow 網站就使用它。它使用非同步編程模型,能夠高效處理大量請求。支持 Redis 的絕大部分功能,包括發佈/訂閱、事務、Lua 腳本等... ...
StackExchange.Redis 是一個高性能的 Redis 客戶端庫,主要用於 .NET 環境下與 Redis 伺服器進行通信,大名鼎鼎的stackoverflow 網站就使用它。它使用非同步編程模型,能夠高效處理大量請求。支持 Redis 的絕大部分功能,包括發佈/訂閱、事務、Lua 腳本等。由 StackExchange 團隊維護,質量和更新頻率有保障。這篇文章就來給大家分享下 StackExchange.Redis 為什麼玩的這麼溜。
我將通過分析 StackExchange.Redis 中的同步調用和非同步調用邏輯,來給大家一步步揭開它的神秘面紗。
同步API
向Redis發送消息
Redis 客戶端的 Get、Set 等操作都會封裝成為 Message,操作最終會走到這個方法,我們先大致看下代碼:
ConnectionMultiplexer.cs
internal T? ExecuteSyncImpl<T>(Message message, ResultProcessor<T>? processor, ServerEndPoint? server, T? defaultValue = default)
{
...
// 創建一個ResultBox對象,這個對象將會放到Message中用來承載Redis的返回值
var source = SimpleResultBox<T>.Get();
WriteResult result;
// 鎖住ResultBox對象,下邊會有大用
lock (source)
{
// 將Message發送到Redis伺服器
result = TryPushMessageToBridgeSync(message, processor, source, ref server);
...
// 調用 Monitor.Wait 釋放對 ResultBox 對象的鎖,同時讓當前線程停在這裡
if (Monitor.Wait(source, TimeoutMilliseconds))
{
Trace("Timely response to " + message);
}
...
}
// 最終從 ResultBox 取出結果
var val = source.GetResult(out var ex, canRecycle: true);
...
return val;
...
}
仔細說一下大概的處理邏輯。
- 先構造一個ResultBox對象,用來承載Message的執行結果。
- 然後嘗試把這個Message推送到Redis伺服器,註意程式內部會把當前Message和ResultBox做一個綁定。
- 等待Redis伺服器返回,返回結果賦值到ResultBox對象上。
- 最後從ResultBox對象中取出結果,返回給調用方。
註意這裡用到了鎖(lock),還使用了Monitor.Wait,這是什麼目的呢?
Monitor.Wait 一般和 Monitor.Pulse 搭配使用,用來線上程間通信。
- 調用 Monitor.Wait 時,lock住的ResultBox會被釋放,同時當前線程就會掛起,停在這裡。
- Redis伺服器返回結果後,把結果數據賦值到ResultBox上。
- 其它線程lock住這個ResultBox,調用Monitor.Pulse,之前被掛起的線程繼續執行。
通過這種方式,我們就達成了一個跨線程的同步調用效果。
為什麼會跨線程呢?直接調用Redis等著返回結果不行嗎?
因為 StackExchange.Redis 底層使用了 System.IO.Pipelines 來優化網路IO,這個庫採用了生產者/消費者的非同步模式來處理網路請求和響應,發送數據和接收數據很可能是在不同的線程中。
以上就是向Redis伺服器發送消息的一個巨集觀理解,但是這裡有一個隱藏的問題:
非同步情況下怎麼把Redis的返回結果和消息對應上?
我們繼續跟蹤向 Redis 伺服器發送 Message 的代碼,也就是深入 TryPushMessageToBridgeSync 的內部。
一路跟隨,代碼會走到這裡:
PhysicalBridge.cs
internal WriteResult WriteMessageTakingWriteLockSync(PhysicalConnection physical, Message message)
{
...
bool gotLock = false;
try
{
...
// 獲取單寫鎖,同時只能寫一個Message
gotLock = _singleWriterMutex.Wait(0);
if (!gotLock)
{
gotLock = _singleWriterMutex.Wait(TimeoutMilliseconds);
if (!gotLock) return TimedOutBeforeWrite(message);
}
...
// 繼續調用內部方法寫數據
WriteMessageInsideLock(physical, message);
...
// 刷新網路管道,將數據通過網路發出去
physical.FlushSync(false, TimeoutMilliseconds);
}
catch (Exception ex) { ... }
finally
{
if (gotLock)
{
_singleWriterMutex.Release();
}
}
}
這裡邊用信號量做了一個鎖,保證同時只有一個寫操作。
那麼為什麼要保證同時只能一個寫操作呢?
我們繼續跟蹤代碼:
private WriteResult WriteMessageToServerInsideWriteLock(PhysicalConnection connection, Message message)
{
...
// 把消息添加到隊列
connection.EnqueueInsideWriteLock(message);
// 把消息寫到網路介面
message.WriteTo(connection);
...
}
這裡有兩個操作,一是將Message添加到隊列,二是向網路介面寫數據。
保證同時只有一個寫操作,或者加鎖的目的,就是讓它倆一起完成,能對應起來,不會錯亂。
那麼我們還要繼續問:寫隊列和寫網路對應起來有什麼用?
這個問題不好回答,我們先來看看這兩個操作都是乾什麼用的?
為什麼要把Message寫入隊列?
同步IO可以直接拿到當前消息的返回結果,但是 System.IO.Pipelines 底層是非同步操作,當處理結果從Redis返回時,我們需要把它對應到一個Messge上。加入隊列就是為了方便找到對應的消息。至於為什麼用隊列,而不用集合,因為隊列能夠很好的滿足這個需求,下邊會有說明。
寫隊列代碼在這裡:
PhysicalConnection.cs
internal void EnqueueInsideWriteLock(Message next)
{
...
bool wasEmpty;
lock (_writtenAwaitingResponse)
{
...
_writtenAwaitingResponse.Enqueue(next);
}
...
}
入隊列需要先加鎖,因為可能是多線程環境下操作,Queue自身不是線程安全的。
再看一下把消息寫到網路介面,這個的目的就是把消息發送到Redis伺服器,看一下代碼:
PhysicalConnection.cs
internal static void WriteUnifiedPrefixedString(PipeWriter? maybeNullWriter, byte[]? prefix, string? value)
{
...
// writer 就是管道的寫入介面
var span = writer.GetSpan(3 + Format.MaxInt32TextLen);
span[0] = (byte)'$';
int bytes = WriteRaw(span, totalLength, offset: 1);
writer.Advance(bytes);
if (prefixLength != 0) writer.Write(prefix);
if (encodedLength != 0) WriteRaw(writer, value, encodedLength);
WriteCrlf(writer);
...
}
源碼最底層是通過 System.IO.Pipelines 中的 PipeWriter 把 Message 命令發送到Redis伺服器的,這段代碼比較複雜,大家先大概知道做什麼用的就行了。
到此,向Redis發送消息就處理完成了。
現在我們已經大概瞭解向Redis伺服器發送消息的過程:在最上層通過Monitor模擬了同步操作,在最底層使用了高效的非同步IO,為了適配同步和非同步,寫操作內含了兩個子操作:寫隊列和寫網路。
但是我們仍然不能回答一個問題:寫隊列和寫網路為什麼要放到一個鎖中執行?或者說為什麼要保證同時只能一個寫操作?
要回答這個問題,我們還得繼續看程式對Redis響應結果的處理。
處理Redis響應結果
Redis 客戶端與 Redis 伺服器建立連接時,會創建一個死迴圈,持續的從 System.IO.Pipelines 的管道中讀取Redis 伺服器返回的消息,併進行相應的處理。最上層方法就是這個 ReadFromPipe:
PhysicalConnection.cs
private async Task ReadFromPipe()
{
...
while (true)
{
...
// 沒有新數據從Redis伺服器返回時,ReadAsync會等在這裡
readResult = await input.ReadAsync().ForAwait();
...
var buffer = readResult.Buffer;
...
if (!buffer.IsEmpty)
{
// 這裡邊解析數據,並賦值到相關對象上
handled = ProcessBuffer(ref buffer);
}
}
}
對返回數據的處理重點在這個 ProcessBuffer 方法中。它會先對數據進行一個簡單的解析,然後再調用 MatchResult,從字面義上看就是匹配結果,匹配到那個結果呢?
private int ProcessBuffer(ref ReadOnlySequence<byte> buffer)
{
...
var reader = new BufferReader(buffer);
var result = TryParseResult(_protocol >= RedisProtocol.Resp3, _arena, in buffer, ref reader, IncludeDetailInExceptions, this);
...
MatchResult(result);
...
}
還記得我們在上邊向Redis發送Message前,先創建了一個 ResultBox 對象,匹配的就是它。
怎麼找到對應的 ResultBox 對象呢?
看下邊的代碼,程式從隊列中取出了一個Message 實例,就是要匹配到這個 Message 實例關聯的ResultBox。
private void MatchResult(in RawResult result)
{
...
// 從隊列中取出最早的一條Redis操作消息
lock (_writtenAwaitingResponse)
{
if (!_writtenAwaitingResponse.TryDequeue(out msg))
{
throw new InvalidOperationException("Received response with no message waiting: " + result.ToString());
}
}
...
// 將Redis返回的結果設置到取出的消息中
if (msg.ComputeResult(this, result))
{
_readStatus = msg.ResultBoxIsAsync ? ReadStatus.CompletePendingMessageAsync : ReadStatus.CompletePendingMessageSync;
// 完成Redis操作
msg.Complete();
}
...
}
為什麼從隊列取出的 Message 就一定能對應到 Redis 伺服器當前返回的結果呢?
要破案了,還記得上邊的那個未解問題嗎:為什麼要保證同時只能一個寫操作?
我們每次操作Redis都是:先把Message壓入隊列,然後再發送到Redis伺服器,這兩個操作緊密的綁定在一起;而Redis伺服器是單線程順序處理的,最先返回的就是最早壓入隊列的。加上每次只有一個寫操作的控制,我們就能保證最先寫入隊列的(也就是最先發到Redis伺服器的)Message,就能對應到最先從Redis伺服器返回的數據。
上面這段程式中的 msg.ComputeResult 就是用來將 Redis 最新返回的數據賦值到最新從隊列中拿出來的 Message 實例中。
現在 Message 實例 已經獲取到了 Redis返回結果,還記得之前的發送線程一直在掛起等待嗎?
下邊的 msg.Complete 就是來讓發送線程恢復執行的,看這個代碼 :
Message.cs(Message)
public void Complete()
{
...
// ResultBox激活繼續處理
currBox?.ActivateContinuations();
}
還有一層封裝,繼續看這個代碼:
ResultBox.cs(SimpleResultBox)。
void IResultBox.ActivateContinuations()
{
lock (this)
{
// 通知等待Redis響應的線程,Redis返回結果了,請繼續你的表演
Monitor.PulseAll(this);
}
...
}
Monitor.PulseAll 一齣,發送線程立馬恢復執行,向調用方返回執行結果。
一次同步調用就這樣完成了。
非同步API
非同步API和同步API使用相同的通信底層,包括寫隊列和寫網路管道的處理,只是在處理返回值的方式上存在不同。大家可以看一下非同步和同步調試堆棧的對比圖:
執行到 PhysicalBridge.WriteMessageInsideLock 這一步時處理就同步了。這一步的代碼上邊也貼過了,這裡再給大家看看:其中的主要邏輯就是寫隊列和寫網路管道。
private WriteResult WriteMessageToServerInsideWriteLock(PhysicalConnection connection, Message message)
{
...
// 把消息添加到隊列
connection.EnqueueInsideWriteLock(message);
// 把消息寫到網路介面
message.WriteTo(connection);
...
}
向Redis發送消息
我們再簡單看看非同步API中是如何發送消息的,看代碼:
internal Task<T?> ExecuteAsyncImpl<T>(Message? message, ResultProcessor<T>? processor, object? state, ServerEndPoint? server)
{
...
// 創建一個Task執行狀態跟蹤對象
TaskCompletionSource<T?>? tcs = null;
// 創建一個ResultBox對象,這個對象將會放到Message中用來承載Redis的返回值
// 非同步這裡特別將 ResultBox 和 TaskCompletionSource 綁定到了一起
// 獲取到Redis伺服器返回的數據後,TaskCompletionSource 的執行狀態將被更新為完成
IResultBox<T?>? source = null;
if (!message.IsFireAndForget)
{
source = TaskResultBox<T?>.Create(out tcs, state);
}
// 將Message消息發送到 Redis伺服器
var write = TryPushMessageToBridgeAsync(message, processor, source!, ref server);
...
// 返回Task,調用方可以 await
return tcs.Task;
}
相比同步API,這裡多創建了一個 TaskCompletionSource 的實例,它用來跟蹤非同步任務的執行狀態,程式會在接收到Redis伺服器的返回數據時,將 TaskCompletionSource 的狀態更新為完成執行。
裡邊的代碼我就不展開講了,大家有興趣的可以按照上方我截圖的調用堆棧去跟蹤下。
處理Redis響應結果
非同步API和同步API使用同一個死迴圈方法:ReadFromPipe,程式啟動時也只有這一個死迴圈在運行。
代碼上邊都講過了,這裡只說下最後“ResultBox激活繼續處理”的部分,這個 ResultBox 和同步調用的 ResultBox 略有不同,看代碼:
void IResultBox.ActivateContinuations()
{
...
ActivateContinuationsImpl();
}
private void ActivateContinuationsImpl()
{
var val = _value;
...
TrySetResult(val);
...
}
public bool TrySetResult(TResult result)
{
// 設置非同步任務執行完成
bool rval = _task.TrySetResult(result);
...
return rval;
}
最重要的就是 _task.TrySetResult 這句,這裡的 _task 就是發起非同步調用時創建的 TaskCompletionSource 實例,TrySetResult 的作用就是設置非同步任務執行完成,對應的 await 代碼就可以繼續向下執行了。
await client.SetAsync("hello", "fireflysoft.net");
// 繼續執行下邊的代碼
...
總結
總體執行邏輯
通過對同步API、非同步API的執行邏輯分析,我這裡總結了一張圖,可以讓大家快速的理清其中的處理邏輯。
我再用文字描述下這個執行邏輯:
1、無論是同步調用還是非同步調用,StackExchange.Redis 底層都是先會創建一個 Message 對象;每個 Message 對象都會關聯一個ResultBox對象(同步和非同步調用對應的ResultBox對象略有不同),這個對象用來承載Redis執行結果;
2、然後程式會把Message存入隊列、發送到網路IO管道,寫隊列和寫網路IO放到了一個互斥鎖中,同時只有一個Message寫入,這是為了保證收到Redis響應時正好對應隊列中的第一條數據。
執行完這些操作後,API會等待,但是同步調用和非同步調用等待的方式不同,同步會掛起線程等待其它線程同步結果,非同步會使用await等待Task執行結果;
3、Redis 命令被髮送到網路,抵達Redis伺服器
4、接收到Redis伺服器的響應數據,這些數據會放到網路IO管道中。
5、有一個線程持續監聽IO管道中收到的數據,一旦拿到數據,就去隊列中取出一個Message,把伺服器返回的數據寫到這個Message的ResultBox中。
給ResultBox賦值完,程式還會通知等待的API繼續執行,同步調用是通過線程通信的方式通知,非同步調用是通過更新Task的執行結果狀態來通知。
最後API從ResultBox中取出數據返回給調用方。
管道技術
無論是同步調用還是非同步調用,它們的底層通信方式都統一到了管道技術,這是 StackExchange.Redis 性能出類拔萃的根基,這部分就專門來介紹下。
這裡說的管道技術指的是使用System.IO.Pipelines庫,這個庫提供了一種高效的方式來優化流式數據處理,具備更高的吞吐量、更低的延遲。具體用途:網路上,可以用來構建高性能的TCP或UDP伺服器;對於大文件的讀寫操作,使用Pipelines可以減少記憶體占用,提高處理速度。
PipeWriter和PipeReader是System.IO.Pipelines中的核心組件,它們用於構建管道處理數據流。這裡分享個例子:
using System;
using System.IO.Pipelines;
using System.Text;
using System.Threading.Tasks;
class Program
{
static async Task Main(string[] args)
{
// 創建一個管道
var pipe = new Pipe();
// 啟動一個任務來寫入數據
var writing = FillPipeAsync(pipe.Writer);
// 啟動一個任務來讀取數據
var reading = ReadPipeAsync(pipe.Reader);
await Task.WhenAll(reading, writing);
}
private static async Task FillPipeAsync(PipeWriter writer)
{
for (int i = 0; i < 5; i++)
{
// 寫入一些數據到管道中
string message = $"Message {i}";
byte[] messageBytes = Encoding.UTF8.GetBytes(message);
// 將數據寫入管道
Memory<byte> memory = writer.GetMemory(messageBytes.Length);
messageBytes.CopyTo(memory);
writer.Advance(messageBytes.Length);
// 通知管道有數據寫入
FlushResult result = await writer.FlushAsync();
if (result.IsCompleted)
{
break;
}
// 模擬一些延遲
await Task.Delay(500);
}
// 告訴管道我們已經完成寫入
await writer.CompleteAsync();
}
private static async Task ReadPipeAsync(PipeReader reader)
{
while (true)
{
// 讀取管道中的數據
ReadResult result = await reader.ReadAsync();
var buffer = result.Buffer;
// 處理讀取到的數據
foreach (var segment in buffer)
{
string message = Encoding.UTF8.GetString(segment.Span);
Console.WriteLine($"Read: {message}");
}
// 告訴管道我們已經處理了這些數據
reader.AdvanceTo(buffer.End);
// 如果沒有更多數據可以讀取,退出迴圈
if (result.IsCompleted)
{
break;
}
}
// 告訴管道我們已經完成讀取
await reader.CompleteAsync();
}
}
在這個示例中,我們創建了一個 Pipe 對象,並分別啟動了兩個任務來寫入和讀取數據:
- FillPipeAsync 方法中,使用 PipeWriter 寫入數據到管道。
- ReadPipeAsync 方法中,使用 PipeReader 從管道中讀取數據並處理。
通過這種方式,我們可以高效地處理流式數據,同時利用管道的優勢來提高吞吐量和降低延遲。
其實在很多的高性能IO庫中,使用的都是管道技術,比如Java的NIO、Windows的IOCP、Linux的epoll,本質上都是通過一個類似管道的東西來統籌管理數據傳輸,減少不必要的調用和檢查,達到高效通信的目的。
以上就是本文的主要內容,如有問題,歡迎討論交流!
![](https://images.cnblogs.com/cnblogs_com/bossma/2327976/o_230719042205_qrcode1689740404996.jpg)