基于C# Socket实现的简单的Redis客户端 - ToB企服应用市场:ToB评测及商务社交产业平台

public class RedisClient : IDisposable, IAsyncDisposable
{
//定义默认端口
private readonly int DefaultPort = 6379;
//定义默认地址
private readonly string Host = "localhost";
//心跳间隔，单位为毫秒
private readonly int HeartbeatInterval = 30000;
private bool _isConnected;
//心跳定时器
private Timer _heartbeatTimer;
private Socket _socket;
public RedisClient(string host = "localhost", int defaultPort = 6379)
{
Host = host;
DefaultPort = defaultPort;
// 初始化心跳定时器
_heartbeatTimer = new Timer(HeartbeatCallback, null, HeartbeatInterval, HeartbeatInterval);
}
//连接方法
public async Task ConnectAsync(int timeoutMilliseconds = 5000)
{
_socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
var cts = new CancellationTokenSource(timeoutMilliseconds);
await _socket.ConnectAsync(Host, DefaultPort, cts.Token);
_isConnected = true;
}
//心跳方法
private async void HeartbeatCallback(object state)
{
if (_isConnected)
{
var pingCommand = "PING\r\n";
await SendCommandAsync(pingCommand);
}
}
//释放逻辑
public void Dispose()
{
// 停止心跳定时器
_heartbeatTimer.Dispose();
if (_socket != null)
{
_socket.Shutdown(SocketShutdown.Both);
_socket.Close();
}
}
public ValueTask DisposeAsync()
{
Dispose();
return ValueTask.CompletedTask;
}
}

复制代码

//发送命令
public async Task<string> SendCommandAsync(string command)
{
// 发送命令的实现
if (!_isConnected)
{
// 如果连接已断开，可以进行重连
await ConnectAsync();
}
//Redis的命令是以\r\n为结尾的
var request = Encoding.UTF8.GetBytes(command + "\r\n");
//发送命令
await _socket.SendAsync(new ArraySegment<byte>(request), SocketFlags.None);
var response = new StringBuilder();
var remainingData = string.Empty;
//初始化响应字符串和剩余数据
byte[] receiveBuffer = ArrayPool<byte>.Shared.Rent(1024);
try
{
while (true)
{
//读取返回信息
var bytesRead = await _socket.ReceiveAsync(new ArraySegment<byte>(receiveBuffer), SocketFlags.None);
//将接收到的数据添加到响应字符串
var responseData = remainingData + Encoding.UTF8.GetString(receiveBuffer, 0, bytesRead);
//提取完整的响应并添加到响应字符串中
var completeResponses = ExtractCompleteResponses(ref responseData);
foreach (var completeResponse in completeResponses)
{
response.Append(completeResponse);
}
remainingData = responseData;
//结果为\r\n读取结束
if (response.ToString().EndsWith("\r\n"))
{
break;
}
}
}
finally
{
//释放缓冲区
ArrayPool<byte>.Shared.Return(receiveBuffer);
}
//返回完整的响应字符串
return response.ToString();
}
private List<string> ExtractCompleteResponses(ref string data)
{
var completeResponses = new List<string>();
while (true)
{
var index = data.IndexOf("\r\n");
if (index >= 0)
{
// 提取一个完整的响应
var completeResponse = data.Substring(0, index + 2);
//将完整的响应添加到列表中
completeResponses.Add(completeResponse);
data = data.Substring(index + 2);
}
else
{
break;
}
}
return completeResponses;
}
private string ParseResponse(string response)
{
if (response.StartsWith("$"))
{
// 处理 Bulk Strings（$）
var lengthStr = response.Substring(1, response.IndexOf('\r') - 1);
if (int.TryParse(lengthStr, out int length))
{
if (length == -1)
{
return null!;
}
string rawRedisData = response.Substring(response.IndexOf('\n') + 1);
byte[] utf8Bytes = Encoding.UTF8.GetBytes(rawRedisData);
string value = Encoding.UTF8.GetString(utf8Bytes, 0, length);
return value;
}
}
else if (response.StartsWith("+"))
{
// 处理 Simple Strings（+）
return response.Substring(1, response.Length - 3);
}
else if (response.StartsWith(":"))
{
// 处理 Integers（:）
var valueStr = response.Substring(1, response.IndexOf('\r') - 1);
if (int.TryParse(valueStr, out int value))
{
return value.ToString();
}
}
// 如果响应格式不符合预期，抛出异常
throw new InvalidOperationException(response);
}

