NewLife
/
Redis
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge pull request #151 from loveMX5201314/dev 

Dev
This commit is contained in:
Stone 2025-03-13 14:14:15 +08:00 committed by GitHub
parent 821e67af50 22ba664c85
commit 24d44f61d5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
58 changed files with 1282 additions and 847 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.github/workflows/publish-beta.yml vendored
View File

@ -16,9 +16,9 @@ jobs:
steps:
- uses: actions/checkout@v4
- name: Setup dotNET
uses: actions/setup-dotnet@v3
uses: actions/setup-dotnet@v4
with:
dotnet-version: 8.x
dotnet-version: 9.x
- name: Build
run: |
dotnet pack --version-suffix $(date "+%Y.%m%d-beta%H%M") -c Release -o out NewLife.Redis/NewLife.Redis.csproj

4
.github/workflows/publish.yml vendored
View File

@ -13,9 +13,9 @@ jobs:
steps:
- uses: actions/checkout@v4
- name: Setup dotNET
uses: actions/setup-dotnet@v3
uses: actions/setup-dotnet@v4
with:
dotnet-version: 8.x
dotnet-version: 9.x
- name: Build
run: |
dotnet pack -c Release -o out NewLife.Redis/NewLife.Redis.csproj

4
.github/workflows/test.yml vendored
View File

@ -15,9 +15,9 @@ jobs:
steps:
- uses: actions/checkout@v4
- name: Setup dotNET
uses: actions/setup-dotnet@v3
uses: actions/setup-dotnet@v4
with:
dotnet-version: 8.x
dotnet-version: 9.x
- name: Build
run: |
dotnet build -c Release

2
NewLife.Extensions.Caching.Redis/NewLife.Extensions.Caching.Redis.csproj
View File

@ -5,7 +5,7 @@
<AssemblyTitle>新生命Redis缓存组件</AssemblyTitle>
<Description>Redis缓存扩展库，便于注入Redis</Description>
<Company>新生命开发团队</Company>
<Copyright>©2002-2023 新生命开发团队</Copyright>
<Copyright>©2002-2025 新生命开发团队</Copyright>
<VersionPrefix>5.5</VersionPrefix>
<VersionSuffix>$([System.DateTime]::Now.ToString(`yyyy.MMdd`))</VersionSuffix>
<Version>$(VersionPrefix).$(VersionSuffix)</Version>

19
NewLife.Redis.Extensions/DependencyInjectionExtensions.cs
View File

@ -3,6 +3,7 @@ using Microsoft.Extensions.Options;
using NewLife;
using NewLife.Caching;
using NewLife.Caching.Services;
using NewLife.Configuration;
using NewLife.Log;
namespace Microsoft.Extensions.DependencyInjection;
@ -22,12 +23,13 @@ public static class DependencyInjectionExtensions
if (redis == null) return services.AddRedisCacheProvider();
services.AddBasic();
services.TryAddSingleton<ICache>(redis);
services.AddSingleton<Redis>(redis);
services.AddSingleton(redis);
// 注册Redis缓存服务
services.TryAddSingleton(p =>
services.TryAddSingleton<ICacheProvider>(p =>
{
var provider = new RedisCacheProvider(p);
if (provider.Cache is not Redis) provider.Cache = redis;
@ -118,6 +120,7 @@ public static class DependencyInjectionExtensions
if (setupAction == null)
throw new ArgumentNullException(nameof(setupAction));
services.AddBasic();
services.AddOptions();
services.Configure(setupAction);
//services.Add(ServiceDescriptor.Singleton<ICache, FullRedis>());
@ -126,7 +129,7 @@ public static class DependencyInjectionExtensions
services.TryAddSingleton<Redis>(p => p.GetRequiredService<FullRedis>());
// 注册Redis缓存服务
services.TryAddSingleton(p =>
services.TryAddSingleton<ICacheProvider>(p =>
{
var redis = p.GetRequiredService<FullRedis>();
var provider = new RedisCacheProvider(p);
@ -154,6 +157,7 @@ public static class DependencyInjectionExtensions
if (setupAction == null)
throw new ArgumentNullException(nameof(setupAction));
services.AddBasic();
services.AddOptions();
services.Configure(setupAction);
services.AddSingleton(sp => new FullRedis(sp, sp.GetRequiredService<IOptions<RedisOptions>>().Value));
@ -166,6 +170,7 @@ public static class DependencyInjectionExtensions
/// <returns></returns>
public static IServiceCollection AddRedisCacheProvider(this IServiceCollection services)
{
services.AddBasic();
services.AddSingleton<ICacheProvider, RedisCacheProvider>();
services.TryAddSingleton<ICache>(p => p.GetRequiredService<ICacheProvider>().Cache);
services.TryAddSingleton<Redis>(p =>
@ -185,4 +190,14 @@ public static class DependencyInjectionExtensions
return services;
}
static void AddBasic(this IServiceCollection services)
{
// 注册依赖项
services.TryAddSingleton<ILog>(XTrace.Log);
services.TryAddSingleton<ITracer>(DefaultTracer.Instance ??= new DefaultTracer());
if (!services.Any(e => e.ServiceType == typeof(IConfigProvider)))
services.TryAddSingleton<IConfigProvider>(JsonConfigProvider.LoadAppSettings());
}
}

57
NewLife.Redis.Extensions/NewLife.Redis.Extensions.csproj
View File

@ -1,12 +1,12 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks>netcoreapp3.1;net5.0;net6.0;net7.0;net8.0;netstandard2.1</TargetFrameworks>
<TargetFrameworks>netcoreapp3.1;net5.0;net6.0;net7.0;net8.0;net9.0;netstandard2.0;netstandard2.1</TargetFrameworks>
<AssemblyTitle>新生命Redis扩展</AssemblyTitle>
<Description>Redis扩展库，便于注入Redis，支持分布式缓存IDistributedCache和数据保护IDataProtection</Description>
<Company>新生命开发团队</Company>
<Copyright>©2002-2024 新生命开发团队</Copyright>
<VersionPrefix>6.0</VersionPrefix>
<Copyright>©2002-2025 新生命开发团队</Copyright>
<VersionPrefix>6.1</VersionPrefix>
<VersionSuffix>$([System.DateTime]::Now.ToString(`yyyy.MMdd`))</VersionSuffix>
<Version>$(VersionPrefix).$(VersionSuffix)</Version>
<FileVersion>$(Version)</FileVersion>
@ -19,6 +19,9 @@
<LangVersion>latest</LangVersion>
<SignAssembly>True</SignAssembly>
<AssemblyOriginatorKeyFile>..\Doc\newlife.snk</AssemblyOriginatorKeyFile>
<NoWarn>1701;1702;NU5104;NU1505;NETSDK1138;CS7035</NoWarn>
<AnalysisLevel>latest</AnalysisLevel>
<WarningsAsErrors>CA2007</WarningsAsErrors>
</PropertyGroup>
<PropertyGroup>
@ -45,34 +48,40 @@
</PackageReference>
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)'=='netcoreapp3.1'">
<PackageReference Include="Microsoft.AspNetCore.DataProtection" Version="8.0.8" />
<PackageReference Include="Microsoft.Extensions.Caching.Abstractions" Version="8.0.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="8.0.1" />
<PackageReference Include="Microsoft.Extensions.Options" Version="8.0.2" />
<PackageReference Include="Microsoft.AspNetCore.DataProtection" Version="3.1.0" />
<PackageReference Include="Microsoft.Extensions.Caching.Abstractions" Version="3.1.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="3.1.0" />
<PackageReference Include="Microsoft.Extensions.Options" Version="3.1.0" />
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)'=='net5.0'">
<PackageReference Include="Microsoft.AspNetCore.DataProtection" Version="8.0.8" />
<PackageReference Include="Microsoft.Extensions.Caching.Abstractions" Version="8.0.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="8.0.1" />
<PackageReference Include="Microsoft.Extensions.Options" Version="8.0.2" />
<PackageReference Include="Microsoft.AspNetCore.DataProtection" Version="5.0.8" />
<PackageReference Include="Microsoft.Extensions.Caching.Abstractions" Version="5.0.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="5.0.0" />
<PackageReference Include="Microsoft.Extensions.Options" Version="5.0.0" />
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)'=='net6.0'">
<PackageReference Include="Microsoft.AspNetCore.DataProtection" Version="8.0.8" />
<PackageReference Include="Microsoft.Extensions.Caching.Abstractions" Version="8.0.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="8.0.1" />
<PackageReference Include="Microsoft.Extensions.Options" Version="8.0.2" />
<ItemGroup Condition="'$(TargetFramework)'=='net6.0' Or '$(TargetFramework)'=='netstandard2.0' Or '$(TargetFramework)'=='netstandard2.1'">
<PackageReference Include="Microsoft.AspNetCore.DataProtection" Version="6.0.0" />
<PackageReference Include="Microsoft.Extensions.Caching.Abstractions" Version="6.0.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="6.0.0" />
<PackageReference Include="Microsoft.Extensions.Options" Version="6.0.0" />
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)'=='net7.0' Or '$(TargetFramework)'=='netstandard2.0' Or '$(TargetFramework)'=='netstandard2.1'">
<PackageReference Include="Microsoft.AspNetCore.DataProtection" Version="8.0.8" />
<PackageReference Include="Microsoft.Extensions.Caching.Abstractions" Version="8.0.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="8.0.1" />
<PackageReference Include="Microsoft.Extensions.Options" Version="8.0.2" />
<ItemGroup Condition="'$(TargetFramework)'=='net7.0'">
<PackageReference Include="Microsoft.AspNetCore.DataProtection" Version="7.0.0" />
<PackageReference Include="Microsoft.Extensions.Caching.Abstractions" Version="7.0.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="7.0.0" />
<PackageReference Include="Microsoft.Extensions.Options" Version="7.0.0" />
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)'=='net8.0'">
<PackageReference Include="Microsoft.AspNetCore.DataProtection" Version="8.0.8" />
<PackageReference Include="Microsoft.AspNetCore.DataProtection" Version="8.0.0" />
<PackageReference Include="Microsoft.Extensions.Caching.Abstractions" Version="8.0.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="8.0.1" />
<PackageReference Include="Microsoft.Extensions.Options" Version="8.0.2" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="8.0.0" />
<PackageReference Include="Microsoft.Extensions.Options" Version="8.0.0" />
</ItemGroup>
<ItemGroup Condition="'$(TargetFramework)'=='net9.0'">
<PackageReference Include="Microsoft.AspNetCore.DataProtection" Version="9.0.0" />
<PackageReference Include="Microsoft.Extensions.Caching.Abstractions" Version="9.0.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="9.0.0" />
<PackageReference Include="Microsoft.Extensions.Options" Version="9.0.0" />
</ItemGroup>
<ItemGroup>

21
NewLife.Redis.sln
View File

@ -20,10 +20,12 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Samples", "Samples", "{E136AE06-927B-4D10-BB83-B607ED0B1FD4}"
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "QueueDemo", "QueueDemo\QueueDemo.csproj", "{5829188A-DC4A-4F00-AD95-5873E4057BDE}"
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "NewLife.Redis.Extensions", "NewLife.Redis.Extensions\NewLife.Redis.Extensions.csproj", "{499A9E8E-050C-40CD-90AF-FE4E499121D1}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "QueueDemo", "Samples\QueueDemo\QueueDemo.csproj", "{AC75EE22-B722-418F-A2A7-66020E29FE4C}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Benchmark", "Samples\Benchmark\Benchmark.csproj", "{FAC752B8-087D-44A9-88E8-3B759A72F936}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
@ -42,20 +44,25 @@ Global
{08A39462-0531-45AB-ACBB-03F62AF4400F}.Debug|Any CPU.Build.0 = Debug|Any CPU
{08A39462-0531-45AB-ACBB-03F62AF4400F}.Release|Any CPU.ActiveCfg = Release|Any CPU
{08A39462-0531-45AB-ACBB-03F62AF4400F}.Release|Any CPU.Build.0 = Release|Any CPU
{5829188A-DC4A-4F00-AD95-5873E4057BDE}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{5829188A-DC4A-4F00-AD95-5873E4057BDE}.Debug|Any CPU.Build.0 = Debug|Any CPU
{5829188A-DC4A-4F00-AD95-5873E4057BDE}.Release|Any CPU.ActiveCfg = Release|Any CPU
{5829188A-DC4A-4F00-AD95-5873E4057BDE}.Release|Any CPU.Build.0 = Release|Any CPU
{499A9E8E-050C-40CD-90AF-FE4E499121D1}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{499A9E8E-050C-40CD-90AF-FE4E499121D1}.Debug|Any CPU.Build.0 = Debug|Any CPU
{499A9E8E-050C-40CD-90AF-FE4E499121D1}.Release|Any CPU.ActiveCfg = Release|Any CPU
{499A9E8E-050C-40CD-90AF-FE4E499121D1}.Release|Any CPU.Build.0 = Release|Any CPU
{AC75EE22-B722-418F-A2A7-66020E29FE4C}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{AC75EE22-B722-418F-A2A7-66020E29FE4C}.Debug|Any CPU.Build.0 = Debug|Any CPU
{AC75EE22-B722-418F-A2A7-66020E29FE4C}.Release|Any CPU.ActiveCfg = Release|Any CPU
{AC75EE22-B722-418F-A2A7-66020E29FE4C}.Release|Any CPU.Build.0 = Release|Any CPU
{FAC752B8-087D-44A9-88E8-3B759A72F936}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{FAC752B8-087D-44A9-88E8-3B759A72F936}.Debug|Any CPU.Build.0 = Debug|Any CPU
{FAC752B8-087D-44A9-88E8-3B759A72F936}.Release|Any CPU.ActiveCfg = Release|Any CPU
{FAC752B8-087D-44A9-88E8-3B759A72F936}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
EndGlobalSection
GlobalSection(NestedProjects) = preSolution
{5829188A-DC4A-4F00-AD95-5873E4057BDE} = {E136AE06-927B-4D10-BB83-B607ED0B1FD4}
{AC75EE22-B722-418F-A2A7-66020E29FE4C} = {E136AE06-927B-4D10-BB83-B607ED0B1FD4}
{FAC752B8-087D-44A9-88E8-3B759A72F936} = {E136AE06-927B-4D10-BB83-B607ED0B1FD4}
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {323831A1-A95B-40AB-B9AD-36A0BC10C2CB}

149
NewLife.Redis/Buffers/BufferedReader.cs Normal file
View File

@ -0,0 +1,149 @@
using NewLife.Data;
namespace NewLife.Caching.Buffers;
/// <summary>缓存的读取器</summary>
public ref struct BufferedReader
{
#region 属性
private ReadOnlySpan<Byte> _span;
/// <summary>数据片段</summary>
public ReadOnlySpan<Byte> Span => _span;
private Int32 _index;
/// <summary>已读取字节数</summary>
public Int32 Position { get => _index; set => _index = value; }
/// <summary>总容量</summary>
public Int32 Capacity => _span.Length;
/// <summary>空闲容量</summary>
public Int32 FreeCapacity => _span.Length - _index;
private readonly Stream _stream;
private readonly Int32 _bufferSize;
private IPacket _data;
#endregion
#region 构造
/// <summary>实例化Span读取器</summary>
/// <param name="stream"></param>
/// <param name="data"></param>
/// <param name="bufferSize"></param>
public BufferedReader(Stream stream, IPacket data, Int32 bufferSize = 8192)
{
_stream = stream;
_bufferSize = bufferSize;
_data = data;
_span = data.GetSpan();
}
#endregion
#region 基础方法
/// <summary>告知有多少数据已从缓冲区读取</summary>
/// <param name="count"></param>
public void Advance(Int32 count)
{
if (count < 0) throw new ArgumentOutOfRangeException(nameof(count));
if (_index + count > _span.Length) throw new ArgumentOutOfRangeException(nameof(count));
_index += count;
}
/// <summary>返回要写入到的Span，其大小按 sizeHint 参数指定至少为所请求的大小</summary>
/// <param name="sizeHint"></param>
/// <returns></returns>
/// <exception cref="ArgumentOutOfRangeException"></exception>
public ReadOnlySpan<Byte> GetSpan(Int32 sizeHint = 0)
{
if (sizeHint > FreeCapacity) throw new ArgumentOutOfRangeException(nameof(sizeHint));
return _span[_index..];
}
#endregion
#region 读取方法
/// <summary>确保缓冲区中有足够的空间。</summary>
/// <param name="size">需要的字节数。</param>
/// <exception cref="InvalidOperationException"></exception>
public void EnsureSpace(Int32 size)
{
// 检查剩余空间大小，不足时，再从数据流中读取。此时需要注意，创建新的OwnerPacket后，需要先把之前剩余的一点数据拷贝过去，然后再读取Stream
var remain = FreeCapacity;
if (remain < size)
{
// 申请指定大小的数据包缓冲区，实际大小可能更大
var idx = 0;
var pk = new OwnerPacket(size <= _bufferSize ? _bufferSize : size);
if (_data != null && remain > 0)
{
if (!_data.TryGetArray(out var arr)) throw new NotSupportedException();
arr.AsSpan(_index, remain).CopyTo(pk.Buffer);
idx += remain;
}
_data.TryDispose();
_data = pk;
_index = 0;
// 多次读取，直到满足需求
//var n = _stream.ReadExactly(pk.Buffer, pk.Offset + idx, pk.Length - idx);
while (idx < size)
{
// 实际缓冲区大小可能大于申请大小，充分利用缓冲区，避免多次读取
var len = pk.Buffer.Length - pk.Offset;
var n = _stream.Read(pk.Buffer, pk.Offset + idx, len - idx);
if (n <= 0) break;
idx += n;
}
if (idx < size)
throw new InvalidOperationException("Not enough data to read.");
pk.Resize(idx);
_span = pk.GetSpan();
}
if (_index + size > _span.Length)
throw new InvalidOperationException("Not enough data to read.");
}
/// <summary>读取单个字节</summary>
/// <returns></returns>
public Byte ReadByte()
{
var size = sizeof(Byte);
EnsureSpace(size);
var result = _span[_index];
_index += size;
return result;
}
/// <summary>读取字节数组</summary>
/// <param name="length"></param>
/// <returns></returns>
/// <exception cref="InvalidOperationException"></exception>
public ReadOnlySpan<Byte> ReadBytes(Int32 length)
{
EnsureSpace(length);
var result = _span.Slice(_index, length);
_index += length;
return result;
}
/// <summary>读取数据包</summary>
/// <param name="length"></param>
/// <returns></returns>
public IPacket ReadPacket(Int32 length)
{
EnsureSpace(length);
var result = _data.Slice(_index, length);
_index += length;
return result;
}
#endregion
}

2
NewLife.Redis/Clusters/RedisCluster.cs
View File

@ -207,7 +207,7 @@ public class RedisCluster : RedisBase, IRedisCluster, IDisposable
}
// 读取指令网络异常时，换一个节点
if (exception is SocketException or IOException)
if (Redis.NoDelay(exception))
{
// 屏蔽旧节点一段时间
if (node is RedisNode redisNode && ++redisNode.Error >= Redis.Retry)

2
NewLife.Redis/Clusters/RedisReplication.cs
View File

@ -335,7 +335,7 @@ public class RedisReplication : RedisBase, IRedisCluster, IDisposable
var now = DateTime.Now;
// 读取指令网络异常时，换一个节点
if (exception is SocketException or IOException)
if (Redis.NoDelay(exception))
{
// 屏蔽旧节点一段时间
if (node is RedisNode redisNode && ++redisNode.Error >= Redis.Retry)

31
NewLife.Redis/Common/IPacket.cs
View File

@ -1,31 +0,0 @@

using System.Text;
namespace NewLife.Caching;
public interface IPacket : IDisposable
{
Int32 Length { get; }
Memory<Byte> Memory { get; }
Span<Byte> GetSpan();
}
public static class SpanHelper
{
public static String ToStr(this ReadOnlySpan<Byte> span, Encoding? encoding = null) => (encoding ?? Encoding.UTF8).GetString(span);
public static String ToStr(this Span<Byte> span, Encoding? encoding = null) => (encoding ?? Encoding.UTF8).GetString(span);
/// <summary>转字符串并释放</summary>
/// <param name="pk"></param>
/// <param name="encoding"></param>
/// <returns></returns>
public static String ToStr(this IPacket pk, Encoding? encoding = null)
{
var rs = pk.GetSpan().ToStr(encoding);
pk.Dispose();
return rs;
}
}

44
NewLife.Redis/Common/MemorySegment.cs
View File

@ -1,44 +0,0 @@
using System.Buffers;
namespace NewLife.Caching;
public struct MemorySegment : IDisposable, IPacket
{
#region 属性
private readonly IMemoryOwner<Byte> _memoryOwner;
private readonly Memory<Byte> _memory;
private readonly Int32 _length;
public Memory<Byte> Memory => _memoryOwner.Memory[.._length];
public Int32 Length => _length;
#endregion
public MemorySegment(IMemoryOwner<Byte> memoryOwner, Int32 length)
{
if (length < 0 || length > memoryOwner.Memory.Length)
{
throw new ArgumentOutOfRangeException(nameof(length), "Length must be non-negative and less than or equal to the memory owner's length.");
}
_memoryOwner = memoryOwner;
_length = length;
}
public MemorySegment(Byte[] buf, Int32 offset = 0, Int32 count = -1)
{
if (count < 0) count = buf.Length - offset;
_memory = new Memory<Byte>(buf, offset, count);
_length = count;
}
public Span<Byte> GetSpan()
{
if (_memoryOwner != null) return _memoryOwner.GetSpan()[.._length];
return _memory.Span;
}
public void Dispose() => _memoryOwner?.Dispose();
}

77
NewLife.Redis/FullRedis.cs
View File

@ -4,11 +4,12 @@ using System.Text;
using NewLife.Caching.Clusters;
using NewLife.Caching.Models;
using NewLife.Caching.Queues;
using NewLife.Caching.Services;
using NewLife.Collections;
using NewLife.Data;
using NewLife.Log;
using NewLife.Messaging;
using NewLife.Model;
using NewLife.Serialization;
namespace NewLife.Caching;
@ -93,6 +94,7 @@ public class FullRedis : Redis
/// <param name="provider">服务提供者，将要解析IConfigProvider</param>
/// <param name="name">缓存名称，也是配置中心key</param>
public FullRedis(IServiceProvider provider, String name) : base(provider, name) { }
/// <summary>按照配置服务实例化Redis，用于NETCore依赖注入</summary>
/// <param name="provider">服务提供者，将要解析IConfigProvider</param>
/// <param name="options">Redis链接配置</param>
@ -318,10 +320,14 @@ public class FullRedis : Redis
InitCluster();
keys = keys.Select(GetKey).ToArray();
//keys = keys.Select(GetKey).ToArray();
for (var i = 0; i < keys.Length; i++)
{
keys[i] = GetKey(keys[i]);
}
// 如果不支持集群，或者只有一个key，直接执行
if (Cluster == null || keys.Length == 1) return [Execute(keys.FirstOrDefault(), (rds, k) => func(rds, keys), write)];
if (Cluster == null || keys.Length == 1) return [Execute(keys[0], (rds, k) => func(rds, keys), write)];
// 计算每个key所在的节点
var dic = new Dictionary<String, List<String>>();
@ -399,11 +405,11 @@ public class FullRedis : Redis
key = GetKey(key);
// 如果不支持集群，直接返回
if (Cluster == null) return await base.ExecuteAsync<T>(key, func, write);
if (Cluster == null) return await base.ExecuteAsync<T>(key, func, write).ConfigureAwait(false);
var node = Cluster.SelectNode(key, write);
//?? throw new XException($"集群[{Name}]没有可用节点");
if (node == null) return await base.ExecuteAsync<T>(key, func, write);
if (node == null) return await base.ExecuteAsync<T>(key, func, write).ConfigureAwait(false);
// 统计性能
var sw = Counter?.StartCount();
@ -419,7 +425,7 @@ public class FullRedis : Redis
try
{
client.Reset();
var rs = await func(client, key);
var rs = await func(client, key).ConfigureAwait(false);
return rs;
}
@ -469,19 +475,16 @@ public class FullRedis : Redis
{
if (keys == null || keys.Length == 0) return 0;
keys = keys.Select(GetKey).ToArray();
if (keys.Length == 1) return base.Remove(keys[0]);
InitCluster();
if (Cluster != null) return Execute(keys, (rds, ks) => rds.Execute<Int32>("DEL", ks), true).Sum();
if (Cluster != null)
for (var i = 0; i < keys.Length; i++)
{
return Execute(keys, (rds, ks) => rds.Execute<Int32>("DEL", ks), true).Sum();
}
else
{
return Execute(keys.FirstOrDefault(), (rds, k) => rds.Execute<Int32>("DEL", keys), true);
keys[i] = GetKey(keys[i]);
}
return Execute(keys[0], (rds, k) => rds.Execute<Int32>("DEL", keys), true);
}
#endregion
@ -494,12 +497,17 @@ public class FullRedis : Redis
{
if (keys == null || !keys.Any()) return new Dictionary<String, T>();
var keys2 = keys.ToArray();
var keys2 = keys as String[] ?? keys.ToArray();
for (var i = 0; i < keys2.Length; i++)
{
keys2[i] = GetKey(keys2[i]);
}
if (keys2.Length == 1) return new Dictionary<String, T> { [keys2[0]] = Get<T>(keys2[0])! };
keys2 = keys2.Select(GetKey).ToArray();
if (keys2.Length == 1 || Cluster == null) return base.GetAll<T>(keys2);
InitCluster();
if (Cluster == null) return base.GetAll<T>(keys2);
//Execute(keys.FirstOrDefault(), (rds, k) => rds.GetAll<T>(keys));
var rs = Execute(keys2, (rds, ks) => rds.GetAll<T>(ks), false);
var dic = new Dictionary<String, T?>();
@ -537,9 +545,14 @@ public class FullRedis : Redis
return;
}
var keys = values.Keys.Select(GetKey).ToArray();
//Execute(values.FirstOrDefault().Key, (rds, k) => rds.SetAll(values), true);
var rs = Execute(keys, (rds, ks) => rds.SetAll(ks.ToDictionary(e => e, e => values[e])), true);
var keys = values.Keys.ToArray();
// 非集群模式
InitCluster();
if (Cluster == null)
Execute(keys[0], (rds, ks) => rds.SetAll(values), true);
else
Execute(keys, (rds, ks) => rds.SetAll(ks.ToDictionary(e => e, e => values[e])), true);
// 使用管道批量设置过期时间
if (expire > 0)
@ -573,6 +586,25 @@ public class FullRedis : Redis
/// <returns></returns>
public override IDictionary<String, T> GetDictionary<T>(String key) => new RedisHash<String, T>(this, key);
/// <summary>
/// 获取哈希表所有数据
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="key"></param>
/// <returns></returns>
public IDictionary<String, T> GetHashAll<T>(String key)
{
var hashMap = new RedisHash<String, T>(this, key);
var nCount = hashMap!.Count();
var sModel = new SearchModel()
{
Pattern = "*",
Position = 0,
Count = nCount
};
return hashMap!.Search(sModel).ToDictionary(kvp => kvp.Key, kvp => kvp.Value);
}
/// <summary>获取队列，快速LIST结构，无需确认</summary>
/// <typeparam name="T"></typeparam>
/// <param name="topic">消息队列主题</param>
@ -614,6 +646,9 @@ public class FullRedis : Redis
/// <param name="key"></param>
/// <returns></returns>
public virtual RedisSortedSet<T> GetSortedSet<T>(String key) => new(this, key);
/// <summary>获取事件总线</summary>
public override IEventBus<T> GetEventBus<T>(String topic, String clientId = "") => new RedisEventBus<T>(this, topic, clientId);
#endregion
#region 字符串操作
@ -961,4 +996,4 @@ public class FullRedis : Redis
/// <returns></returns>
public virtual T[]? SPOP<T>(String key, Int32 count) => Execute(key, (r, k) => r.Execute<T[]>("SPOP", k, count), true);
#endregion
}
}