复制代码

//切换db操作
public async Task SelectAsync(int dbIndex)
{
var command = $"SELECT {dbIndex}";
await SendCommandAsync(command);
}
//get操作
public async Task<string> GetAsync(string key)
{
var command = $"GET {key}";
return ParseResponse(await SendCommandAsync(command));
}
//set操作
public async Task<bool> SetAsync(string key, string value, TimeSpan? expiry = null)
{
var command = $"SET {key} '{value}'";
//判断会否追加过期时间
if (expiry.HasValue)
{
command += $" EX {expiry.Value.TotalSeconds}";
}
var response = ParseResponse(await SendCommandAsync(command));
return response == "OK";
}
//支持过期时间的setnx操作
public async Task<bool> SetNxAsync(string key, string value, TimeSpan? expiry = null)
{
//因为默认的setnx方法不支持添加过期时间，为了保证操作的原子性，使用了lua
var command = $"EVAL "if redis.call('SETNX', KEYS[1], ARGV[1]) == 1 then if ARGV[2] then redis.call('EXPIRE', KEYS[1], ARGV[2]) end return true else return false end" 1 {key} '{value}'";
if (expiry.HasValue)
{
command += $" {expiry.Value.TotalSeconds}";
}
var response = ParseResponse(await SendCommandAsync(command));
return response == "1";
}
//添加支持函过期时间的list push操作
public async Task<long> ListPushAsync(string key, string value, TimeSpan? expiry = null)
{
var script = @"local len = redis.call('LPUSH', KEYS[1], ARGV[1])
if tonumber(ARGV[2]) > 0 then
redis.call('EXPIRE', KEYS[1], ARGV[2])
end
return len";
var keys = new string[] { key };
var args = new string[] { value, (expiry?.TotalSeconds ?? 0).ToString() };
var response = await ExecuteLuaScriptAsync(script, keys, args);
return long.Parse(response);
}
//list pop操作
public async Task<string> ListPopAsync(string key)
{
var command = $"LPOP {key}";
return ParseResponse(await SendCommandAsync(command));
}
//listrange操作
public async Task<List<string>> ListRangeAsync(string key, int start, int end)
{
var command = $"LRANGE {key} {start} {end}";
var response = await SendCommandAsync(command);
if (response.StartsWith("*0\r\n"))
{
return new List<string>();
}
//由于list range返回了是一个数组，所以单独处理了一下，这里我使用了正则，解析字符串也可以，方法随意
var values = new List<string>();
var pattern = @"\$\d+\r\n(.*?)\r\n";
MatchCollection matches = Regex.Matches(response, pattern);
foreach (Match match in matches)
{
values.Add(match.Groups[1].Value);
}
return values;
}
//执行lua脚本的方法
public async Task<string> ExecuteLuaScriptAsync(string script, string[]? keys = null, string[]? args = null)
{
//去除lua里的换行
script = Regex.Replace(script, @"[\r\n]", "");
// 构建EVAL命令，将Lua脚本、keys和args发送到Redis服务器
var command = $"EVAL "{script}" { keys?.Length??0 } ";
//拼接key和value参数
if (keys != null && keys.Length != 0)
{
command += string.Join(" ", keys.Select(key => $"{key}"));
}
if (args != null && args.Length != 0)
{
command += " " + string.Join(" ", args.Select(arg => $"{arg}"));
}
return ParseResponse(await SendCommandAsync(command));
}
//redis发布操作
public async Task SubscribeAsync(string channel, Action<string, string> handler)
{
await SendCommandAsync($"SUBSCRIBE {channel}");
while (true)
{
var response = await SendCommandAsync(string.Empty);
string pattern = @"\*\d+\r\n\$\d+\r\n(.*?)\r\n\$\d+\r\n(.*?)\r\n\$\d+\r\n(.*?)\r\n";
Match match = Regex.Match(response, pattern);
if (match.Success)
{
string ch = match.Groups[2].Value;
string message = match.Groups[3].Value;
handler(ch, message);
}
}
}
//redis订阅操作
public async Task PublishAsync(string channel, string message)
{
await SendCommandAsync($"PUBLISH {channel} {message}");
}

复制代码

public class RedisClient : IDisposable, IAsyncDisposable
{
private readonly int DefaultPort = 6379;
private readonly string Host = "localhost";
private readonly int HeartbeatInterval = 30000;
private bool _isConnected;
private Timer _heartbeatTimer;
private Socket _socket;
public RedisClient(string host = "localhost", int defaultPort = 6379)
{
Host = host;
DefaultPort = defaultPort;
_heartbeatTimer = new Timer(HeartbeatCallback, null, HeartbeatInterval, HeartbeatInterval);
}
public async Task ConnectAsync(int timeoutMilliseconds = 5000)
{
_socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
var cts = new CancellationTokenSource(timeoutMilliseconds);
await _socket.ConnectAsync(Host, DefaultPort, cts.Token);
_isConnected = true;
}
public async Task SelectAsync(int dbIndex)
{
var command = $"SELECT {dbIndex}";
await SendCommandAsync(command);
}
public async Task<string> GetAsync(string key)
{
var command = $"GET {key}";
return ParseResponse(await SendCommandAsync(command));
}
public async Task<bool> SetAsync(string key, string value, TimeSpan? expiry = null)
{
var command = $"SET {key} '{value}'";
if (expiry.HasValue)
{
command += $" EX {expiry.Value.TotalSeconds}";
}
var response = ParseResponse(await SendCommandAsync(command));
return response == "OK";
}
public async Task<bool> SetNxAsync(string key, string value, TimeSpan? expiry = null)
{
var command = $"EVAL "if redis.call('SETNX', KEYS[1], ARGV[1]) == 1 then if ARGV[2] then redis.call('EXPIRE', KEYS[1], ARGV[2]) end return true else return false end" 1 {key} '{value}'";
if (expiry.HasValue)
{
command += $" {expiry.Value.TotalSeconds}";
}
var response = ParseResponse(await SendCommandAsync(command));
return response == "1";
}
public async Task<long> ListPushAsync(string key, string value, TimeSpan? expiry = null)
{
var script = @"local len = redis.call('LPUSH', KEYS[1], ARGV[1])
if tonumber(ARGV[2]) > 0 then
redis.call('EXPIRE', KEYS[1], ARGV[2])
end
return len";
var keys = new string[] { key };
var args = new string[] { value, (expiry?.TotalSeconds ?? 0).ToString() };
var response = await ExecuteLuaScriptAsync(script, keys, args);
return long.Parse(response);
}
public async Task<string> ListPopAsync(string key)
{
var command = $"LPOP {key}";
return ParseResponse(await SendCommandAsync(command));
}
public async Task<long> ListLengthAsync(string key)
{
var command = $"LLEN {key}";
return long.Parse(ParseResponse(await SendCommandAsync(command)));
}
public async Task<List<string>> ListRangeAsync(string key, int start, int end)
{
var command = $"LRANGE {key} {start} {end}";
var response = await SendCommandAsync(command);
if (response.StartsWith("*0\r\n"))
{
return new List<string>();
}
var values = new List<string>();
var pattern = @"\$\d+\r\n(.*?)\r\n";
MatchCollection matches = Regex.Matches(response, pattern);
foreach (Match match in matches)
{
values.Add(match.Groups[1].Value);
}
return values;
}
public async Task<string> ExecuteLuaScriptAsync(string script, string[]? keys = null, string[]? args = null)
{
script = Regex.Replace(script, @"[\r\n]", "");
var command = $"EVAL "{script}" { keys?.Length??0 } ";
if (keys != null && keys.Length != 0)
{
command += string.Join(" ", keys.Select(key => $"{key}"));
}
if (args != null && args.Length != 0)
{
command += " " + string.Join(" ", args.Select(arg => $"{arg}"));
}
return ParseResponse(await SendCommandAsync(command));
}
public async Task SubscribeAsync(string channel, Action<string, string> handler)
{
await SendCommandAsync($"SUBSCRIBE {channel}");
while (true)
{
var response = await SendCommandAsync(string.Empty);
string pattern = @"\*\d+\r\n\$\d+\r\n(.*?)\r\n\$\d+\r\n(.*?)\r\n\$\d+\r\n(.*?)\r\n";
Match match = Regex.Match(response, pattern);
if (match.Success)
{
string ch = match.Groups[2].Value;
string message = match.Groups[3].Value;
handler(ch, message);
}
}
}
public async Task PublishAsync(string channel, string message)
{
await SendCommandAsync($"PUBLISH {channel} {message}");
}
public async Task<string> SendCommandAsync(string command)
{
if (!_isConnected)
{
await ConnectAsync();
}
var request = Encoding.UTF8.GetBytes(command + "\r\n");
await _socket.SendAsync(new ArraySegment<byte>(request), SocketFlags.None);
var response = new StringBuilder();
var remainingData = string.Empty;
byte[] receiveBuffer = ArrayPool<byte>.Shared.Rent(1024);
try
{
while (true)
{
var bytesRead = await _socket.ReceiveAsync(new ArraySegment<byte>(receiveBuffer), SocketFlags.None);
var responseData = remainingData + Encoding.UTF8.GetString(receiveBuffer, 0, bytesRead);
var completeResponses = ExtractCompleteResponses(ref responseData);
foreach (var completeResponse in completeResponses)
{
response.Append(completeResponse);
}
remainingData = responseData;
if (response.ToString().EndsWith("\r\n"))
{
break;
}
}
}
finally
{
ArrayPool<byte>.Shared.Return(receiveBuffer);
}
return response.ToString();
}
private List<string> ExtractCompleteResponses(ref string data)
{
var completeResponses = new List<string>();
while (true)
{
var index = data.IndexOf("\r\n");
if (index >= 0)
{
var completeResponse = data.Substring(0, index + 2);
completeResponses.Add(completeResponse);
data = data.Substring(index + 2);
}
else
{
break;
}
}
return completeResponses;
}
private string ParseResponse(string response)
{
if (response.StartsWith("$"))
{
var lengthStr = response.Substring(1, response.IndexOf('\r') - 1);
if (int.TryParse(lengthStr, out int length))
{
if (length == -1)
{
return null!;
}
string rawRedisData = response.Substring(response.IndexOf('\n') + 1);
byte[] utf8Bytes = Encoding.UTF8.GetBytes(rawRedisData);
string value = Encoding.UTF8.GetString(utf8Bytes, 0, length);
return value;
}
}
else if (response.StartsWith("+"))
{
return response.Substring(1, response.Length - 3);
}
else if (response.StartsWith(":"))
{
var valueStr = response.Substring(1, response.IndexOf('\r') - 1);
if (int.TryParse(valueStr, out int value))
{
return value.ToString();
}
}
throw new InvalidOperationException(response);
}
private async void HeartbeatCallback(object state)
{
if (_isConnected)
{
var pingCommand = "PING\r\n";
await SendCommandAsync(pingCommand);
}
}
public void Dispose()
{
_heartbeatTimer.Dispose();
if (_socket != null)
{
_socket.Shutdown(SocketShutdown.Both);
_socket.Close();
}
}
public ValueTask DisposeAsync()
{
Dispose();
return ValueTask.CompletedTask;
}
}