15
NewLife.Redis/NewLife.Redis.csproj
View File

@ -1,13 +1,13 @@
<Project Sdk="Microsoft.NET.Sdk">
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks>net45;net461;netstandard2.0;netstandard2.1</TargetFrameworks>
<AssemblyName>NewLife.Redis</AssemblyName>
<RootNamespace>NewLife.Caching</RootNamespace>
<AssemblyTitle>新生命Redis客户端组件</AssemblyTitle>
<Description>Redis基础操作、消息队列，经过日均100亿次调用量的项目验证。旧版NET45找2021年</Description>
<Description>Redis基础操作、消息队列，经过日均100亿次调用量的项目验证</Description>
<Company>新生命开发团队</Company>
<Copyright>©2002-2024 新生命开发团队</Copyright>
<VersionPrefix>6.0</VersionPrefix>
<Copyright>©2002-2025 新生命开发团队</Copyright>
<VersionPrefix>6.1</VersionPrefix>
<VersionSuffix>$([System.DateTime]::Now.ToString(`yyyy.MMdd`))</VersionSuffix>
<Version>$(VersionPrefix).$(VersionSuffix)</Version>
<FileVersion>$(Version)</FileVersion>
@ -19,6 +19,9 @@
<LangVersion>latest</LangVersion>
<SignAssembly>True</SignAssembly>
<AssemblyOriginatorKeyFile>..\Doc\newlife.snk</AssemblyOriginatorKeyFile>
<NoWarn>1701;1702;NU5104;NU1505;NETSDK1138;CS7035</NoWarn>
<AnalysisLevel>latest</AnalysisLevel>
<WarningsAsErrors>CA2007</WarningsAsErrors>
</PropertyGroup>
<PropertyGroup>
@ -29,7 +32,7 @@
<RepositoryUrl>https://github.com/NewLifeX/NewLife.Redis</RepositoryUrl>
<RepositoryType>git</RepositoryType>
<PackageTags>新生命团队;X组件;NewLife;$(AssemblyName)</PackageTags>
<PackageReleaseNotes>默认使用System.Text.Json序列化；支持DateOnly/TimeOnly</PackageReleaseNotes>
<PackageReleaseNotes>恢复RedisClient同步方法，减少线程饥渴；内存优化，在高并发场合减少内存分配</PackageReleaseNotes>
<PackageLicenseExpression>MIT</PackageLicenseExpression>
<PublishRepositoryUrl>true</PublishRepositoryUrl>
<EmbedUntrackedSources>true</EmbedUntrackedSources>
@ -53,7 +56,7 @@
</ItemGroup>
<ItemGroup>
<PackageReference Include="NewLife.Core" Version="11.0.2024.923-beta0014" />
<PackageReference Include="NewLife.Core" Version="11.4.2025.301" />
</ItemGroup>
<ItemGroup>

6
NewLife.Redis/PubSub.cs
View File

@ -57,7 +57,7 @@ public class PubSub : RedisBase
client.Reset();
var channels = Key.Split(",", ";").Cast<Object>().ToArray();
await client.ExecuteAsync<String[]>("SUBSCRIBE", channels);
await client.ExecuteAsync<String[]>("SUBSCRIBE", channels, cancellationToken).ConfigureAwait(false);
while (!cancellationToken.IsCancellationRequested)
{
@ -65,11 +65,11 @@ public class PubSub : RedisBase
var source2 = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, source.Token);
//var rs = await client.ExecuteAsync<String[]>(null, new Object[] { new Object() }, source2.Token);
var rs = await client.ReadMoreAsync<String[]>(source2.Token);
var rs = await client.ReadMoreAsync<String[]>(source2.Token).ConfigureAwait(false);
if (rs != null && rs.Length == 3 && rs[0] == "message") onMessage(rs[1], rs[2]);
}
await client.ExecuteAsync<String[]>("SUBSCRIBE", channels);
await client.ExecuteAsync<String[]>("SUBSCRIBE", channels, cancellationToken).ConfigureAwait(false);
Redis.Pool.Return(client);
}

2
NewLife.Redis/Queues/MultipleConsumerGroupsQueue.cs
View File

@ -214,7 +214,7 @@ public class MultipleConsumerGroupsQueue<T> : IDisposable
while (_Cts != null && !_Cts.IsCancellationRequested)
{
var msg = await _Queue.TakeMessageAsync(10);
var msg = await _Queue.TakeMessageAsync(10).ConfigureAwait(false);
if (msg != null && !msg.Id.IsNullOrEmpty())
{
try

14
NewLife.Redis/Queues/QueueExtensions.cs
View File

@ -173,7 +173,7 @@ public static class QueueExtensions
try
{
// 异步阻塞消费
mqMsg = await queue.TakeOneAsync(timeout, cancellationToken);
mqMsg = await queue.TakeOneAsync(timeout, cancellationToken).ConfigureAwait(false);
if (mqMsg != null)
{
// 埋点
@ -197,7 +197,7 @@ public static class QueueExtensions
}
// 处理消息
if (msg != null) await onMessage(msg, mqMsg, cancellationToken);
if (msg != null) await onMessage(msg, mqMsg, cancellationToken).ConfigureAwait(false);
// 确认消息
queue.Acknowledge(mqMsg);
@ -205,7 +205,7 @@ public static class QueueExtensions
else
{
// 没有消息，歇一会
await Task.Delay(1000, cancellationToken);
await Task.Delay(1000, cancellationToken).ConfigureAwait(false);
}
}
catch (ThreadAbortException) { break; }
@ -249,9 +249,9 @@ public static class QueueExtensions
/// <param name="log">日志对象</param>
/// <param name="idField">消息标识字段名，用于处理错误重试</param>
/// <returns></returns>
public static async Task ConsumeAsync<T>(this RedisReliableQueue<String> queue, Action<T> onMessage, CancellationToken cancellationToken = default, ILog? log = null, String? idField = null)
public static Task ConsumeAsync<T>(this RedisReliableQueue<String> queue, Action<T> onMessage, CancellationToken cancellationToken = default, ILog? log = null, String? idField = null)
{
await queue.ConsumeAsync<T>((m, k, t) => { onMessage(m); return Task.FromResult(0); }, cancellationToken, log, idField);
return queue.ConsumeAsync<T>((m, k, t) => { onMessage(m); return Task.FromResult(0); }, cancellationToken, log, idField);
}
/// <summary>队列消费大循环，处理消息后自动确认</summary>
@ -297,7 +297,7 @@ public static class QueueExtensions
try
{
// 异步阻塞消费
mqMsg = await queue.TakeOneAsync(timeout, cancellationToken);
mqMsg = await queue.TakeOneAsync(timeout, cancellationToken).ConfigureAwait(false);
if (mqMsg != null)
{
// 埋点
@ -312,7 +312,7 @@ public static class QueueExtensions
}
else
// 没有消息，歇一会
await Task.Delay(1000, cancellationToken);
await Task.Delay(1000, cancellationToken).ConfigureAwait(false);
}
catch (ThreadAbortException) { break; }
catch (ThreadInterruptedException) { break; }

10
NewLife.Redis/Queues/RedisDelayQueue.cs
View File

@ -143,13 +143,13 @@ public class RedisDelayQueue<T> : QueueBase, IProducerConsumer<T>
while (!cancellationToken.IsCancellationRequested)
{
var score = DateTime.UtcNow.ToInt();
var rs = await _sort.RangeByScoreAsync(0, score, 0, 1, cancellationToken);
var rs = await _sort.RangeByScoreAsync(0, score, 0, 1, cancellationToken).ConfigureAwait(false);
if (rs != null && rs.Length > 0 && TryPop(rs[0])) return rs[0];
// 是否需要等待
if (timeout <= 0) break;
await Task.Delay(1000, cancellationToken);
await Task.Delay(1000, cancellationToken).ConfigureAwait(false);
timeout--;
}
@ -231,7 +231,7 @@ public class RedisDelayQueue<T> : QueueBase, IProducerConsumer<T>
{
// 异步阻塞消费
var score = DateTime.UtcNow.ToInt();
var msgs = await _sort.RangeByScoreAsync(0, score, 0, 10, cancellationToken);
var msgs = await _sort.RangeByScoreAsync(0, score, 0, 10, cancellationToken).ConfigureAwait(false);
if (msgs != null && msgs.Length > 0)
{
// 删除消息后直接进入目标队列，无需进入Ack
@ -248,8 +248,10 @@ public class RedisDelayQueue<T> : QueueBase, IProducerConsumer<T>
if (list.Count > 0) queue.Add(list.ToArray());
}
else
{
// 没有消息，歇一会
await Task.Delay(TransferInterval * 1000, cancellationToken);
await Task.Delay(TransferInterval * 1000, cancellationToken).ConfigureAwait(false);
}
}
catch (ThreadAbortException) { break; }
catch (ThreadInterruptedException) { break; }

8
NewLife.Redis/Queues/RedisQueue.cs
View File

@ -109,7 +109,7 @@ public class RedisQueue<T> : QueueBase, IProducerConsumer<T>
{
if (timeout < 0) return Execute((rc, k) => rc.Execute<T>("RPOP", Key), true);
if (timeout > 0 && Redis.Timeout < timeout * 1000) Redis.Timeout = (timeout + 1) * 1000;
if (timeout > 0 && Redis.Timeout < (timeout + 1) * 1000) Redis.Timeout = (timeout + 1) * 1000;
var rs = Execute((rc, k) => rc.Execute<IPacket[]>("BRPOP", Key, timeout), true);
return rs == null || rs.Length < 2 ? default : (T?)Redis.Encoder.Decode(rs[1], typeof(T));
@ -121,11 +121,11 @@ public class RedisQueue<T> : QueueBase, IProducerConsumer<T>
/// <returns></returns>
public async Task<T?> TakeOneAsync(Int32 timeout = 0, CancellationToken cancellationToken = default)
{
if (timeout < 0) return await ExecuteAsync((rc, k) => rc.ExecuteAsync<T>("RPOP", Key), true);
if (timeout < 0) return await ExecuteAsync((rc, k) => rc.ExecuteAsync<T>("RPOP", Key), true).ConfigureAwait(false);
if (timeout > 0 && Redis.Timeout < timeout * 1000) Redis.Timeout = (timeout + 1) * 1000;
if (timeout > 0 && Redis.Timeout < (timeout + 1) * 1000) Redis.Timeout = (timeout + 1) * 1000;
var rs = await ExecuteAsync((rc, k) => rc.ExecuteAsync<IPacket[]>("BRPOP", new Object[] { Key, timeout }, cancellationToken), true);
var rs = await ExecuteAsync((rc, k) => rc.ExecuteAsync<IPacket[]>("BRPOP", [Key, timeout], cancellationToken), true).ConfigureAwait(false);
return rs == null || rs.Length < 2 ? default : (T?)Redis.Encoder.Decode(rs[1], typeof(T));
}

31
NewLife.Redis/Queues/RedisReliableQueue.cs
View File

@ -146,7 +146,7 @@ public class RedisReliableQueue<T> : QueueBase, IProducerConsumer<T>, IDisposabl
{
RetryAck();
if (timeout > 0 && Redis.Timeout < timeout * 1000) Redis.Timeout = (timeout + 1) * 1000;
if (timeout > 0 && Redis.Timeout < (timeout + 1) * 1000) Redis.Timeout = (timeout + 1) * 1000;
var rs = timeout >= 0 ?
Execute((rc, k) => rc.Execute<T>("BRPOPLPUSH", Key, AckKey, timeout), true) :
@ -165,11 +165,11 @@ public class RedisReliableQueue<T> : QueueBase, IProducerConsumer<T>, IDisposabl
{
RetryAck();
if (timeout > 0 && Redis.Timeout < timeout * 1000) Redis.Timeout = (timeout + 1) * 1000;
if (timeout > 0 && Redis.Timeout < (timeout + 1) * 1000) Redis.Timeout = (timeout + 1) * 1000;
var rs = timeout < 0 ?
await ExecuteAsync((rc, k) => rc.ExecuteAsync<T>("RPOPLPUSH", new Object[] { Key, AckKey }, cancellationToken), true) :
await ExecuteAsync((rc, k) => rc.ExecuteAsync<T>("BRPOPLPUSH", new Object[] { Key, AckKey, timeout }, cancellationToken), true);
await ExecuteAsync((rc, k) => rc.ExecuteAsync<T>("RPOPLPUSH", [Key, AckKey], cancellationToken), true).ConfigureAwait(false) :
await ExecuteAsync((rc, k) => rc.ExecuteAsync<T>("BRPOPLPUSH", [Key, AckKey, timeout], cancellationToken), true).ConfigureAwait(false);
if (rs != null) _Status.Consumes++;
@ -238,11 +238,17 @@ public class RedisReliableQueue<T> : QueueBase, IProducerConsumer<T>, IDisposabl
var rs2 = rds.StopPipeline(true);
foreach (var item in rs2)
{
rs += (Int32)item;
}
}
else
{
foreach (var item in keys)
{
rs += Execute((r, k) => r.Execute<Int32>("LREM", AckKey, 1, item), true);
}
}
return rs;
}
@ -257,16 +263,23 @@ public class RedisReliableQueue<T> : QueueBase, IProducerConsumer<T>, IDisposabl
public RedisDelayQueue<T> InitDelay()
{
if (_delay == null)
{
lock (this)
if (_delay == null)
_delay = new RedisDelayQueue<T>(Redis, $"{Key}:Delay");
{
_delay ??= new RedisDelayQueue<T>(Redis, $"{Key}:Delay");
}
}
if (_delayTask == null || _delayTask.IsCompleted)
{
lock (this)
{
if (_delayTask == null || _delayTask.IsCompleted)
{
_source = new CancellationTokenSource();
_delayTask = Task.Run(() => _delay.TransferAsync(this, null, _source.Token));
}
}
}
return _delay;
}
@ -319,8 +332,8 @@ public class RedisReliableQueue<T> : QueueBase, IProducerConsumer<T>, IDisposabl
// 取出消息键
var msgId = timeout < 0 ?
await ExecuteAsync((rc, k) => rc.ExecuteAsync<String>("RPOPLPUSH", Key, AckKey), true) :
await ExecuteAsync((rc, k) => rc.ExecuteAsync<String>("BRPOPLPUSH", Key, AckKey, timeout), true);
await ExecuteAsync((rc, k) => rc.ExecuteAsync<String>("RPOPLPUSH", Key, AckKey), true).ConfigureAwait(false) :
await ExecuteAsync((rc, k) => rc.ExecuteAsync<String>("BRPOPLPUSH", Key, AckKey, timeout), true).ConfigureAwait(false);
if (msgId.IsNullOrEmpty()) return default;
_Status.Consumes++;
@ -334,7 +347,7 @@ public class RedisReliableQueue<T> : QueueBase, IProducerConsumer<T>, IDisposabl
}
// 处理消息。如果消息已被删除，此时调用func将受到空引用
var rs = await func(messge);
var rs = await func(messge).ConfigureAwait(false);
// 确认并删除消息
Redis.Remove(msgId);

53
NewLife.Redis/Queues/RedisStream.cs
View File

@ -288,7 +288,7 @@ public class RedisStream<T> : QueueBase, IProducerConsumer<T>, IDisposable
/// <returns></returns>
public async Task<T?> TakeOneAsync(Int32 timeout = 0, CancellationToken cancellationToken = default)
{
var msg = await TakeMessageAsync(timeout, cancellationToken);
var msg = await TakeMessageAsync(timeout, cancellationToken).ConfigureAwait(false);
if (msg == null) return default;
return msg.GetBody<T>();
@ -305,7 +305,7 @@ public class RedisStream<T> : QueueBase, IProducerConsumer<T>, IDisposable
/// <returns></returns>
public async Task<Message?> TakeMessageAsync(Int32 timeout = 0, CancellationToken cancellationToken = default)
{
var rs = await TakeMessagesAsync(1, timeout, cancellationToken);
var rs = await TakeMessagesAsync(1, timeout, cancellationToken).ConfigureAwait(false);
return rs?.FirstOrDefault();
}
@ -328,7 +328,7 @@ public class RedisStream<T> : QueueBase, IProducerConsumer<T>, IDisposable
// 抢过来的消息，优先处理，可能需要多次消费才能消耗完
if (_claims > 0)
{
var rs2 = await ReadGroupAsync(group, Consumer, count, 3_000, "0", cancellationToken);
var rs2 = await ReadGroupAsync(group, Consumer, count, 3_000, "0", cancellationToken).ConfigureAwait(false);
if (rs2 != null && rs2.Count > 0)
{
_claims -= rs2.Count;
@ -347,8 +347,8 @@ public class RedisStream<T> : QueueBase, IProducerConsumer<T>, IDisposable
//var id = FromLastOffset ? "$" : ">";
var rs = !group.IsNullOrEmpty() ?
await ReadGroupAsync(group, Consumer, count, t, ">", cancellationToken) :
await ReadAsync(StartId, count, t, cancellationToken);
await ReadGroupAsync(group, Consumer, count, t, ">", cancellationToken).ConfigureAwait(false) :
await ReadAsync(StartId, count, t, cancellationToken).ConfigureAwait(false);
if (rs == null || rs.Count == 0)
{
// id为>时，消费从未传递给消费者的消息
@ -357,7 +357,7 @@ public class RedisStream<T> : QueueBase, IProducerConsumer<T>, IDisposable
// 使用消费组时，如果拿不到消息，则尝试当前消费者之前消费但没有确认的消息
if (!group.IsNullOrEmpty())
{
rs = await ReadGroupAsync(group, Consumer, count, 3_000, "0", cancellationToken);
rs = await ReadGroupAsync(group, Consumer, count, 3_000, "0", cancellationToken).ConfigureAwait(false);
if (rs == null || rs.Count == 0) return null;
XTrace.WriteLine("[{0}]处理历史：{1}", group, rs.Join(",", e => e.Id));
@ -569,7 +569,7 @@ public class RedisStream<T> : QueueBase, IProducerConsumer<T>, IDisposable
var rs = count > 0 ?
Execute((rc, k) => rc.Execute<Object[]>("XRANGE", Key, startId, endId, "COUNT", count), false) :
Execute((rc, k) => rc.Execute<Object[]>("XRANGE", Key, startId, endId), false);
if (rs == null) return new Message[0];
if (rs == null) return [];
return Parse(rs);
}
@ -630,7 +630,7 @@ XREAD count 3 streams stream_key 0-0
if (vs[1] is Object[] vs2) return Parse(vs2);
}
return new Message[0];
return [];
}
/// <summary>异步原始独立消费</summary>
@ -661,13 +661,13 @@ XREAD count 3 streams stream_key 0-0
args.Add(Key);
args.Add(startId);
var rs = await ExecuteAsync((rc, k) => rc.ExecuteAsync<Object[]>("XREAD", args.ToArray(), cancellationToken), true);
var rs = await ExecuteAsync((rc, k) => rc.ExecuteAsync<Object[]>("XREAD", args.ToArray(), cancellationToken), true).ConfigureAwait(false);
if (rs != null && rs.Length == 1 && rs[0] is Object[] vs && vs.Length == 2)
{
if (vs[1] is Object[] vs2) return Parse(vs2);
}
return new Message[0];
return [];
}
private IList<Message> Parse(Object[] vs)
@ -721,7 +721,7 @@ XREAD count 3 streams stream_key 0-0
var rs = count > 0 ?
Execute((rc, k) => rc.Execute<Object[]>("XPENDING", Key, group, startId, endId, count), false) :
Execute((rc, k) => rc.Execute<Object[]>("XPENDING", Key, group, startId, endId), false);
if (rs == null) return new PendingItem[0];
if (rs == null) return [];
var list = new List<PendingItem>();
foreach (var item in rs.Cast<Object[]>())
@ -808,7 +808,7 @@ XREAD count 3 streams stream_key 0-0
if (vs[1] is Object[] vs2) return Parse(vs2);
}
return new Message[0];
return [];
}
/// <summary>异步消费组消费</summary>
@ -829,14 +829,14 @@ XREAD count 3 streams stream_key 0-0
if (id.IsNullOrEmpty()) id = ">";
var rs = count > 0 ?
await ExecuteAsync((rc, k) => rc.ExecuteAsync<Object[]>("XREADGROUP", ["GROUP", group, consumer, "BLOCK", block, "COUNT", count, "STREAMS", Key, id], cancellationToken), true) :
await ExecuteAsync((rc, k) => rc.ExecuteAsync<Object[]>("XREADGROUP", ["GROUP", group, consumer, "BLOCK", block, "STREAMS", Key, id], cancellationToken), true);
await ExecuteAsync((rc, k) => rc.ExecuteAsync<Object[]>("XREADGROUP", ["GROUP", group, consumer, "BLOCK", block, "COUNT", count, "STREAMS", Key, id], cancellationToken), true).ConfigureAwait(false) :
await ExecuteAsync((rc, k) => rc.ExecuteAsync<Object[]>("XREADGROUP", ["GROUP", group, consumer, "BLOCK", block, "STREAMS", Key, id], cancellationToken), true).ConfigureAwait(false);
if (rs != null && rs.Length == 1 && rs[0] is Object[] vs && vs.Length == 2)
{
if (vs[1] is Object[] vs2) return Parse(vs2);
}
return new Message[0];
return [];
}
#endregion
@ -859,7 +859,7 @@ XREAD count 3 streams stream_key 0-0
public GroupInfo[] GetGroups()
{
var rs = Execute((rc, k) => rc.Execute<Object[]>("XINFO", "GROUPS", Key), false);
if (rs == null) return new GroupInfo[0];
if (rs == null) return [];
var gs = new GroupInfo[rs.Length];
for (var i = 0; i < rs.Length; i++)
@ -922,7 +922,7 @@ XREAD count 3 streams stream_key 0-0
try
{
// 异步阻塞消费
var mqMsg = await TakeMessageAsync(timeout, cancellationToken);
var mqMsg = await TakeMessageAsync(timeout, cancellationToken).ConfigureAwait(false);
if (mqMsg != null && !mqMsg.Id.IsNullOrEmpty())
{
// 埋点
@ -943,7 +943,7 @@ XREAD count 3 streams stream_key 0-0
var msg = mqMsg.GetBody<T>();
// 处理消息
if (msg != null) await onMessage(msg, mqMsg, cancellationToken);
if (msg != null) await onMessage(msg, mqMsg, cancellationToken).ConfigureAwait(false);
// 确认消息
Acknowledge(mqMsg.Id);
@ -951,7 +951,7 @@ XREAD count 3 streams stream_key 0-0
else
{
// 没有消息，歇一会
await Task.Delay(1000, cancellationToken);
await Task.Delay(1000, cancellationToken).ConfigureAwait(false);
}
}
catch (ThreadAbortException) { break; }
@ -983,7 +983,7 @@ XREAD count 3 streams stream_key 0-0
/// <param name="onMessage">消息处理。如果处理消息时抛出异常，消息将延迟后回到队列</param>
/// <param name="cancellationToken">取消令牌</param>
/// <returns></returns>
public async Task ConsumeAsync(Action<T> onMessage, CancellationToken cancellationToken = default) => await ConsumeAsync((m, k, t) =>
public Task ConsumeAsync(Action<T> onMessage, CancellationToken cancellationToken = default) => ConsumeAsync((m, k, t) =>
{
onMessage(m);
return Task.FromResult(0);
@ -1023,7 +1023,7 @@ XREAD count 3 streams stream_key 0-0
try
{
// 异步阻塞消费
var mqMsgs = await TakeMessagesAsync(batchSize, timeout, cancellationToken);
var mqMsgs = await TakeMessagesAsync(batchSize, timeout, cancellationToken).ConfigureAwait(false);
if (mqMsgs != null && mqMsgs.Count > 0)
{
// 埋点
@ -1044,7 +1044,7 @@ XREAD count 3 streams stream_key 0-0
var msgs = mqMsgs.Select(e => e.GetBody<T>()!).ToArray();
// 处理消息
await onMessage(msgs, mqMsgs.ToArray(), cancellationToken);
await onMessage(msgs, mqMsgs.ToArray(), cancellationToken).ConfigureAwait(false);
// 确认消息
Acknowledge(mqMsgs.Select(e => e.Id!).ToArray());
@ -1052,7 +1052,7 @@ XREAD count 3 streams stream_key 0-0
else
{
// 没有消息，歇一会
await Task.Delay(1000, cancellationToken);
await Task.Delay(1000, cancellationToken).ConfigureAwait(false);
}
}
catch (ThreadAbortException) { break; }
@ -1086,10 +1086,7 @@ XREAD count 3 streams stream_key 0-0
/// <param name="batchSize">批大小。默认100</param>
/// <param name="cancellationToken">取消令牌</param>
/// <returns></returns>
public async Task ConsumeAsync(Func<T[], Task> onMessage, Int32 batchSize = 100, CancellationToken cancellationToken = default) => await ConsumeAsync(async (m, k, t) =>
{
await onMessage(m);
}, batchSize, cancellationToken);
public Task ConsumeAsync(Func<T[], Task> onMessage, Int32 batchSize = 100, CancellationToken cancellationToken = default) => ConsumeAsync((m, k, t) => onMessage(m), batchSize, cancellationToken);
/// <summary>队列批量消费大循环，批量处理消息后自动确认</summary>
/// <remarks>批量消费最大的问题是部分消费成功，需要用户根据实际情况妥善处理</remarks>
@ -1097,7 +1094,7 @@ XREAD count 3 streams stream_key 0-0
/// <param name="batchSize">批大小。默认100</param>
/// <param name="cancellationToken">取消令牌</param>
/// <returns></returns>
public async Task ConsumeAsync(Action<T[]> onMessage, Int32 batchSize = 100, CancellationToken cancellationToken = default) => await ConsumeAsync((m, k, t) =>
public Task ConsumeAsync(Action<T[]> onMessage, Int32 batchSize = 100, CancellationToken cancellationToken = default) => ConsumeAsync((m, k, t) =>
{
onMessage(m);
return Task.FromResult(0);

56
NewLife.Redis/Redis.cs
View File

@ -149,8 +149,9 @@ public class Redis : Cache, IConfigMapping, ILogFeature
var log = provider.GetService<ILog>();
if (log != null) Log = log;
var configProvider = provider.GetRequiredService<IConfigProvider>();
configProvider.Bind(this, true, name);
var config = provider.GetService<IConfigProvider>();
config ??= JsonConfigProvider.LoadAppSettings();
config.Bind(this, true, name);
}
/// <summary>实例化Redis，指定名称，支持从环境变量Redis_{Name}读取配置，或者逐个属性配置</summary>
@ -473,8 +474,11 @@ public class Redis : Cache, IConfigMapping, ILogFeature
// 网络异常时，自动切换到其它节点
if (ex is AggregateException ae) ex = ae.InnerException;
if (ex is SocketException or IOException && _servers != null && i < _servers.Length)
if (NoDelay(ex))
{
if (_servers == null || i >= _servers.Length) throw;
_idxServer++;
}
else
Thread.Sleep(delay *= 2);
}
@ -487,6 +491,11 @@ public class Redis : Cache, IConfigMapping, ILogFeature
} while (true);
}
/// <summary>网络IO类异常，无需等待，因为遇到此类异常时重发请求也是错</summary>
/// <param name="ex"></param>
/// <returns></returns>
internal Boolean NoDelay(Exception ex) => ex is SocketException or IOException or InvalidOperationException;
/// <summary>直接执行命令，不考虑集群读写</summary>
/// <typeparam name="TResult">返回类型</typeparam>
/// <param name="func">回调函数</param>
@ -517,8 +526,12 @@ public class Redis : Cache, IConfigMapping, ILogFeature
client.TryDispose();
// 网络异常时，自动切换到其它节点
if (ex is SocketException or IOException && _servers != null && i < _servers.Length)
if (ex is AggregateException ae) ex = ae.InnerException;
if (NoDelay(ex))
{
if (_servers == null || i >= _servers.Length) throw;
_idxServer++;
}
else
Thread.Sleep(delay *= 2);
}
@ -547,7 +560,7 @@ public class Redis : Cache, IConfigMapping, ILogFeature
if (rds == null && AutoPipeline > 0) rds = StartPipeline();
if (rds != null)
{
var rs = await func(rds, key);
var rs = await func(rds, key).ConfigureAwait(false);
// 命令数足够，自动提交
if (AutoPipeline > 0 && rds.PipelineCommands >= AutoPipeline)
@ -575,7 +588,7 @@ public class Redis : Cache, IConfigMapping, ILogFeature
try
{
client.Reset();
return await func(client, key);
return await func(client, key).ConfigureAwait(false);
}
catch (RedisException) { throw; }
catch (Exception ex)
@ -586,8 +599,12 @@ public class Redis : Cache, IConfigMapping, ILogFeature
client.TryDispose();
// 网络异常时，自动切换到其它节点
if (ex is SocketException or IOException && _servers != null && i < _servers.Length)
if (ex is AggregateException ae) ex = ae.InnerException;
if (NoDelay(ex))
{
if (_servers == null || i >= _servers.Length) throw;
_idxServer++;
}
else
Thread.Sleep(delay *= 2);
}
@ -631,7 +648,7 @@ public class Redis : Cache, IConfigMapping, ILogFeature
// 管道处理不需要重试
try
{
return rds.StopPipeline(requireResult).Result;
return rds.StopPipeline(requireResult);
}
finally
{
@ -792,7 +809,13 @@ public class Redis : Cache, IConfigMapping, ILogFeature
/// <typeparam name="T"></typeparam>
/// <param name="keys"></param>
/// <returns></returns>
public override IDictionary<String, T> GetAll<T>(IEnumerable<String> keys) => Execute(keys.FirstOrDefault(), (rds, k) => rds.GetAll<T>(keys));
public override IDictionary<String, T> GetAll<T>(IEnumerable<String> keys)
{
var ks = keys as String[] ?? keys.ToArray();
if (ks.Length == 0) return new Dictionary<String, T>();
return Execute(ks[0], (rds, k) => rds.GetAll<T>(ks));
}
/// <summary>批量设置缓存项</summary>
/// <typeparam name="T"></typeparam>
@ -923,7 +946,7 @@ public class Redis : Cache, IConfigMapping, ILogFeature
T? v1 = default;
var rs1 = Execute(key, (rds, k) =>
{
var rs2 = rds.TryExecute("GET", new[] { k }, out T? v2);
var rs2 = rds.TryExecute("GET", [k], out T? v2);
v1 = v2;
return rs2;
});
@ -1065,6 +1088,19 @@ public class Redis : Cache, IConfigMapping, ILogFeature
if (rand && batch > 10) times /= 10;
return base.BenchInc(keys, times, threads, rand, batch);
}
/// <summary>删除测试</summary>
/// <param name="keys"></param>
/// <param name="times"></param>
/// <param name="threads"></param>
/// <param name="rand"></param>
/// <param name="batch"></param>
/// <returns></returns>
protected override Int64 BenchRemove(String[] keys, Int64 times, Int32 threads, Boolean rand, Int32 batch)
{
if (rand && batch > 10) times *= 10;
return base.BenchRemove(keys, times, threads, rand, batch);
}
#endregion
#region 日志

2
NewLife.Redis/RedisBase.cs
View File

@ -40,6 +40,6 @@ public abstract class RedisBase
/// <param name="func"></param>
/// <param name="write">是否写入操作</param>
/// <returns></returns>
public virtual async Task<T> ExecuteAsync<T>(Func<RedisClient, String, Task<T>> func, Boolean write = false) => await Redis.ExecuteAsync(Key, func, write);
public virtual Task<T> ExecuteAsync<T>(Func<RedisClient, String, Task<T>> func, Boolean write = false) => Redis.ExecuteAsync(Key, func, write);
#endregion
}

436
NewLife.Redis/RedisClient.cs
View File

@ -1,11 +1,13 @@
using System.Buffers;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Net.Security;
using System.Net.Sockets;
using System.Security.Authentication;
using System.Security.Cryptography.X509Certificates;
using System.Text;
using NewLife.Buffers;
using NewLife.Caching.Buffers;
using NewLife.Collections;
using NewLife.Data;
using NewLife.Log;
@ -84,6 +86,63 @@ public class RedisClient : DisposeBase
#region 核心方法
private Stream? _stream;
/// <summary>新建连接获取数据流</summary>
/// <param name="create">新建连接</param>
/// <returns></returns>
private Stream? GetStream(Boolean create)
{
var tc = Client;
var ns = _stream;
// 判断连接是否可用
var active = false;
try
{
active = ns != null && tc is { Connected: true } && ns is { CanWrite: true, CanRead: true };
}
catch { }
// 如果连接不可用，则重新建立连接
if (!active)
{
Logined = false;
Client = null;
tc.TryDispose();
if (!create) return null;
var timeout = Timeout;
if (timeout == 0) timeout = Host.Timeout;
tc = new TcpClient
{
SendTimeout = timeout,
ReceiveTimeout = timeout
};
var uri = Server;
var addrs = uri.GetAddresses();
DefaultSpan.Current?.AppendTag($"addrs={addrs.Join()} port={uri.Port}");
tc.Connect(addrs, uri.Port);
Client = tc;
ns = tc.GetStream();
// 客户端SSL
var sp = Host.SslProtocol;
if (sp != SslProtocols.None)
{
var sslStream = new SslStream(ns, false, OnCertificateValidationCallback);
sslStream.AuthenticateAsClient(uri.Host ?? uri.Address + "", [], sp, false);
ns = sslStream;
}
_stream = ns;
}
return ns;
}
/// <summary>新建连接获取数据流</summary>
/// <param name="create">新建连接</param>
/// <returns></returns>
@ -118,7 +177,9 @@ public class RedisClient : DisposeBase
};
var uri = Server;
await tc.ConnectAsync(uri.Address, uri.Port);
var addrs = uri.GetAddresses();
DefaultSpan.Current?.AppendTag($"addrs={addrs.Join()} port={uri.Port}");
await tc.ConnectAsync(addrs, uri.Port).ConfigureAwait(false);
Client = tc;
ns = tc.GetStream();
@ -128,7 +189,7 @@ public class RedisClient : DisposeBase
if (sp != SslProtocols.None)
{
var sslStream = new SslStream(ns, false, OnCertificateValidationCallback);
await sslStream.AuthenticateAsClientAsync(uri.Host ?? uri.Address + "", [], sp, false);
await sslStream.AuthenticateAsClientAsync(uri.Host ?? uri.Address + "", [], sp, false).ConfigureAwait(false);
ns = sslStream;
}
@ -243,12 +304,51 @@ public class RedisClient : DisposeBase
return writer.Position;
}
/// <summary>接收响应</summary>
/// <param name="ns">网络数据流</param>
/// <param name="count">响应个数</param>
/// <returns></returns>
protected virtual IList<Object?> GetResponse(Stream ns, Int32 count)
{
//var ms = new BufferedStream(ns);
var ms = ns;
using var pk = new OwnerPacket(8192);
var n = ms.Read(pk.Buffer, pk.Offset, pk.Length);
if (n <= 0) return [];
pk.Resize(n);
var reader = new BufferedReader(ms, pk, 8192);
return ParseResponse(ms, count, ref reader);
}
/// <summary>异步接收响应</summary>
/// <param name="ns">网络数据流</param>
/// <param name="count">响应个数</param>
/// <param name="cancellationToken">取消通知</param>
/// <returns></returns>
protected virtual async Task<IList<Object?>> GetResponseAsync(Stream ns, Int32 count, CancellationToken cancellationToken = default)
{
var ms = ns;
using var pk = new OwnerPacket(8192);
// 取巧进行异步操作，只要异步读取到第一个字节，后续同步读取
if (cancellationToken == CancellationToken.None)
cancellationToken = new CancellationTokenSource(Timeout > 0 ? Timeout : Host.Timeout).Token;
var n = await ns.ReadAsync(pk.Buffer, pk.Offset, pk.Length, cancellationToken).ConfigureAwait(false);
if (n <= 0) return [];
pk.Resize(n);
var reader = new BufferedReader(ms, pk, 8192);
return ParseResponse(ms, count, ref reader);
}
private IList<Object?> ParseResponse(Stream ms, Int32 count, ref BufferedReader reader)
{
/*
* 响应格式
@ -260,51 +360,30 @@ public class RedisClient : DisposeBase
*/
var list = new List<Object?>();
var ms = ns;
var log = Log == null || Log == Logger.Null ? null : Pool.StringBuilder.Get();
Char header;
var buf = Pool.Shared.Rent(1);
try
{
// 取巧进行异步操作，只要异步读取到第一个字节，后续同步读取
if (cancellationToken == CancellationToken.None)
cancellationToken = new CancellationTokenSource(Timeout > 0 ? Timeout : Host.Timeout).Token;
var n = await ms.ReadAsync(buf, 0, 1, cancellationToken);
if (n <= 0) return list;
header = (Char)buf[0];
}
finally
{
Pool.Shared.Return(buf);
}
//var reader = new SpanReader(pk.GetSpan());
var header = (Char)reader.ReadByte();
// 多行响应
for (var i = 0; i < count; i++)
{
// 解析响应
if (i > 0)
{
var b = ms.ReadByte();
if (b == -1) break;
header = (Char)b;
}
if (i > 0) header = (Char)reader.ReadByte();
log?.Append(header);
if (header == '$')
{
list.Add(ReadBlock(ms, log));
list.Add(ReadBlock(ref reader, log));
}
else if (header == '*')
{
list.Add(ReadBlocks(ms, log));
list.Add(ReadBlocks(ref reader, log));
}
else
{
// 字符串以换行为结束符
var str = ReadLine(ms);
var str = ReadLine(ref reader);
log?.Append(str);
if (header is '+' or ':')
@ -324,23 +403,22 @@ public class RedisClient : DisposeBase
return list;
}
/// <summary>异步执行命令，发请求，取响应</summary>
/// <summary>执行命令，发请求，取响应</summary>
/// <param name="cmd">命令</param>
/// <param name="args">参数数组</param>
/// <param name="cancellationToken">取消通知</param>
/// <returns></returns>
protected virtual async Task<Object?> ExecuteCommandAsync(String cmd, Object?[]? args, CancellationToken cancellationToken)
protected virtual Object? ExecuteCommand(String cmd, Object?[]? args)
{
var isQuit = cmd == "QUIT";
var ns = await GetStreamAsync(!isQuit);
var ns = GetStream(!isQuit);
if (ns == null) return null;
if (!cmd.IsNullOrEmpty())
{
// 验证登录
await CheckLogin(cmd);
await CheckSelect(cmd);
CheckLogin(cmd);
CheckSelect(cmd);
// 估算数据包大小，从内存池借出
var total = GetCommandSize(cmd, args);
@ -353,26 +431,69 @@ public class RedisClient : DisposeBase
var max = Host.MaxMessageSize;
if (max > 0 && memory.Length >= max) throw new InvalidOperationException($"命令[{cmd}]的数据包大小[{memory.Length}]超过最大限制[{max}]，大key会拖累整个Redis实例，可通过Redis.MaxMessageSize调节。");
if (memory.Length > 0) await ns.WriteAsync(memory, cancellationToken);
if (memory.Length > 0) ns.Write(memory);
if (total < MAX_POOL_SIZE) Pool.Shared.Return(buffer);
await ns.FlushAsync(cancellationToken);
ns.Flush();
}
var rs = await GetResponseAsync(ns, 1, cancellationToken);
var rs = GetResponse(ns, 1);
if (isQuit) Logined = false;
return rs.FirstOrDefault();
}
private async Task CheckLogin(String? cmd)
/// <summary>异步执行命令，发请求，取响应</summary>
/// <param name="cmd">命令</param>
/// <param name="args">参数数组</param>
/// <param name="cancellationToken">取消通知</param>
/// <returns></returns>
protected virtual async Task<Object?> ExecuteCommandAsync(String cmd, Object?[]? args, CancellationToken cancellationToken)
{
var isQuit = cmd == "QUIT";
var ns = await GetStreamAsync(!isQuit).ConfigureAwait(false);
if (ns == null) return null;
if (!cmd.IsNullOrEmpty())
{
// 验证登录
CheckLogin(cmd);
CheckSelect(cmd);
// 估算数据包大小，从内存池借出
var total = GetCommandSize(cmd, args);
var buffer = total < MAX_POOL_SIZE ? Pool.Shared.Rent(total) : new Byte[total];
var memory = buffer.AsMemory();
var p = GetRequest(memory, cmd, args);
memory = memory[..p];
var max = Host.MaxMessageSize;
if (max > 0 && memory.Length >= max) throw new InvalidOperationException($"命令[{cmd}]的数据包大小[{memory.Length}]超过最大限制[{max}]，大key会拖累整个Redis实例，可通过Redis.MaxMessageSize调节。");
if (memory.Length > 0) await ns.WriteAsync(memory, cancellationToken).ConfigureAwait(false);
if (total < MAX_POOL_SIZE) Pool.Shared.Return(buffer);
await ns.FlushAsync(cancellationToken).ConfigureAwait(false);
}
var rs = await GetResponseAsync(ns, 1, cancellationToken).ConfigureAwait(false);
if (isQuit) Logined = false;
return rs.FirstOrDefault();
}
private void CheckLogin(String? cmd)
{
if (Logined) return;
if (cmd.EqualIgnoreCase("Auth")) return;
if (!Host.Password.IsNullOrEmpty() && !(await Auth(Host.UserName, Host.Password)))
if (!Host.Password.IsNullOrEmpty() && !Auth(Host.UserName, Host.Password))
throw new Exception("登录失败！");
Logined = true;
@ -380,13 +501,13 @@ public class RedisClient : DisposeBase
}
private Int32 _selected = -1;
private async Task CheckSelect(String? cmd)
private void CheckSelect(String? cmd)
{
var db = Host.Db;
if (_selected == db) return;
if (cmd.EqualIgnoreCase("Auth", "Select", "Info")) return;
if (db > 0 && (Host is not FullRedis rds || !rds.Mode.EqualIgnoreCase("cluster", "sentinel"))) await Select(db);
if (db > 0 && (Host is not FullRedis rds || !rds.Mode.EqualIgnoreCase("cluster", "sentinel"))) Select(db);
_selected = db;
}
@ -394,7 +515,7 @@ public class RedisClient : DisposeBase
/// <summary>重置。干掉历史残留数据</summary>
public void Reset()
{
var ns = GetStreamAsync(false).Result;
var ns = GetStream(false);
if (ns == null) return;
// 干掉历史残留数据
@ -412,95 +533,68 @@ public class RedisClient : DisposeBase
}
}
private static IPacket? ReadBlock(Stream ms, StringBuilder? log) => ReadPacket(ms, log);
private static IPacket? ReadBlock(ref BufferedReader reader, StringBuilder? log) => ReadPacket(ref reader, log);
private Object?[] ReadBlocks(Stream ms, StringBuilder? log)
private Object?[] ReadBlocks(ref BufferedReader reader, StringBuilder? log)
{
// 结果集数量
var len = ReadLength(ms);
var len = ReadLength(ref reader);
log?.Append(len);
if (len < 0) return [];
var arr = new Object?[len];
for (var i = 0; i < len; i++)
{
var b = ms.ReadByte();
if (b == -1) break;
var header = (Char)b;
var header = (Char)reader.ReadByte();
log?.Append(' ');
log?.Append(header);
if (header == '$')
{
arr[i] = ReadPacket(ms, log);
arr[i] = ReadPacket(ref reader, log);
}
else if (header is '+' or ':')
{
arr[i] = ReadLine(ms);
arr[i] = ReadLine(ref reader);
log?.Append(arr[i]);
}
else if (header == '*')
{
arr[i] = ReadBlocks(ms, log);
arr[i] = ReadBlocks(ref reader, log);
}
}
return arr;
}
private static IPacket? ReadPacket(Stream ms, StringBuilder? log)
private static IPacket? ReadPacket(ref BufferedReader reader, StringBuilder? log)
{
var len = ReadLength(ms);
var len = ReadLength(ref reader);
log?.Append(len);
if (len == 0)
{
// 某些字段即使长度是0，还是要把换行符读走
ReadLine(ms);
ReadLine(ref reader);
return null;
}
if (len <= 0) return null;
len += 2;
//// 很多时候，数据长度为1，特殊优化
//if (len == 3)
//{
// var rs = ms.ReadByte();
// // 再读取两个换行符，网络流不支持Seek
// //ms.Seek(2, SeekOrigin.Current);
// ms.ReadByte();
// ms.ReadByte();
// return rs;
//}
// 读取数据包，并跳过换行符
var pk = reader.ReadPacket(len);
if (reader.FreeCapacity >= 2) reader.Advance(2);
// 从内存池借出，包装到MemorySegment中，一路向上传递，用完后Dispose还到池里
var owner = new OwnerPacket(len);
var span = owner.GetSpan();
var p = 0;
while (p < len)
{
// 等待，直到读完需要的数据，避免大包丢数据
var count = ms.Read(span.Slice(p, len - p));
if (count <= 0) break;
p += count;
}
return owner.Slice(0, p - 2);
return pk;
}
private static String ReadLine(Stream ms)
private static String ReadLine(ref BufferedReader reader)
{
var sb = Pool.StringBuilder.Get();
while (true)
var count = reader.FreeCapacity;
for (var i = 0; i < count; i++)
{
var b = ms.ReadByte();
if (b < 0) break;
if (b == '\r')
var b = (Char)reader.ReadByte();
if (b == '\r' && i + 1 < count)
{
var b2 = ms.ReadByte();
if (b2 < 0) break;
var b2 = (Char)reader.ReadByte();
if (b2 == '\n') break;
sb.Append((Char)b);
@ -513,19 +607,17 @@ public class RedisClient : DisposeBase
return sb.Return(true);
}
private static Int32 ReadLength(Stream ms)
private static Int32 ReadLength(ref BufferedReader reader)
{
Span<Char> span = stackalloc Char[32];
var k = 0;
while (true)
var count = reader.FreeCapacity;
for (var i = 0; i < count; i++)
{
var b = ms.ReadByte();
if (b < 0) break;
if (b == '\r')
var b = (Char)reader.ReadByte();
if (b == '\r' && i + 1 < count)
{
var b2 = ms.ReadByte();
if (b2 < 0) break;
var b2 = (Char)reader.ReadByte();
if (b2 == '\n') break;
span[k++] = (Char)b;
@ -560,7 +652,7 @@ public class RedisClient : DisposeBase
else if (item.GetType().GetTypeCode() != TypeCode.Object)
total += 16 + 20;
else
total += 16 + Host.Encoder.Encode(item).Length;
total += 16 + Host.Encoder.Encode(item)?.Length ?? 0;
}
}
@ -581,19 +673,35 @@ public class RedisClient : DisposeBase
/// <returns></returns>
public virtual TResult? Execute<TResult>(String cmd, params Object?[] args)
{
// 管道模式
if (_ps != null)
// 埋点名称，支持二级命令
var act = cmd.EqualIgnoreCase("cluster", "xinfo", "xgroup", "xreadgroup") ? $"{cmd}-{args?.FirstOrDefault()}" : cmd;
using var span = cmd.IsNullOrEmpty() ? null : Host.Tracer?.NewSpan($"redis:{Name}:{act}", args);
try
{
_ps.Add(new Command(cmd, args, typeof(TResult)));
// 管道模式
if (_ps != null)
{
_ps.Add(new Command(cmd, args, typeof(TResult)));
return default;
}
var rs = ExecuteCommand(cmd, args);
if (rs == null) return default;
if (rs is TResult rs2) return rs2;
if (TryChangeType(rs, typeof(TResult), out var target))
{
// 释放内部申请的OwnerPacket
rs.TryDispose();
return (TResult?)target;
}
return default;
}
var rs = ExecuteAsync(cmd, args).Result;
if (rs == null) return default;
if (rs is TResult rs2) return rs2;
if (TryChangeType(rs, typeof(TResult), out var target)) return (TResult?)target;
return default;
catch (Exception ex)
{
span?.SetError(ex, null);
throw;
}
}
/// <summary>尝试执行命令。返回基本类型、对象、对象数组</summary>
@ -603,22 +711,35 @@ public class RedisClient : DisposeBase
/// <returns></returns>
public virtual Boolean TryExecute<TResult>(String cmd, Object?[] args, out TResult? value)
{
var rs = ExecuteAsync(cmd, args).Result;
if (rs is TResult rs2)
// 埋点名称，支持二级命令
var act = cmd.EqualIgnoreCase("cluster", "xinfo", "xgroup", "xreadgroup") ? $"{cmd}-{args?.FirstOrDefault()}" : cmd;
using var span = cmd.IsNullOrEmpty() ? null : Host.Tracer?.NewSpan($"redis:{Name}:{act}", args);
try
{
value = rs2;
return true;
}
var rs = ExecuteCommand(cmd, args);
if (rs is TResult rs2)
{
value = rs2;
return true;
}
value = default;
if (rs == null) return false;
if (TryChangeType(rs, typeof(TResult), out var target))
value = default;
if (rs == null) return false;
if (TryChangeType(rs, typeof(TResult), out var target))
{
// 释放内部申请的OwnerPacket
rs.TryDispose();
value = (TResult?)target;
return true;
}
return false;
}
catch (Exception ex)
{
value = (TResult?)target;
return true;
span?.SetError(ex, null);
throw;
}
return false;
}
/// <summary>异步执行命令。返回字符串、IPacket、IPacket[]</summary>
@ -633,7 +754,7 @@ public class RedisClient : DisposeBase
using var span = cmd.IsNullOrEmpty() ? null : Host.Tracer?.NewSpan($"redis:{Name}:{act}", args);
try
{
return await ExecuteCommandAsync(cmd, args, cancellationToken);
return await ExecuteCommandAsync(cmd, args, cancellationToken).ConfigureAwait(false);
}
catch (Exception ex)
{
@ -646,7 +767,7 @@ public class RedisClient : DisposeBase
/// <param name="cmd">命令</param>
/// <param name="args">参数数组</param>
/// <returns></returns>
public virtual async Task<TResult?> ExecuteAsync<TResult>(String cmd, params Object?[] args) => await ExecuteAsync<TResult>(cmd, args, CancellationToken.None);
public virtual Task<TResult?> ExecuteAsync<TResult>(String cmd, params Object?[] args) => ExecuteAsync<TResult>(cmd, args, CancellationToken.None);
/// <summary>异步执行命令。返回基本类型、对象、对象数组</summary>
/// <param name="cmd">命令</param>
@ -662,10 +783,15 @@ public class RedisClient : DisposeBase
return default;
}
var rs = await ExecuteAsync(cmd, args, cancellationToken);
var rs = await ExecuteAsync(cmd, args, cancellationToken).ConfigureAwait(false);
if (rs == null) return default;
if (rs is TResult rs2) return rs2;
if (TryChangeType(rs, typeof(TResult), out var target)) return (TResult?)target;
if (TryChangeType(rs, typeof(TResult), out var target))
{
// 释放内部申请的OwnerPacket
rs.TryDispose();
return (TResult?)target;
}
return default;
}
@ -675,10 +801,10 @@ public class RedisClient : DisposeBase
/// <returns></returns>
public virtual async Task<TResult?> ReadMoreAsync<TResult>(CancellationToken cancellationToken)
{
var ns = await GetStreamAsync(false);
var ns = await GetStreamAsync(false).ConfigureAwait(false);
if (ns == null) return default;
var rss = await GetResponseAsync(ns, 1, cancellationToken);
var rss = await GetResponseAsync(ns, 1, cancellationToken).ConfigureAwait(false);
var rs = rss.FirstOrDefault();
//var rs = ExecuteCommand(null, null, null);
@ -760,22 +886,22 @@ public class RedisClient : DisposeBase
/// <summary>结束管道模式</summary>
/// <param name="requireResult">要求结果</param>
public virtual async Task<Object?[]?> StopPipeline(Boolean requireResult)
public virtual Object?[]? StopPipeline(Boolean requireResult)
{
var ps = _ps;
if (ps == null) return null;
_ps = null;
var ns = await GetStreamAsync(true);
var ns = GetStream(true);
if (ns == null) return null;
using var span = Host.Tracer?.NewSpan($"redis:{Name}:Pipeline", null);
try
{
// 验证登录
await CheckLogin(null);
await CheckSelect(null);
CheckLogin(null);
CheckSelect(null);
// 估算数据包大小，从内存池借出
var total = 0;
@ -801,17 +927,22 @@ public class RedisClient : DisposeBase
span?.SetTag(cmds);
// 整体发出
if (memory.Length > 0) await ns.WriteAsync(memory);
if (memory.Length > 0) ns.Write(memory);
if (total < MAX_POOL_SIZE) Pool.Shared.Return(buffer);
if (!requireResult) return new Object[ps.Count];
// 获取响应
var list = await GetResponseAsync(ns, ps.Count);
var list = GetResponse(ns, ps.Count);
for (var i = 0; i < list.Count; i++)
{
var rs = list[i];
if (rs != null && TryChangeType(rs, ps[i].Type, out var target) && target != null) list[i] = target;
if (rs != null && TryChangeType(rs, ps[i].Type, out var target) && target != null)
{
// 释放内部申请的OwnerPacket
rs.TryDispose();
list[i] = target;
}
}
return list.ToArray();
@ -823,10 +954,10 @@ public class RedisClient : DisposeBase
}
}
private class Command(String name, Object?[] args, Type type)
private class Command(String name, Object?[]? args, Type type)
{
public String Name { get; } = name;
public Object?[] Args { get; } = args;
public Object?[]? Args { get; } = args;
public Type Type { get; } = type;
}
#endregion
@ -834,22 +965,22 @@ public class RedisClient : DisposeBase
#region 基础功能
/// <summary>心跳</summary>
/// <returns></returns>
public async Task<Boolean> Ping() => await ExecuteAsync<String>("PING") == "PONG";
public Boolean Ping() => Execute<String>("PING") == "PONG";
/// <summary>选择Db</summary>
/// <param name="db"></param>
/// <returns></returns>
public async Task<Boolean> Select(Int32 db) => await ExecuteAsync<String>("SELECT", db + "") == "OK";
public Boolean Select(Int32 db) => Execute<String>("SELECT", db + "") == "OK";
/// <summary>验证密码</summary>
/// <param name="username"></param>
/// <param name="password"></param>
/// <returns></returns>
public async Task<Boolean> Auth(String? username, String password)
public Boolean Auth(String? username, String password)
{
var rs = username.IsNullOrEmpty() ?
await ExecuteAsync<String>("AUTH", password) :
await ExecuteAsync<String>("AUTH", username, password);
Execute<String>("AUTH", password) :
Execute<String>("AUTH", username, password);
return rs == "OK";
}
@ -868,16 +999,17 @@ public class RedisClient : DisposeBase
{
if (values == null || values.Count == 0) throw new ArgumentNullException(nameof(values));
var ps = new List<Object>();
var k = 0;
var ps = new Object[values.Count * 2];
foreach (var item in values)
{
ps.Add(item.Key);
ps[k++] = item.Key;
if (item.Value == null) throw new NullReferenceException();
ps.Add(item.Value);
ps[k++] = item.Value;
}
var rs = Execute<String>("MSET", ps.ToArray());
var rs = Execute<String>("MSET", ps);
if (rs != "OK")
{
using var span = Host.Tracer?.NewSpan($"redis:{Name}:ErrorSetAll", values);
@ -891,20 +1023,18 @@ public class RedisClient : DisposeBase
/// <typeparam name="T"></typeparam>
/// <param name="keys"></param>
/// <returns></returns>
public IDictionary<String, T?> GetAll<T>(IEnumerable<String> keys)
public IDictionary<String, T?> GetAll<T>(String[] keys)
{
if (keys == null || !keys.Any()) throw new ArgumentNullException(nameof(keys));
if (keys == null || keys.Length == 0) throw new ArgumentNullException(nameof(keys));
var ks = keys.ToArray();
var dic = new Dictionary<String, T?>(keys.Length);
if (Execute<Object[]>("MGET", keys) is not Object[] rs) return dic;
var dic = new Dictionary<String, T?>(ks.Length);
if (Execute<Object[]>("MGET", ks) is not Object[] rs) return dic;
for (var i = 0; i < ks.Length && i < rs.Length; i++)
for (var i = 0; i < keys.Length && i < rs.Length; i++)
{
if (rs[i] is IPacket pk)
{
dic[ks[i]] = (T?)Host.Encoder.Decode(pk, typeof(T));
dic[keys[i]] = (T?)Host.Encoder.Decode(pk, typeof(T));
}
}

85
NewLife.Redis/RedisEncoder.cs
View File

@ -5,20 +5,17 @@ using NewLife.Serialization;
namespace NewLife.Caching;
/// <summary>Redis编码器</summary>
public class RedisJsonEncoder : IPacketEncoder
public class RedisJsonEncoder : DefaultPacketEncoder
{
#region 属性
/// <summary>解码出错时抛出异常。默认false不抛出异常，仅返回默认值</summary>
public Boolean ThrowOnError { get; set; }
/// <summary>用于对复杂对象进行Json序列化的主机。优先SystemJson，内部FastJson兜底</summary>
public IJsonHost JsonHost { get; set; } = _host;
private static IJsonHost _host;
#endregion
static RedisJsonEncoder() => _host = GetJsonHost();
/// <summary>实例化Redis编码器</summary>
public RedisJsonEncoder() => JsonHost = _host;
internal static IJsonHost GetJsonHost()
{
// 尝试使用System.Text.Json，不支持时使用FastJson
@ -40,80 +37,14 @@ public class RedisJsonEncoder : IPacketEncoder
return host ?? JsonHelper.Default;
}
/// <summary>数值转数据包</summary>
/// <summary>字符串解码为对象。复杂类型采用Json反序列化</summary>
/// <param name="value"></param>
/// <returns></returns>
public virtual IPacket? Encode(Object value)
{
if (value == null) return null;
if (value is IPacket pk) return pk;
if (value is Byte[] buf) return (ArrayPacket)buf;
if (value is IAccessor acc) return acc.ToPacket();
var type = value.GetType();
var str = type.GetTypeCode() switch
{
TypeCode.Object => JsonHost.Write(value),
TypeCode.String => (value as String)!,
TypeCode.DateTime => ((DateTime)value).ToString("yyyy-MM-dd HH:mm:ss.fff"),
_ => value + "",
};
return (ArrayPacket)str.GetBytes();
}
/// <summary>解码数据包为目标类型</summary>
/// <param name="pk"></param>
/// <param name="type"></param>
/// <returns></returns>
public virtual Object? Decode(IPacket pk, Type type)
protected override Object? OnDecode(String value, Type type)
{
try
{
if (type == typeof(IPacket)) return pk;
if (type == typeof(Byte[])) return pk.ReadBytes();
if (type.As<IAccessor>()) return type.AccessorRead(pk);
if (type == typeof(Boolean) && value == "OK") return true;
// 支持可空类型，遇到无数据时返回null
var ntype = Nullable.GetUnderlyingType(type);
if (pk.Length == 0 && ntype != null && ntype != type) return null;
if (ntype != null) type = ntype;
var str = pk.ToStr();
if (type.GetTypeCode() == TypeCode.String) return str;
if (type.GetTypeCode() != TypeCode.Object)
{
if (type == typeof(Boolean) && str == "OK") return true;
return str.ChangeType(type);
}
// 判断是否Json字符串
if (str[0] == '{' && str[^1] == '}')
return JsonHost.Read(str, type);
return null;
}
catch
{
if (ThrowOnError) throw;
return null;
}
return base.OnDecode(value, type);
}
}
/// <summary>编解码助手</summary>
public static class RedisJsonEncoderHelper
{
//public static T Decode<T>(this RedisJsonEncoder encoder, Packet pk) => (T)encoder.Decode(pk, typeof(T))!;
/// <summary>解码数据包</summary>
/// <typeparam name="T"></typeparam>
/// <param name="encoder"></param>
/// <param name="pk"></param>
/// <returns></returns>
public static T Decode<T>(this IPacketEncoder encoder, IPacket pk) => (T)encoder.Decode(pk, typeof(T))!;
}

4
NewLife.Redis/RedisHash.cs
View File

@ -85,7 +85,7 @@ public class RedisHash<TKey, TValue> : RedisBase, IDictionary<TKey, TValue>
/// <summary>迭代</summary>
public IEnumerator<KeyValuePair<TKey, TValue>> GetEnumerator()
{
foreach (var item in Search("*", 10000))
foreach (var item in Search("*", 1000000))
{
yield return item;
}
@ -226,4 +226,4 @@ public class RedisHash<TKey, TValue> : RedisBase, IDictionary<TKey, TValue>
/// <returns></returns>
public virtual IEnumerable<KeyValuePair<TKey, TValue>> Search(String pattern, Int32 count) => Search(new SearchModel { Pattern = pattern, Count = count });
#endregion
}
}

5
NewLife.Redis/RedisList.cs
View File

@ -1,4 +1,5 @@
using System.Collections;
using System.Diagnostics.CodeAnalysis;
namespace NewLife.Caching;
@ -22,7 +23,7 @@ public class RedisList<T> : RedisBase, IList<T>
/// <returns></returns>
public T this[Int32 index]
{
get => Execute((r, k) => r.Execute<T>("LINDEX", Key, index));
get => Execute((r, k) => r.Execute<T>("LINDEX", Key, index))!;
set => Execute((r, k) => r.Execute<String>("LSET", Key, index, value), true);
}
@ -196,7 +197,7 @@ public class RedisList<T> : RedisBase, IList<T>
/// <param name="start"></param>
/// <param name="stop"></param>
/// <returns></returns>
public T[] LRange(Int32 start, Int32 stop) => Execute((r, k) => r.Execute<T[]>("LRANGE", Key, start, stop));
public T[] LRange(Int32 start, Int32 stop) => Execute((r, k) => r.Execute<T[]>("LRANGE", Key, start, stop)) ?? [];
/// <summary>获取所有元素</summary>
/// <returns></returns>

2
NewLife.Redis/RedisSortedSet.cs
View File

@ -189,7 +189,7 @@ public class RedisSortedSet<T> : RedisBase
/// <param name="count">个数</param>
/// <param name="cancellationToken">取消令牌</param>
/// <returns></returns>
public async Task<T[]> RangeByScoreAsync(Double min, Double max, Int32 offset, Int32 count, CancellationToken cancellationToken = default) => await ExecuteAsync((r,k) => r.ExecuteAsync<T[]>("ZRANGEBYSCORE", new Object[] { Key, min, max, "LIMIT", offset, count }, cancellationToken));
public Task<T[]> RangeByScoreAsync(Double min, Double max, Int32 offset, Int32 count, CancellationToken cancellationToken = default) => ExecuteAsync((r, k) => r.ExecuteAsync<T[]>("ZRANGEBYSCORE", [Key, min, max, "LIMIT", offset, count], cancellationToken));
/// <summary>返回指定分数区间的成员分数对，低分到高分排序</summary>
/// <param name="min">低分，包含</param>

4
NewLife.Redis/RedisStack.cs
View File

@ -92,9 +92,9 @@ public class RedisStack<T> : RedisBase, IProducerConsumer<T>
/// <returns></returns>
public async Task<T?> TakeOneAsync(Int32 timeout = 0, CancellationToken cancellationToken = default)
{
if (timeout < 0) return await ExecuteAsync((rc, k) => rc.ExecuteAsync<T>("RPOP", Key), true);
if (timeout < 0) return await ExecuteAsync((rc, k) => rc.ExecuteAsync<T>("RPOP", Key), true).ConfigureAwait(false);
var rs = await ExecuteAsync((rc, k) => rc.ExecuteAsync<IPacket[]>("BRPOP", new Object[] { Key, timeout }, cancellationToken), true);
var rs = await ExecuteAsync((rc, k) => rc.ExecuteAsync<IPacket[]>("BRPOP", [Key, timeout], cancellationToken), true).ConfigureAwait(false);
return rs == null || rs.Length < 2 ? default : (T?)Redis.Encoder.Decode(rs[1], typeof(T));
}

43
NewLife.Redis/Services/RedisCacheProvider.cs
View File

@ -32,6 +32,7 @@ public class RedisCacheProvider : CacheProvider
public RedisCacheProvider(IServiceProvider serviceProvider)
{
var config = serviceProvider?.GetService<IConfigProvider>();
config ??= JsonConfigProvider.LoadAppSettings();
if (config != null) Init(config, serviceProvider);
}
#endregion
@ -46,40 +47,50 @@ public class RedisCacheProvider : CacheProvider
var queueConn = config["RedisQueue"];
// 实例化全局缓存和队列，如果未设置队列，则使用缓存对象
FullRedis? redis = null;
if (!cacheConn.IsNullOrEmpty())
{
if (serviceProvider != null)
{
_redis = new FullRedis(serviceProvider, "RedisCache")
redis = serviceProvider.GetService<FullRedis>();
if (redis != null && redis.Name != "RedisCache") redis = null;
redis ??= new FullRedis(serviceProvider, "RedisCache")
{
Log = serviceProvider.GetRequiredService<ILog>(),
Tracer = serviceProvider.GetRequiredService<ITracer>(),
Log = serviceProvider.GetService<ILog>()!,
Tracer = serviceProvider.GetService<ITracer>(),
};
}
else
{
_redis = new FullRedis { Name = "RedisCache", Log = XTrace.Log };
_redis.Init(cacheConn);
redis = new FullRedis { Name = "RedisCache", Log = XTrace.Log };
redis.Init(cacheConn);
}
_redisQueue = _redis;
Cache = _redis;
_redis = redis;
_redisQueue = redis;
Cache = redis;
}
if (!queueConn.IsNullOrEmpty())
{
if (serviceProvider != null)
{
_redisQueue = new FullRedis(serviceProvider, "RedisQueue")
redis = serviceProvider.GetService<FullRedis>();
if (redis != null && redis.Name != "RedisQueue") redis = null;
redis ??= new FullRedis(serviceProvider, "RedisQueue")
{
Log = serviceProvider.GetRequiredService<ILog>(),
Tracer = serviceProvider.GetRequiredService<ITracer>(),
Log = serviceProvider.GetService<ILog>()!,
Tracer = serviceProvider.GetService<ITracer>(),
};
}
else
{
_redisQueue = new FullRedis { Name = "RedisQueue", Log = XTrace.Log };
_redisQueue.Init(queueConn);
redis = new FullRedis { Name = "RedisQueue", Log = XTrace.Log };
redis.Init(queueConn);
}
_redisQueue = redis;
}
}
@ -99,15 +110,19 @@ public class RedisCacheProvider : CacheProvider
{
IProducerConsumer<T>? queue = null;
if (group.IsNullOrEmpty())
{
queue = _redisQueue.GetQueue<T>(topic);
XTrace.WriteLine("[{0}]队列消息数：{1}", topic, queue.Count);
}
else
{
var rs = _redisQueue.GetStream<T>(topic);
rs.Group = group;
queue = rs;
}
XTrace.WriteLine("[{0}]队列消息数：{1}", topic, queue.Count);
XTrace.WriteLine("[{0}/{2}]队列消息数：{1}", topic, queue.Count, group);
}
return queue;
}

116
NewLife.Redis/Services/RedisEventBus.cs Normal file
View File

@ -0,0 +1,116 @@
using NewLife.Caching.Queues;
using NewLife.Log;
using NewLife.Messaging;
using System.Diagnostics.CodeAnalysis;
namespace NewLife.Caching.Services;
/// <summary>Redis事件上下文</summary>
public class RedisEventContext<TEvent>(IEventBus<TEvent> eventBus, Queues.Message message) : IEventContext<TEvent>
{
/// <summary>事件总线</summary>
public IEventBus<TEvent> EventBus { get; set; } = eventBus;
/// <summary>原始消息</summary>
public Queues.Message Message { get; set; } = message;
}
/// <summary>Redis事件总线</summary>
/// <typeparam name="TEvent"></typeparam>
/// <remarks>实例化消息队列事件总线</remarks>
public class RedisEventBus<TEvent>(FullRedis cache, String topic, String group) : EventBus<TEvent>
{
private RedisStream<TEvent>? _queue;
private CancellationTokenSource? _source;
/// <summary>销毁</summary>
/// <param name="disposing"></param>
protected override void Dispose(Boolean disposing)
{
base.Dispose(disposing);
_source?.TryDispose();
}
/// <summary>初始化</summary>
[MemberNotNull(nameof(_queue))]
protected virtual void Init()
{
if (_queue != null) return;
// 创建Stream队列，指定消费组，从最后位置开始消费
_queue = cache.GetStream<TEvent>(topic);
_queue.Group = group;
_queue.FromLastOffset = true;
}
/// <summary>发布消息到消息队列</summary>
/// <param name="event">事件</param>
/// <param name="context">上下文</param>
/// <param name="cancellationToken">取消令牌</param>
public override Task<Int32> PublishAsync(TEvent @event, IEventContext<TEvent>? context = null, CancellationToken cancellationToken = default)
{
Init();
var rs = _queue.Add(@event);
return Task.FromResult(1);
}
/// <summary>订阅消息。启动大循环，从消息队列订阅消息，再分发到本地订阅者</summary>
/// <param name="handler">处理器</param>
/// <param name="clientId">客户标识。每个客户只能订阅一次，重复订阅将会挤掉前一次订阅</param>
public override Boolean Subscribe(IEventHandler<TEvent> handler, String clientId = "")
{
if (_source == null)
{
var source = new CancellationTokenSource();
if (Interlocked.CompareExchange(ref _source, source, null) == null)
{
Init();
_ = Task.Run(() => ConsumeMessage(_source));
}
}
// 本进程订阅。从队列中消费到消息时，会发布到本进程的事件总线，这里订阅可以让目标处理器直接收到消息
return base.Subscribe(handler, clientId);
}
/// <summary>从队列中消费消息，经事件总线送给设备会话</summary>
/// <param name="source"></param>
/// <returns></returns>
protected virtual async Task ConsumeMessage(CancellationTokenSource source)
{
DefaultSpan.Current = null;
var cancellationToken = source.Token;
try
{
while (!cancellationToken.IsCancellationRequested)
{
var msg = await _queue!.TakeMessageAsync(15, cancellationToken).ConfigureAwait(false);
if (msg != null)
{
var msg2 = msg.GetBody<TEvent>();
if (msg2 != null)
{
// 发布到事件总线
await base.PublishAsync(msg2, new RedisEventContext<TEvent>(this, msg), cancellationToken).ConfigureAwait(false);
}
}
else
{
await Task.Delay(1_000, cancellationToken).ConfigureAwait(false);
}
}
}
catch (TaskCanceledException) { }
catch (OperationCanceledException) { }
catch (Exception ex)
{
XTrace.WriteException(ex);
}
finally
{
source.Cancel();
}
}
}

252
NewLife.Redis/SpanWriter.cs
View File

@ -1,252 +0,0 @@
using System.Buffers;
using System.Buffers.Binary;
using System.Runtime.InteropServices;
using System.Text;
namespace NewLife.Caching;
/// <summary>Span写入器</summary>
/// <param name="buffer"></param>
public ref struct SpanWriter(Span<Byte> buffer)
{
#region 属性
private readonly Span<Byte> _buffer = buffer;
private Int32 _index;
/// <summary>已写入字节数</summary>
public Int32 WrittenCount => _index;
/// <summary>总容量</summary>
public Int32 Capacity => _buffer.Length;
/// <summary>空闲容量</summary>
public Int32 FreeCapacity => _buffer.Length - _index;
/// <summary>是否小端字节序。默认true</summary>
public Boolean IsLittleEndian { get; set; } = true;
#endregion
#region 基础方法
/// <summary>告知有多少数据已写入缓冲区</summary>
/// <param name="count"></param>
public void Advance(Int32 count)
{
if (count < 0) throw new ArgumentOutOfRangeException(nameof(count));
if (_index > _buffer.Length - count) throw new ArgumentOutOfRangeException(nameof(count));
_index += count;
}
//public Memory<Byte> GetMemory(Int32 sizeHint = 0)
//{
// if (sizeHint > FreeCapacity) throw new ArgumentOutOfRangeException(nameof(sizeHint));
// return _buffer[.._index];
//}
/// <summary>返回要写入到的Span，其大小按 sizeHint 参数指定至少为所请求的大小</summary>
/// <param name="sizeHint"></param>
/// <returns></returns>
/// <exception cref="ArgumentOutOfRangeException"></exception>
public Span<Byte> GetSpan(Int32 sizeHint = 0)
{
if (sizeHint > FreeCapacity) throw new ArgumentOutOfRangeException(nameof(sizeHint));
return _buffer[.._index];
}
//public Span<Byte> GetRemain() => _buffer[_index..];
#endregion
#region 写入方法
/// <summary>确保缓冲区中有足够的空间。</summary>
/// <param name="size">需要的字节数。</param>
private void EnsureSpace(Int32 size)
{
if (_index + size > _buffer.Length)
{
throw new InvalidOperationException("缓冲区空间不足。");
}
}
/// <summary>写入字节。</summary>
/// <param name="value">要写入的字节值。</param>
public Int32 Write(Byte value)
{
var size = sizeof(Byte);
EnsureSpace(size);
_buffer[_index] = value;
_index += size;
return size;
}
/// <summary>写入 32 位整数。</summary>
/// <param name="value">要写入的整数值。</param>
public Int32 Write(Int32 value)
{
var size = sizeof(Int32);
EnsureSpace(size);
if (IsLittleEndian)
BinaryPrimitives.WriteInt32LittleEndian(_buffer.Slice(_index), value);
else
BinaryPrimitives.WriteInt32BigEndian(_buffer.Slice(_index), value);
_index += size;
return size;
}
/// <summary>写入无符号 32 位整数。</summary>
/// <param name="value">要写入的无符号整数值。</param>
public Int32 Write(UInt32 value)
{
var size = sizeof(UInt32);
EnsureSpace(size);
if (IsLittleEndian)
BinaryPrimitives.WriteUInt32LittleEndian(_buffer.Slice(_index), value);
else
BinaryPrimitives.WriteUInt32BigEndian(_buffer.Slice(_index), value);
_index += size;
return size;
}
/// <summary>写入 64 位整数。</summary>
/// <param name="value">要写入的整数值。</param>
public Int32 Write(Int64 value)
{
var size = sizeof(Int64);
EnsureSpace(size);
if (IsLittleEndian)
BinaryPrimitives.WriteInt64LittleEndian(_buffer.Slice(_index), value);
else
BinaryPrimitives.WriteInt64BigEndian(_buffer.Slice(_index), value);
_index += size;
return size;
}
/// <summary>写入无符号 64 位整数。</summary>
/// <param name="value">要写入的无符号整数值。</param>
public Int32 Write(UInt64 value)
{
var size = sizeof(UInt64);
EnsureSpace(size);
if (IsLittleEndian)
BinaryPrimitives.WriteUInt64LittleEndian(_buffer.Slice(_index), value);
else
BinaryPrimitives.WriteUInt64BigEndian(_buffer.Slice(_index), value);
_index += size;
return size;
}
/// <summary>写入单精度浮点数。</summary>
/// <param name="value">要写入的浮点值。</param>
public unsafe Int32 Write(Single value)
{
#if NETSTANDARD2_1_OR_GREATER
return Write(BitConverter.SingleToInt32Bits(value));
#else
return Write(*(Int32*)(&value));
#endif
}
/// <summary>写入双精度浮点数。</summary>
/// <param name="value">要写入的浮点值。</param>
public unsafe Int32 Write(Double value)
{
#if NETSTANDARD2_1_OR_GREATER
return Write(BitConverter.DoubleToInt64Bits(value));
#else
return Write(*(Int64*)(&value));
#endif
}
/// <summary>写入字符串</summary>
/// <param name="value"></param>
/// <returns></returns>
/// <exception cref="ArgumentNullException"></exception>
public Int32 Write(String value)
{
if (value == null) throw new ArgumentNullException(nameof(value));
var count = Encoding.UTF8.GetBytes(value.AsSpan(), _buffer.Slice(_index));
_index += count;
return count;
}
/// <summary>写入字节数组</summary>
/// <param name="value"></param>
/// <returns></returns>
/// <exception cref="ArgumentNullException"></exception>
public Int32 Write(Byte[] value)
{
if (value == null)
throw new ArgumentNullException(nameof(value));
value.CopyTo(_buffer.Slice(_index));
_index += value.Length;
return value.Length;
}
/// <summary>写入Span</summary>
/// <param name="span"></param>
/// <returns></returns>
public Int32 Write(ReadOnlySpan<Byte> span)
{
span.CopyTo(_buffer.Slice(_index));
_index += span.Length;
return span.Length;
}
#endregion
}
static class SpanHelper
{
public static unsafe Int32 GetBytes(this Encoding encoding, ReadOnlySpan<Char> chars, Span<Byte> bytes)
{
fixed (Char* chars2 = &MemoryMarshal.GetReference(chars))
{
fixed (Byte* bytes2 = &MemoryMarshal.GetReference(bytes))
{
return encoding.GetBytes(chars2, chars.Length, bytes2, bytes.Length);
}
}
}
public static unsafe String GetString(this Encoding encoding, ReadOnlySpan<Byte> bytes)
{
if (bytes.IsEmpty) return String.Empty;
#if NET45
return encoding.GetString(bytes.ToArray());
#else
fixed (Byte* bytes2 = &MemoryMarshal.GetReference(bytes))
{
return encoding.GetString(bytes2, bytes.Length);
}
#endif
}
public static Task WriteAsync(this Stream stream, ReadOnlyMemory<Byte> buffer, CancellationToken cancellationToken = default)
{
if (MemoryMarshal.TryGetArray(buffer, out var segment))
return stream.WriteAsync(segment.Array, segment.Offset, segment.Count, cancellationToken);
var array = ArrayPool<Byte>.Shared.Rent(buffer.Length);
buffer.Span.CopyTo(array);
var writeTask = stream.WriteAsync(array, 0, buffer.Length, cancellationToken);
return Task.Run(async () =>
{
try
{
await writeTask.ConfigureAwait(false);
}
finally
{
ArrayPool<Byte>.Shared.Return(array);
}
});
}
}

29
Readme.MD
View File

@ -1,4 +1,4 @@
# NewLife.Redis - Redis客户端组件
# NewLife.Redis - Redis客户端组件
![GitHub top language](https://img.shields.io/github/languages/top/newlifex/newlife.redis?logo=github)
![GitHub License](https://img.shields.io/github/license/newlifex/newlife.redis?logo=github)
@ -9,7 +9,7 @@
## [[English]](https://github.com/NewLifeX/NewLife.Redis/blob/master/Readme.en.md)
`NewLife.Redis` 是一个Redis客户端组件，以高性能处理大数据实时计算为目标。
Redis协议基础实现Redis/RedisClient位于[X组件](https://github.com/NewLifeX/X)，本库为扩展实现，主要增加列表结构、哈希结构、队列等高级功能。
Redis协议基础实现Redis/RedisClient位于Redis类，FullRedis为扩展实现，主要增加列表结构、哈希结构、队列等高级功能。
源码： https://github.com/NewLifeX/NewLife.Redis
Nuget：NewLife.Redis
@ -198,18 +198,18 @@ Redis实现ICache接口，它的孪生兄弟MemoryCache，内存缓存，千万
数据增大（10万）以后，改用Redis实现，不需要修改业务代码。
## 新生命项目矩阵
各项目默认支持net7.0/netstandard2.1/netstandard2.0/net4.61，旧版（2022.1225）支持net4.5/net4.0/net2.0
各项目默认支持net9.0/netstandard2.1/netstandard2.0/net4.62/net4.5，旧版（2024.0801）支持net4.0/net2.0
| 项目 | 年份 | 说明 |
| :--------------------------------------------------------------: | :---: | -------------------------------------------------------------------------------------- |
| 基础组件 | | 支撑其它中间件以及产品项目 |
| [NewLife.Core](https://github.com/NewLifeX/X) | 2002 | 核心库，日志、配置、缓存、网络、序列化、APM性能追踪 |
| [NewLife.XCode](https://github.com/NewLifeX/NewLife.XCode) | 2005 | 大数据中间件，单表百亿级，MySql/SQLite/SqlServer/Oracle/TDengine/达梦，自动分表 |
| [NewLife.XCode](https://github.com/NewLifeX/NewLife.XCode) | 2005 | 大数据中间件，单表百亿级，MySql/SQLite/SqlServer/Oracle/PostgreSql/达梦，自动分表 |
| [NewLife.Net](https://github.com/NewLifeX/NewLife.Net) | 2005 | 网络库，单机千万级吞吐率（2266万tps），单机百万级连接（400万Tcp） |
| [NewLife.Remoting](https://github.com/NewLifeX/NewLife.Remoting) | 2011 | RPC通信框架，内网高吞吐或物联网硬件设备场景 |
| [NewLife.Remoting](https://github.com/NewLifeX/NewLife.Remoting) | 2011 | RPC通信框架，内网高吞吐，物联网设备低开销易接入 |
| [NewLife.Cube](https://github.com/NewLifeX/NewLife.Cube) | 2010 | 魔方快速开发平台，集成了用户权限、SSO登录、OAuth服务端等，单表100亿级项目验证 |
| [NewLife.Agent](https://github.com/NewLifeX/NewLife.Agent) | 2008 | 服务管理组件，把应用安装成为操作系统守护进程，Windows服务、Linux的Systemd |
| [NewLife.Zero](https://github.com/NewLifeX/NewLife.Zero) | 2020 | Zero零代脚手架，基于NewLife组件生态的项目模板，Web、WebApi、Service |
| [NewLife.Zero](https://github.com/NewLifeX/NewLife.Zero) | 2020 | Zero零代脚手架，基于NewLife组件生态的项目模板NewLife.Templates，Web、WebApi、Service |
| 中间件 | | 对接知名中间件平台 |
| [NewLife.Redis](https://github.com/NewLifeX/NewLife.Redis) | 2017 | Redis客户端，微秒级延迟，百万级吞吐，丰富的消息队列，百亿级数据量项目验证 |
| [NewLife.RocketMQ](https://github.com/NewLifeX/NewLife.RocketMQ) | 2018 | RocketMQ纯托管客户端，支持Apache RocketMQ和阿里云消息队列，十亿级项目验 |
@ -217,29 +217,30 @@ Redis实现ICache接口，它的孪生兄弟MemoryCache，内存缓存，千万
| [NewLife.IoT](https://github.com/NewLifeX/NewLife.IoT) | 2022 | IoT标准库，定义物联网领域的各种通信协议标准规范 |
| [NewLife.Modbus](https://github.com/NewLifeX/NewLife.Modbus) | 2022 | ModbusTcp/ModbusRTU/ModbusASCII，基于IoT标准库实现，支持IoT平台和IoTEdge |
| [NewLife.Siemens](https://github.com/NewLifeX/NewLife.Siemens) | 2022 | 西门子PLC协议，基于IoT标准库实现，支持IoT平台和IoTEdge |
| [NewLife.Map](https://github.com/NewLifeX/NewLife.Map) | 2022 | 地图组件库，封装百度地图、高德地图和腾讯地图 |
| [NewLife.IP](https://github.com/NewLifeX/NewLife.IP) | 2022 | IP地址库，IP地址转物理地址 |
| [NewLife.Map](https://github.com/NewLifeX/NewLife.Map) | 2022 | 地图组件库，封装百度地图、高德地图、腾讯地图、天地图 |
| [NewLife.Audio](https://github.com/NewLifeX/NewLife.Audio) | 2023 | 音频编解码库，PCM/ADPCMA/G711A/G722U/WAV/AAC |
| 产品平台 | | 产品平台级，编译部署即用，个性化自定义 |
| [AntJob](https://github.com/NewLifeX/AntJob) | 2019 | 蚂蚁调度，分布式大数据计算平台（实时/离线），蚂蚁搬家分片思想，万亿级数据量项目验证 |
| [Stardust](https://github.com/NewLifeX/Stardust) | 2018 | 星尘，分布式服务平台，节点管理、APM监控中心、配置中心、注册中心、发布中心 |
| [AntJob](https://github.com/NewLifeX/AntJob) | 2019 | 蚂蚁调度，分布式大数据计算平台（实时/离线），蚂蚁搬家分片思想，万亿级数据量项目验证 |
| [NewLife.ERP](https://github.com/NewLifeX/NewLife.ERP) | 2021 | 企业ERP，产品管理、客户管理、销售管理、供应商管理 |
| [CrazyCoder](https://github.com/NewLifeX/XCoder) | 2006 | 码神工具，众多开发者工具，网络、串口、加解密、正则表达式、Modbus |
| [EasyIO](https://github.com/NewLifeX/EasyIO) | 2023 | 简易文件存储，支持分布式系统中文件集中存储。 |
| [XProxy](https://github.com/NewLifeX/XProxy) | 2005 | 产品级反向代理，NAT代理、Http代理 |
| [HttpMeter](https://github.com/NewLifeX/HttpMeter) | 2022 | Http压力测试工具 |
| [GitCandy](https://github.com/NewLifeX/GitCandy) | 2015 | Git源代码管理系统 |
| [SmartOS](https://github.com/NewLifeX/SmartOS) | 2014 | 嵌入式操作系统，完全独立自主，支持ARM Cortex-M芯片架构 |
| [SmartA2](https://github.com/NewLifeX/SmartA2) | 2019 | 嵌入式工业计算机，物联网边缘网关，高性能.NET6主机，应用于工业、农业、交通、医疗 |
| 菲凡物联FIoT | 2020 | 物联网整体解决方案，建筑、环保、农业，软硬件及大数据分析一体化，单机十万级点位项目验证 |
| NewLife.UWB | 2020 | 厘米级（10~20cm）高精度室内定位，软硬件一体化，与其它系统联动，大型展厅项目验证 |
| FIoT物联网平台 | 2020 | 物联网整体解决方案，建筑、环保、农业，软硬件及大数据分析一体化，单机十万级点位项目验证 |
| UWB高精度室内定位 | 2020 | 厘米级（10~20cm）高精度室内定位，软硬件一体化，与其它系统联动，大型展厅项目验证 |
## 新生命开发团队
![XCode](https://newlifex.com/logo.png)
新生命团队（NewLife）成立于2002年，是新时代物联网行业解决方案提供者，致力于提供软硬件应用方案咨询、系统架构规划与开发服务。
团队主导的开源NewLife系列组件已被广泛应用于各行业，Nuget累计下载量高达60余万次。
团队开发的大数据核心组件NewLife.XCode、蚂蚁调度计算平台AntJob、星尘分布式平台Stardust、缓存队列组件NewLife.Redis以及物联网平台NewLife.IoT，均成功应用于电力、高校、互联网、电信、交通、物流、工控、医疗、文博等行业，为客户提供了大量先进、可靠、安全、高质量、易扩展的产品和系统集成服务。
团队主导的80多个开源项目已被广泛应用于各行业，Nuget累计下载量高达300余万次。
团队开发的大数据中间件NewLife.XCode、蚂蚁调度计算平台AntJob、星尘分布式平台Stardust、缓存队列组件NewLife.Redis以及物联网平台FIoT，均成功应用于电力、高校、互联网、电信、交通、物流、工控、医疗、文博等行业，为客户提供了大量先进、可靠、安全、高质量、易扩展的产品和系统集成服务。
我们将不断通过服务的持续改进，成为客户长期信赖的合作伙伴，通过不断的创新和发展，成为国内优秀的IT服务供应商。
我们将不断通过服务的持续改进，成为客户长期信赖的合作伙伴，通过不断的创新和发展，成为国内优秀的IoT服务供应商。
`新生命团队始于2002年，部分开源项目具有20年以上漫长历史，源码库保留有2010年以来所有修改记录`
网站：https://newlifex.com

202
Readme.en.md
View File

@ -1,70 +1,79 @@
# NewLife. Redis - redis client component
# NewLife.Redis - Redis Client Component
![GitHub top language](https://img.shields.io/github/languages/top/newlifex/newlife.redis?logo=github) ![GitHub License](https://img.shields.io/github/license/newlifex/newlife.redis?logo=github) ![Nuget Downloads](https://img.shields.io/nuget/dt/newlife.redis?logo=nuget) ![Nuget](https://img.shields.io/nuget/v/newlife.redis?logo=nuget) ![Nuget (with prereleases)](https://img.shields.io/nuget/vpre/newlife.redis?label=dev%20nuget&amp;logo=nuget)
![GitHub top language](https://img.shields.io/github/languages/top/newlifex/newlife.redis?logo=github)
![GitHub License](https://img.shields.io/github/license/newlifex/newlife.redis?logo=github)
![Nuget Downloads](https://img.shields.io/nuget/dt/newlife.redis?logo=nuget)
![Nuget](https://img.shields.io/nuget/v/newlife.redis?logo=nuget)
![Nuget (with prereleases)](https://img.shields.io/nuget/vpre/newlife.redis?label=dev%20nuget&logo=nuget)
`NewLife.Redis` is a Redis client component targeting high-performance processing of big data real-time computations.
Redis protocol base implementation Redis/RedisClient is located in [X components](https://github.com/NewLifeX/X), this library for the extension of the implementation, mainly to increase the list structure, hash structure, queues and other advanced features.
Source : https://github.com/NewLifeX/NewLife.Redis
Nuget: NewLife.Redis
`NewLife.Redis` is a Redis client component designed for high-performance real-time big data processing.
The Redis protocol is implemented in the Redis/RedisClient located in the [X component](https://github.com/NewLifeX/X), and this library extends it with advanced features such as list structures, hash structures, and queues.
Source: https://github.com/NewLifeX/NewLife.Redis
Nuget: NewLife.Redis
Tutorial: [https://newlifex.com/core/redis](https://newlifex.com/core/redis)
---
### Characteristics
* Widely used in ZTO big data real-time computing, more than 200 Redis instances have been working stably for more than a year, processing nearly 100 million parcels of data per day, with an average daily call volume of 8 billion times
* Low latency, with Get/Set operations taking an average of 200 to 600 us (including round-trip network communications)
* High throughput, with its own connection pool, supporting up to 1,000 concurrencies
* High-performance, with support for binary serialization
### Features
* Widely used for real-time big data processing at ZTO since 2017. More than 200 Redis instances have been running stably for over a year, processing nearly 100 million package data entries daily, with 8 billion calls per day.
* Low latency, with Get/Set operations averaging 200-600μs (including round-trip network communication).
* High throughput with an in-built connection pool, supporting up to 100,000 concurrent connections.
* High performance, supports binary serialization.
---
### Redis experience sharing
* Multi-instance deployment on Linux, where the number of instances is equal to the number of processors, and the maximum memory of each instance is directly localized to the physical memory of the machine, avoiding the bursting of the memory of a single instance
* Store massive data (1 billion+) on multiple instances based on key hash (Crc16/Crc32) for exponential growth in read/write performance
* Use binary serialization instead of the usual Json serialization
* Rationalize the design of the Value size of each pair of Keys, including but not limited to the use of bulk acquisition, with the principle of keeping each network packet in the vicinity of 1.4k bytes to reduce the number of communications
* Redis client Get/Set operations took an average of 200-600 us (including round-trip network communication), which was used as a reference to evaluate the network environment and Redis client components
* Consolidation of a batch of commands using the pipeline Pipeline
* The main performance bottlenecks in Redis are serialization, network bandwidth, and memory size, with processors also bottlenecking during abuse
* Other searchable optimization techniques
The above experience, derived from more than 300 instances of more than 4T space more than a year of stable work experience, and in accordance with the degree of importance of the ranking of the order, according to the needs of the scene can be used as appropriate!
### Redis Experience Sharing
* Deploy multiple instances on Linux, with the number of instances equal to the number of processors. Each instance's maximum memory equals the physical memory of the machine to avoid memory overflow in a single instance.
* Store massive data (over 1 billion entries) across multiple instances by hashing keys (Crc16/Crc32), greatly improving read and write performance.
* Use binary serialization instead of common JSON serialization.
* Design the size of each key-value pair carefully, including but not limited to batch retrievals. The goal is to keep each network packet around 1.4k bytes to reduce communication overhead.
* Redis Get/Set operations average 200-600μs (including round-trip network communication). Use this as a benchmark to assess the network environment and Redis client components.
* Use pipelining to combine a batch of commands.
* Redis’s main performance bottlenecks are serialization, network bandwidth, and memory size. Excessive use can also cause processor bottlenecks.
* Other optimizations can be explored.
The above experiences come from over a year of stable operation with over 300 instances and more than 4TB of space, ordered by importance and should be applied as needed based on the scenario.
---
### Recommended usage
It is recommended to use the singleton pattern, Redis has an internal connection pool and supports multi-threaded concurrent access.
``` csharp
### Recommended Usage
It is recommended to use the Singleton pattern. Redis internally uses a connection pool and supports multi-threaded concurrent access.
```csharp
public static class RedisHelper
{
/// <summary>
/// Redis Instance
/// Redis instance
/// </summary>
public static FullRedis redisConnection=FullRedis.Create("server=127.0.0.1:6379;password=123456;db=4");
public static FullRedis redisConnection { get; set; } = new FullRedis("127.0.0.1:6379", "123456", 4);
}
Console.WriteLine(RedisHelper.redisConnection.Keys);
```
```
---
### Basic Redis
Redis implements the standard protocols as well as basic string manipulation , the complete implementation of the independent open source project [NewLife.Redis](https://github.com/NewLifeX/NewLife.Redis) provides .
Adopting connection pooling and synchronous blocking architecture, it features ultra-low latency (200~600us) and ultra-high throughput.
It should be widely used in the real-time calculation of big data in the logistics industry, and has been verified by 10 billion calls per day.
```csharp
// Instantiate Redis, the default port 6379 can be omitted, and the password can be written in two ways
### Basic Redis
The Redis implementation follows the standard protocol and basic string operations, with the complete implementation provided by the independent open-source project [NewLife.Redis](https://github.com/NewLifeX/NewLife.Redis).
It uses a connection pool and synchronous blocking architecture, offering ultra-low latency (200-600μs) and extremely high throughput.
It is widely used in real-time big data processing in the logistics industry and has been validated with daily call volumes of 10 billion.
```csharp
// Instantiate Redis, default port 6379 can be omitted, two ways to write the password
//var rds = new FullRedis("127.0.0.1", null, 7);
var rds = new FullRedis("127.0.0.1:6379", "pass", 7);
//var rds = new FullRedis();
//rds.Init("server=127.0.0.1:6379;password=pass;db=7");
rds.Log = XTrace.Log; // Debug log. Comments for formal use
rds.Log = XTrace.Log;
```
### Basic operations
Before the basic operation, let's do some preparation:
+ Create a new console project and add `XTrace.UseConsole();` at the beginning of the entry function to make it easier to view the debug logs.
+ Before you can test the code, you need to add the code that instantiates MemoryCache or Redis.
+ Prepare a model class User
``` csharp
### Basic Operations
Before performing basic operations, we need to do some preparation:
+ Create a new console project and add `XTrace.UseConsole();` at the beginning of the main function to easily view debug logs.
+ Before testing specific code, ensure you have instantiated either MemoryCache or Redis.
+ Prepare a model class `User`:
```csharp
class User
{
public String Name { get; set; }
@ -72,20 +81,22 @@ class User
}
```
Add, delete and check:
``` csharp
Add, Delete, Update, Query:
```csharp
var rds = new FullRedis("127.0.0.1", null, 7);
rds.Log = XTrace.Log;
rds.ClientLog = XTrace.Log; // Debug log. Comment out for production use.
var user = new User { Name = "NewLife", CreateTime = DateTime.Now };
rds.Set("user", user, 3600);
var user2 = rds.Get<User>("user");
XTrace.WriteLine("Json: {0}", user2.ToJson());
XTrace.WriteLine("Json: {0}", rds.Get<String>("user"));
if (rds.ContainsKey("user")) XTrace.WriteLine("exists!");
if (rds.ContainsKey("user")) XTrace.WriteLine("Exists!");
rds.Remove("user");
```
```
Implementation results:
``` csharp
Execution Result:
```csharp
14:14:25.990 1 N - SELECT 7
14:14:25.992 1 N - => OK
14:14:26.008 1 N - SETEX user 3600 [53]
@ -97,20 +108,22 @@ Implementation results:
14:14:26.066 1 N - Json: {"Name":"NewLife","CreateTime":"2018-09-25 14:14:25"}
14:14:26.067 1 N - EXISTS user
14:14:26.068 1 N - => 1
14:14:26.068 1 N - 存在！
14:14:26.068 1 N - Exists!
14:14:26.069 1 N - DEL user
14:14:26.070 1 N - => 1
```
```
When saving complex objects, Json serialization is used by default, so above you can get the results back by string and find exactly the Json string.
Redis strings are, essentially, binary data with a length prefix; [53] represents a 53-byte length piece of binary data.
When saving complex objects, the default serialization method is JSON. Therefore, when retrieving the result as a string, you'll find it is in JSON format.
Redis strings are essentially binary data with length prefixes, where [53] indicates 53 bytes of binary data.
### Collection operations
GetAll/SetAll is a very common batch operation on Redis to get or set multiple keys at the same time, typically with 10x more throughput.
### Collection Operations
GetAll/SetAll are commonly used batch operations in Redis, allowing you to get or set multiple keys simultaneously, often achieving more than 10x throughput.
Batch operation:
``` csharp
Batch operations:
```csharp
var rds = new FullRedis("127.0.0.1", null, 7);
rds.Log = XTrace.Log;
rds.ClientLog = XTrace.Log; // Debug log. Comment out for production use.
var dic = new Dictionary<String, Object>
{
["name"] = "NewLife",
@ -121,10 +134,10 @@ rds.SetAll(dic, 120);
var vs = rds.GetAll<String>(dic.Keys);
XTrace.WriteLine(vs.Join(",", e => $"{e.Key}={e.Value}"));
```
```
Implementation results:
``` csharp
Execution Result:
```csharp
MSET name NewLife time 2018-09-25 15:56:26 count 1234
=> OK
EXPIRE name 120
@ -132,56 +145,75 @@ EXPIRE time 120
EXPIRE count 120
MGET name time count
name=NewLife,time=2018-09-25 15:56:26,count=1234
```
```
There are also `GetList/GetDictionary/GetQueue/GetSet` four types of collections inside the set operation, which represent Redis lists, hashes, queues, Set collections, and so on.
The base version of Redis does not support these four collections, the full version [NewLife.Redis](https://github.com/NewLifeX/NewLife.Redis) does, and MemoryCache supports them directly.
In collection operations, there are also `GetList/GetDictionary/GetQueue/GetSet` for various types of collections like Redis lists, hashes, queues, and sets.
The basic Redis version does not support these collections, but the full version [NewLife.Redis](https://github.com/NewLifeX/NewLife.Redis) does, while MemoryCache directly supports them.
### Advanced operations
+ Add Adds the key when it does not exist, and returns false when it already exists.
+ Replace Replacement replaces the existing value with the new value and returns the old value.
+ Increment Accumulation, atomic operation
+ Decrement, atomic operation
### Advanced Operations
+ Add: Adds a key if it does not exist, returns false if the key already exists.
+ Replace: Replaces an existing value with a new value and returns the old value.
+ Increment: Increments a value atomically.
+ Decrement: Decrements a value atomically.
Advanced Operations:
``` csharp
Advanced operations:
```csharp
var rds = new FullRedis("127.0.0.1", null, 7);
rds.Log = XTrace.Log;
rds.ClientLog = XTrace.Log; // Debug log. Comment out for production use.
var flag = rds.Add("count", 5678);
XTrace.WriteLine(flag ? "Add Success" : "Add Failure");
XTrace.WriteLine(flag ? "Add Success" : "Add Failed");
var ori = rds.Replace("count", 777);
var count = rds.Get<Int32>("count");
XTrace.WriteLine("count replaced by {0} with {1}", ori, count);
XTrace.WriteLine("count changed from {0} to {1}", ori, count);
rds.Increment("count", 11);
var count2 = rds.Decrement("count", 10);
XTrace.WriteLine("count={0}", count2);
```
```
Implementation results:
``` csharp
Execution Result:
```csharp
SETNX count 5678
=> 0
Add failed
Add Failed
GETSET count 777
=> 1234
GET count
=> 777
Replace count with 777 instead of 1234
count changed from 1234 to 777
INCRBY count 11
=> 788
DECRBY count 10
=> 778
count=778
```
```
### Performance testing
Bench conducts stress tests on additions, deletions, and modifications in multiple groups based on the number of threads.
rand parameter, whether to generate random key/value.
batch Batch size, perform read and write operations in batches, optimized with GetAll/SetAll.
---
Redis sets AutoPipeline=100 by default to turn on pipeline operations when there is no batching, optimized for add/delete.
### Performance Testing
### Siblings of Redis ###
Redis implements the ICache interface, its twin MemoryCache, an in-memory cache with a ten million throughput rate.
Each application is strongly recommended to use ICache interface coding design and MemoryCache implementation for small data;
After the data increase (100,000), switch to Redis implementation without modifying the business code.
The **Bench** tool will perform pressure testing by dividing the workload into multiple groups based on the number of threads.
Parameters:
- **rand**: Whether to randomly generate keys/values.
- **batch**: Batch size, which optimizes read/write operations using GetAll/SetAll.
By default, Redis sets **AutoPipeline=100**, meaning it automatically pipelines operations for better performance during read and write operations when there’s no batching.
---
### Redis's Siblings
Redis implements the **ICache** interface, with its sibling, **MemoryCache**, being an in-memory cache with a throughput in the tens of millions.
It is highly recommended to design applications using the **ICache** interface. Use **MemoryCache** for small data, and when the data grows (to 100,000 entries), switch to **Redis** without needing to modify the business logic code.
### Redis Server Support
1. **Version 3.2 and above**:
From version 3.2 onwards, **FullRedis** supports all Redis versions.
2. **Stream Data Type**:
FullRedis supports the **Stream** data type, which was introduced in Redis 5.0, allowing the storage and manipulation of message streams.
3. **LUA Not Supported**:
FullRedis **does not support** LUA scripting, meaning you cannot execute Lua scripts through this client.

76
Samples/Benchmark/BasicBenchmark.cs Normal file
View File

@ -0,0 +1,76 @@
using System.Security.Cryptography;
using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Engines;
using NewLife.Caching;
using NewLife.Security;
namespace Benchmark;
[SimpleJob(RunStrategy.ColdStart, iterationCount: 1)]
[MemoryDiagnoser]
public class BasicBenchmark
{
public FullRedis Redis { get; set; }
private String _key;
private String[] _keys;
[GlobalSetup]
public void Setup()
{
var rds = new FullRedis
{
Tracer = DefaultTracer.Instance,
Log = XTrace.Log,
};
rds.Init("server=127.0.0.1:6379;password=;db=3;timeout=5000");
Redis = rds;
_key = Rand.NextString(16);
var ks = new String[100_000];
for (var i = 0; i < ks.Length; i++)
{
ks[i] = Rand.NextString(16);
}
_keys = ks;
rds.Set(_key, _key);
var v = rds.Get<String>(_key);
rds.Remove(_key);
}
[Benchmark]
public void SetTest()
{
var rds = Redis;
var value = Rand.NextString(16);
for (var i = 0; i < _keys.Length; i++)
{
rds.Set(_keys[i], value);
}
}
[Benchmark]
public void GetTest()
{
var rds = Redis;
for (var i = 0; i < _keys.Length; i++)
{
var value = rds.Get<String>(_keys[i]);
}
}
[Benchmark]
public void RemoveTest()
{
var rds = Redis;
for (var i = 0; i < _keys.Length; i++)
{
rds.Remove(_keys[i]);
}
}
}

33
Samples/Benchmark/Benchmark.csproj Normal file
View File

@ -0,0 +1,33 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net8.0</TargetFramework>
<AssemblyTitle>应用后台任务</AssemblyTitle>
<Description>数据处理、定时任务、MQ生产消费、系统监控等超长独立工作的后台任务</Description>
<Company>新生命开发团队</Company>
<Copyright>©2002-2025 NewLife</Copyright>
<VersionPrefix>1.0</VersionPrefix>
<VersionSuffix>$([System.DateTime]::Now.ToString(`yyyy.MMdd`))</VersionSuffix>
<Version>$(VersionPrefix).$(VersionSuffix)</Version>
<FileVersion>$(Version)</FileVersion>
<OutputPath>..\..\Bin\Benchmark</OutputPath>
<AppendTargetFrameworkToOutputPath>false</AppendTargetFrameworkToOutputPath>
<ImplicitUsings>enable</ImplicitUsings>
<LangVersion>latest</LangVersion>
</PropertyGroup>
<ItemGroup>
<Using Include="NewLife" />
<Using Include="NewLife.Log" />
<Using Include="NewLife.Model" />
<Using Include="NewLife.Reflection" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="BenchmarkDotNet" Version="0.14.0" />
<PackageReference Include="NewLife.Stardust" Version="3.3.2025.202" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\NewLife.Redis\NewLife.Redis.csproj" />
</ItemGroup>
</Project>

20
Samples/Benchmark/Program.cs Normal file
View File

@ -0,0 +1,20 @@
using Benchmark;
using BenchmarkDotNet.Running;
using NewLife.Caching;
using Stardust;
//!!! 标准后台服务项目模板，新生命团队强烈推荐
// 启用控制台日志，拦截所有异常
XTrace.UseConsole();
// 初始化对象容器，提供依赖注入能力
var services = ObjectContainer.Current;
services.AddSingleton(XTrace.Log);
// 配置星尘。自动读取配置文件 config/star.config 中的服务器地址
var star = services.AddStardust();
var summary = BenchmarkRunner.Run<BasicBenchmark>();
Console.ReadLine();

22
Samples/Benchmark/appsettings.json Normal file
View File

@ -0,0 +1,22 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft": "Warning",
"Microsoft.Hosting.Lifetime": "Information"
}
},
"AllowedHosts": "*",
//"StarServer": "http://s.newlifex.com:6600",
//"RedisCache": "server=127.0.0.1:6379;password=;db=3",
//"RedisQueue": "server=127.0.0.1:6379;password=;db=5",
"ConnectionStrings": {
"Zero": "Data Source=..\\Data\\Zero.db;Provider=SQLite"
// 各种数据库连接字符串模版，连接名Zero对应Zero.Data/Projects/Model.xml中的ConnName
//"Zero": "Server=.;Port=3306;Database=zero;Uid=root;Pwd=root;Provider=MySql",
//"Zero": "Data Source=.;Initial Catalog=zero;user=sa;password=sa;Provider=SqlServer",
//"Zero": "Server=.;Database=zero;Uid=root;Pwd=root;Provider=PostgreSql",
//"Zero": "Data Source=Tcp://127.0.0.1/ORCL;User Id=scott;Password=tiger;Provider=Oracle"
}
}

0
QueueDemo/AckQueue.cs → Samples/QueueDemo/AckQueue.cs
View File

0
QueueDemo/Area.cs → Samples/QueueDemo/Area.cs
View File

0
QueueDemo/DelayQueue.cs → Samples/QueueDemo/DelayQueue.cs
View File

0
QueueDemo/EasyQueue.cs → Samples/QueueDemo/EasyQueue.cs
View File

0
QueueDemo/FullQueue.cs → Samples/QueueDemo/FullQueue.cs
View File

0
QueueDemo/MemoryQueue.cs → Samples/QueueDemo/MemoryQueue.cs
View File

0
QueueDemo/MultipleConsumer.cs → Samples/QueueDemo/MultipleConsumer.cs
View File

0
QueueDemo/Program.cs → Samples/QueueDemo/Program.cs
View File

4
QueueDemo/QueueDemo.csproj → Samples/QueueDemo/QueueDemo.csproj
View File

@ -5,14 +5,14 @@
<TargetFramework>net8.0</TargetFramework>
<AssemblyVersion>1.0.*</AssemblyVersion>
<Deterministic>false</Deterministic>
<OutputPath>..\Bin\QueueDemo</OutputPath>
<OutputPath>..\..\Bin\QueueDemo</OutputPath>
<AppendTargetFrameworkToOutputPath>false</AppendTargetFrameworkToOutputPath>
<ImplicitUsings>enable</ImplicitUsings>
<LangVersion>latest</LangVersion>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\NewLife.Redis\NewLife.Redis.csproj" />
<ProjectReference Include="..\..\NewLife.Redis\NewLife.Redis.csproj" />
</ItemGroup>
</Project>

12
Test/Program.cs
View File

@ -7,6 +7,7 @@ using System.Threading.Tasks;
using NewLife;
using NewLife.Caching;
using NewLife.Caching.Clusters;
using NewLife.Configuration;
using NewLife.Log;
using NewLife.Security;
using NewLife.Serialization;
@ -97,7 +98,16 @@ class Program
/// <summary>性能压测</summary>
static void Test2()
{
var ic = new FullRedis("127.0.0.1", null, 3);
var args = Environment.GetCommandLineArgs();
var cp = new CommandParser();
var dic = cp.Parse(args);
if (!dic.TryGetValue("server", out var server)) server = "127.0.0.1";
if (!dic.TryGetValue("pass", out var pass)) pass = "";
if (server.IsNullOrEmpty()) server = "127.0.0.1";
var ic = new FullRedis(server, pass, 3);
// 性能压测
//ic.AutoPipeline = -1;

8
Test/Test.csproj
View File

@ -1,10 +1,10 @@
<Project Sdk="Microsoft.NET.Sdk">
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFrameworks>net45;net461;net8.0</TargetFrameworks>
<TargetFrameworks>net45;net461;net8.0;net9.0</TargetFrameworks>
<Company>新生命开发团队</Company>
<Copyright>©2002-2024 新生命开发团队</Copyright>
<Copyright>©2002-2025 新生命开发团队</Copyright>
<VersionPrefix>1.0</VersionPrefix>
<VersionSuffix>$([System.DateTime]::Now.ToString(`yyyy.MMdd`))</VersionSuffix>
<Version>$(VersionPrefix).$(VersionSuffix)</Version>
@ -16,7 +16,7 @@
</PropertyGroup>
<ItemGroup>
<PackageReference Include="NewLife.Core" Version="11.0.2024.923-beta0014" />
<PackageReference Include="NewLife.Core" Version="11.4.2025.301" />
</ItemGroup>
<ItemGroup>

21
XUnitTest/HashTest.cs
View File

@ -1,6 +1,9 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Runtime.Intrinsics.Arm;
using NewLife.Caching;
using NewLife.Log;
using Xunit;
@ -93,6 +96,24 @@ public class HashTest
Assert.Equal(org5, hash["org5"]);
}
[Fact]
public void CheckHashTest()
{
var key = $"NewLife:eventinfo:adsfasdfasdfdsaf";
var hash = _redis.GetDictionary<EventInfo>(key);
Assert.NotNull(hash);
var l = hash as RedisHash<String, String>;
foreach(var item in l.GetAll())
{
XTrace.WriteLine(item.Key);
}
l["0"] = "0";
}
}
public class HashTest2 : HashTest

1
XUnitTest/Queues/QueueTests.cs
View File

@ -13,6 +13,7 @@ using Xunit;
namespace XUnitTest.Queues;
//[Collection("Queue")]
[TestCaseOrderer("NewLife.UnitTest.DefaultOrderer", "NewLife.UnitTest")]
public class QueueTests
{
private FullRedis _redis;

6
XUnitTest/Queues/ReliableQueueTests.cs
View File

@ -14,6 +14,7 @@ using Xunit;
namespace XUnitTest.Queues;
//[Collection("Queue")]
[TestCaseOrderer("NewLife.UnitTest.DefaultOrderer", "NewLife.UnitTest")]
public class ReliableQueueTests
{
private readonly FullRedis _redis;
@ -672,13 +673,12 @@ public class ReliableQueueTests
{
var queue = _redis.GetReliableQueue<RedisMessage<MyModel>>("TakeOneNotDataAsync");
queue.RetryInterval = 60;//重新处理确认队列中死信的间隔。默认60s
RedisMessage<MyModel>? message = await queue.TakeOneAsync(10);
var message = await queue.TakeOneAsync(3);
Assert.Null(message);
var queue2 = _redis.GetReliableQueue<Int32>("TakeOneNotDataAsync_Int32");
queue2.RetryInterval = 60;//重新处理确认队列中死信的间隔。默认60s
int messageInt = await queue2.TakeOneAsync(10);
var messageInt = await queue2.TakeOneAsync(3);
Assert.Equal(0, messageInt);
}
}

2
XUnitTest/Queues/StreamTests.cs
View File

@ -4,6 +4,7 @@ using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using NewLife;
using NewLife.Caching;
using NewLife.Caching.Queues;
using NewLife.Log;
@ -13,6 +14,7 @@ using Xunit;
namespace XUnitTest.Queues;
[TestCaseOrderer("NewLife.UnitTest.DefaultOrderer", "NewLife.UnitTest")]
public class StreamTests
{
private FullRedis _redis;

19
XUnitTest/RedisLockTest.cs
View File

@ -80,45 +80,44 @@ public class RedisLockTest
[TestOrder(54)]
[Fact(DisplayName = "抢锁失败2")]
public void TestLock22()
public void TestLock3()
{
var ic = _redis;
var ck1 = ic.AcquireLock("lock:TestLock2", 2000);
var ck1 = ic.AcquireLock("lock:TestLock3", 2000);
// 故意不用using，验证GC是否能回收
//using var ck1 = ic.AcquireLock("TestLock2", 3000);
//using var ck1 = ic.AcquireLock("TestLock3", 3000);
var sw = Stopwatch.StartNew();
// 抢相同锁，不可能成功。超时时间必须小于3000，否则前面的锁过期后，这里还是可以抢到的
var ck2 = ic.AcquireLock("lock:TestLock2", 1000, 1000, false);
var ck2 = ic.AcquireLock("lock:TestLock3", 1000, 1000, false);
Assert.Null(ck2);
// 耗时必须超过有效期
sw.Stop();
XTrace.WriteLine("TestLock2 ElapsedMilliseconds={0}ms", sw.ElapsedMilliseconds);
XTrace.WriteLine("TestLock3 ElapsedMilliseconds={0}ms", sw.ElapsedMilliseconds);
Assert.True(sw.ElapsedMilliseconds >= 1000);
Thread.Sleep(2000 - 1000 + 100);
// 那个锁其实已经不在了，缓存应该把它干掉
Assert.False(ic.ContainsKey("lock:TestLock2"));
Assert.False(ic.ContainsKey("lock:TestLock3"));
}
[TestOrder(56)]
[Fact(DisplayName = "抢死锁")]
public void TestLock3()
public void TestLock4()
{
var ic = _redis;
using var ck = ic.AcquireLock("TestLock3", 1000);
using var ck = ic.AcquireLock("TestLock4", 1000);
// 已经过了一点时间
Thread.Sleep(500);
// 循环多次后，可以抢到
using var ck2 = ic.AcquireLock("TestLock3", 1000);
using var ck2 = ic.AcquireLock("TestLock4", 1000);
Assert.NotNull(ck2);
}
}

33
XUnitTest/RedisTest.cs
View File

@ -1,10 +1,7 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Runtime.Serialization;
using System.Security.Cryptography;
using System.Threading;
using System.Threading.Tasks;
using NewLife;
@ -37,6 +34,9 @@ public class RedisTest
#if DEBUG
_redis.ClientLog = XTrace.Log;
#endif
// 测试高级功能，如果keys过多，则清空
if (_redis.Count > 10000) _redis.Clear();
}
[TestOrder(0)]
@ -294,6 +294,24 @@ public class RedisTest
Assert.Equal(pk.ToHex(), pk2.ToHex());
}
/// <summary>超大数据包</summary>
/// <remarks>https://github.com/NewLifeX/NewLife.Redis/issues/149</remarks>
[TestOrder(31)]
[Fact(DisplayName = "超大数据包")]
public void TestBigPacket()
{
var ic = _redis;
var key = "buf";
var buf = Rand.NextBytes(8192 + 1);
var pk = new ArrayPacket(buf);
ic.Set(key, pk);
var pk2 = ic.Get<IPacket>(key);
Assert.Equal(pk.ToHex(), pk2.ToHex());
}
[TestOrder(40)]
[Fact(DisplayName = "管道")]
public void TestPipeline()
@ -516,10 +534,11 @@ public class RedisTest
var ic = _redis;
ic.MaxMessageSize = 1028;
//ic.Retry = 0;
var ex = Assert.Throws<AggregateException>(() => ic.Set("ttt", Rand.NextString(1029)));
var ex2 = ex.GetTrue() as InvalidOperationException;
Assert.NotNull(ex2);
Assert.Equal("命令[SET]的数据包大小[1060]超过最大限制[1028]，大key会拖累整个Redis实例，可通过Redis.MaxMessageSize调节。", ex2.Message);
var ex = Assert.Throws<InvalidOperationException>(() => ic.Set("ttt", Rand.NextString(1029)));
//var ex2 = ex.GetTrue() as InvalidOperationException;
Assert.NotNull(ex);
Assert.Equal("命令[SET]的数据包大小[1060]超过最大限制[1028]，大key会拖累整个Redis实例，可通过Redis.MaxMessageSize调节。", ex.Message);
}
}

63
XUnitTest/SearchTest.cs Normal file
View File

@ -0,0 +1,63 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using NewLife.Caching;
using NewLife.Log;
using Xunit;
namespace XUnitTest
{
[Collection("Basic")]
public class SearchTest
{
protected readonly FullRedis _redis;
public SearchTest()
{
var config = BasicTest.GetConfig();
_redis = new FullRedis();
_redis.Init(config);
_redis.Db = 2;
_redis.Retry = 0;
_redis.Log = XTrace.Log;
#if DEBUG
_redis.ClientLog = XTrace.Log;
#endif
}
[Fact(DisplayName = "搜索测试")]
public void GetSearchTest()
{
var ic = _redis;
for (int i = 0; i < 1000; i++)
{
PlayGameVo playGameVo = new PlayGameVo()
{
MemberId = (i + 1).ToString(),
GameMode = 1,
Num = 10000
};
//cache.Cache.Set(RedisConst.PlayGameKey + playGameVo.MemberId, playGameVo);
//redis.Prefix=RedisConst.PlayGameKey;
ic.Set("jinshi:member:battle-royale:play-game:" + playGameVo.MemberId, playGameVo);
}
IDictionary<string, string> all = new Dictionary<string, string>();
List<string> list = ic.Search("jinshi:member:battle-royale:play-game:*", 1000).ToList();
all = ic.GetAll<string>(list);
Assert.True(list.Count==1000);
Assert.False(list.Count < 1000);
}
}
public class PlayGameVo
{
public string MemberId { get; set; }
public int GameMode { get; set; }
public int Num { get; set; }
}
}

6
XUnitTest/Services/RedisCacheProviderTests.cs
View File

@ -1,5 +1,6 @@
using System;
using NewLife.Caching.Services;
using NewLife.Configuration;
using NewLife.Model;
using Xunit;
@ -10,7 +11,10 @@ public class RedisCacheProviderTests
[Fact]
public void Ctor()
{
var sp = ObjectContainer.Provider;
var services = ObjectContainer.Current;
services.AddSingleton<IConfigProvider>(JsonConfigProvider.LoadAppSettings());
var sp = services.BuildServiceProvider();
var provider = new RedisCacheProvider(sp);
Assert.NotNull(provider.Cache);

12
XUnitTest/XUnitTest.csproj
View File

@ -8,12 +8,12 @@
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="8.0.0" />
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.11.1" />
<PackageReference Include="NewLife.Core" Version="11.0.2024.923-beta0014" />
<PackageReference Include="NewLife.UnitTest" Version="1.0.2023.1204" />
<PackageReference Include="xunit" Version="2.9.2" />
<PackageReference Include="xunit.runner.visualstudio" Version="2.8.2">
<PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="9.0.0" />
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.13.0" />
<PackageReference Include="NewLife.Core" Version="11.4.2025.301" />
<PackageReference Include="NewLife.UnitTest" Version="1.0.2025.101" />
<PackageReference Include="xunit" Version="2.9.3" />
<PackageReference Include="xunit.runner.visualstudio" Version="3.0.2">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>