复制代码

using RedisClient redisClient = new RedisClient();
await redisClient.ConnectAsync();
//切换db
await redisClient.SelectAsync(3);
bool setResult = await redisClient.SetAsync("key:foo", "are you ok,你好吗？", TimeSpan.FromSeconds(120));
string getResult = await redisClient.GetAsync("key:foo");
Console.WriteLine("get key:foo:" + getResult);

复制代码

//第一次setnx返回true
bool setNxResult = await redisClient.SetNxAsync("order:lock", "123_lock", TimeSpan.FromSeconds(120));
Console.WriteLine("first setnx order:lock:" + setNxResult);
//第一次setnx返回false
setNxResult = await redisClient.SetNxAsync("order:lock", "123_lock", TimeSpan.FromSeconds(120));
Console.WriteLine("second setnx aname:foo:" + setNxResult);

复制代码

_ = redisClient.SubscribeAsync("order_msg_ch", (ch, msg) => { Console.WriteLine($"接收消息:[{ch}]---[{msg}]"); });
Thread.Sleep(2000);
using RedisClient redisClient2 = new RedisClient();
await redisClient2.ConnectAsync();
for (int i = 0; i < 5; i++)
{
await redisClient2.PublishAsync("order_msg_ch", $"发送消息{i}");
Thread.Sleep(2000);
}

复制代码

//抢单的lua
string takeOrderLuaScript = @"
local ordersTaken = tonumber(redis.call('GET', KEYS[1]) or '0')
if ordersTaken < tonumber(ARGV[1]) then
redis.call('INCR', KEYS[1])
return 1
else
return 0
end";
//完成你手里的订单操作
string completeOrderLuaScript = @"
local ordersTaken = tonumber(redis.call('GET', KEYS[1]) or '0')
if ordersTaken > 0 then
redis.call('DECR', KEYS[1])
return 1
else
return 0
end";
//模拟抢单，最多抢两单
string result = await redisClient.ExecuteLuaScriptAsync(takeOrderLuaScript, new[] { "user:123" }, new[] { "2" });
result = await redisClient.ExecuteLuaScriptAsync(takeOrderLuaScript, new[] { "user:123" }, new[] { "2" });
result = await redisClient.ExecuteLuaScriptAsync(takeOrderLuaScript, new[] { "user:123" }, new[] { "2" });
result = await redisClient.ExecuteLuaScriptAsync(takeOrderLuaScript, new[] { "user:123" }, new[] { "2" });
//完成订单
string anotherResult = await redisClient.ExecuteLuaScriptAsync(completeOrderLuaScript, keys: new[] { "user:123" });
anotherResult = await redisClient.ExecuteLuaScriptAsync(completeOrderLuaScript, keys: new[] { "user:123" });
anotherResult = await redisClient.ExecuteLuaScriptAsync(completeOrderLuaScript, keys: new[] { "user:123" });
anotherResult = await redisClient.ExecuteLuaScriptAsync(completeOrderLuaScript, keys: new[] { "user:123" });

复制代码

string luaScript = @"
local record_key = KEYS[1]
local max_records = tonumber(ARGV[1])
local new_record = ARGV[2]
local current_count = redis.call('LLEN', record_key)
if current_count >= max_records then
redis.call('LPOP', record_key)
end
redis.call('RPUSH', record_key, new_record)
";
//这里限制保存最新的50条数据，旧的数据则被抛弃
for (int i = 0; i < 60; i++)
{
_ = await redisClient.ExecuteLuaScriptAsync(luaScript, keys: new[] { "msg:list" }, new[] { "50", i.ToString() });
}

复制代码

//lis入队操作
var res = await redisClient.ListPushAsync("list:2", "123", TimeSpan.FromHours(1));
res = await redisClient.ListPushAsync("list:2", "1234", TimeSpan.FromHours(1));
res = await redisClient.ListPushAsync("list:2", "12345", TimeSpan.FromHours(1));
//list出队操作
var str = await redisClient.ListPopAsync("list:2");
//list长度
var length = await redisClient.ListLengthAsync("list:2");
//list range操作
var list = await redisClient.ListRangeAsync("article:list", 0, 10);

复制代码