perf: append chat cache atomically
This commit is contained in:
xiangning
parent
3b918f5c0b
commit
156fb01fdb
@ -1,39 +1,95 @@
|
||||
using System.Text.Json;
|
||||
using BuildFlow.Domain.Interfaces;
|
||||
using Microsoft.Extensions.Options;
|
||||
using StackExchange.Redis;
|
||||
|
||||
namespace BuildFlow.Infrastructure.Cache;
|
||||
|
||||
/// <summary>
|
||||
/// 聊天消息 Redis 缓存。以整个会话的消息列表为粒度序列化为单个 JSON 串存储（Key=chat:messages:{convId}），
|
||||
/// TTL 1 小时，超时自动失效后回源 DB。用作流式接口与最近消息的热读加速层，DB 才是真源。
|
||||
/// 聊天消息 Redis 缓存。每条消息是 List 的一个 JSON 元素，DB 是权威数据源。
|
||||
/// 追加和 TTL 刷新由同一个 Lua 脚本原子执行，避免并发读改写丢消息。
|
||||
/// </summary>
|
||||
public class ChatMessageCache(IConnectionMultiplexer redis) : IChatMessageCache
|
||||
public class ChatMessageCache(
|
||||
IConnectionMultiplexer redis,
|
||||
IOptions<ChatOptions> options) : IChatMessageCache
|
||||
{
|
||||
private const string AppendScript = """
|
||||
redis.call('RPUSH', KEYS[1], ARGV[1])
|
||||
redis.call('PEXPIRE', KEYS[1], ARGV[2])
|
||||
return redis.call('LLEN', KEYS[1])
|
||||
""";
|
||||
|
||||
private const string ReplaceScript = """
|
||||
redis.call('DEL', KEYS[1])
|
||||
for index = 1, #ARGV - 1 do
|
||||
redis.call('RPUSH', KEYS[1], ARGV[index])
|
||||
end
|
||||
redis.call('PEXPIRE', KEYS[1], ARGV[#ARGV])
|
||||
return #ARGV - 1
|
||||
""";
|
||||
|
||||
private readonly IDatabase _db = redis.GetDatabase();
|
||||
private readonly long _ttlMilliseconds =
|
||||
checked((long)TimeSpan.FromMinutes(options.Value.CacheTtlMinutes).TotalMilliseconds);
|
||||
private static readonly JsonSerializerOptions JsonOptions = new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase };
|
||||
|
||||
private static string Key(Guid conversationId) => $"chat:messages:{conversationId}";
|
||||
|
||||
public async Task<List<CachedChatMessage>?> GetMessagesAsync(Guid conversationId, CancellationToken ct)
|
||||
{
|
||||
var value = await _db.StringGetAsync(Key(conversationId));
|
||||
if (value.IsNullOrEmpty) return null;
|
||||
return JsonSerializer.Deserialize<List<CachedChatMessage>>((string)value!, JsonOptions);
|
||||
ct.ThrowIfCancellationRequested();
|
||||
var values = await _db.ListRangeAsync(Key(conversationId), 0, -1);
|
||||
ct.ThrowIfCancellationRequested();
|
||||
if (values.Length == 0)
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
var messages = new List<CachedChatMessage>(values.Length);
|
||||
foreach (var value in values)
|
||||
{
|
||||
var message = JsonSerializer.Deserialize<CachedChatMessage>((string)value!, JsonOptions);
|
||||
if (message is not null)
|
||||
{
|
||||
messages.Add(message);
|
||||
}
|
||||
}
|
||||
|
||||
return messages;
|
||||
}
|
||||
|
||||
public async Task SetMessagesAsync(Guid conversationId, List<CachedChatMessage> messages, CancellationToken ct)
|
||||
{
|
||||
var json = JsonSerializer.Serialize(messages, JsonOptions);
|
||||
await _db.StringSetAsync(Key(conversationId), json, TimeSpan.FromHours(1));
|
||||
ct.ThrowIfCancellationRequested();
|
||||
if (messages.Count == 0)
|
||||
{
|
||||
await _db.KeyDeleteAsync(Key(conversationId));
|
||||
ct.ThrowIfCancellationRequested();
|
||||
return;
|
||||
}
|
||||
|
||||
var values = new RedisValue[messages.Count + 1];
|
||||
for (var index = 0; index < messages.Count; index++)
|
||||
{
|
||||
values[index] = JsonSerializer.Serialize(messages[index], JsonOptions);
|
||||
}
|
||||
values[^1] = _ttlMilliseconds;
|
||||
|
||||
await _db.ScriptEvaluateAsync(
|
||||
ReplaceScript,
|
||||
[Key(conversationId)],
|
||||
values);
|
||||
ct.ThrowIfCancellationRequested();
|
||||
}
|
||||
|
||||
// 追加采用读-改-写：本场景无并发写入同一会话（单用户顺序发送），未引入锁以保持简单
|
||||
public async Task AppendMessageAsync(Guid conversationId, CachedChatMessage message, CancellationToken ct)
|
||||
{
|
||||
var messages = await GetMessagesAsync(conversationId, ct) ?? [];
|
||||
messages.Add(message);
|
||||
await SetMessagesAsync(conversationId, messages, ct);
|
||||
ct.ThrowIfCancellationRequested();
|
||||
await _db.ScriptEvaluateAsync(
|
||||
AppendScript,
|
||||
[Key(conversationId)],
|
||||
[JsonSerializer.Serialize(message, JsonOptions), _ttlMilliseconds]);
|
||||
ct.ThrowIfCancellationRequested();
|
||||
}
|
||||
|
||||
public async Task RemoveAsync(Guid conversationId, CancellationToken ct)
|
||||
|
||||
@ -0,0 +1,11 @@
|
||||
using Xunit;
|
||||
|
||||
namespace BuildFlow.Rag.Tests.Chat;
|
||||
|
||||
/// <summary>串行执行真实 Redis 聊天缓存测试，避免共享实例上的时序干扰。</summary>
|
||||
[CollectionDefinition(Name, DisableParallelization = true)]
|
||||
public sealed class RedisChatMessageCacheCollection : ICollectionFixture<RedisChatMessageCacheFixture>
|
||||
{
|
||||
/// <summary>集合名称。</summary>
|
||||
public const string Name = "rag-chat-redis";
|
||||
}
|
||||
@ -0,0 +1,33 @@
|
||||
using StackExchange.Redis;
|
||||
using Xunit;
|
||||
|
||||
namespace BuildFlow.Rag.Tests.Chat;
|
||||
|
||||
/// <summary>管理连接到本地独立基础设施 Redis 的测试连接。</summary>
|
||||
public sealed class RedisChatMessageCacheFixture : IAsyncLifetime
|
||||
{
|
||||
private const string DefaultConnection = "127.0.0.1:56379,abortConnect=false";
|
||||
|
||||
/// <summary>测试复用的 Redis 连接。</summary>
|
||||
public IConnectionMultiplexer Redis { get; private set; } = default!;
|
||||
|
||||
public async Task InitializeAsync()
|
||||
{
|
||||
var connection = Environment.GetEnvironmentVariable("BUILDFLOW_RAG_REDIS_TEST_CONNECTION")
|
||||
?? DefaultConnection;
|
||||
Redis = await ConnectionMultiplexer.ConnectAsync(connection);
|
||||
}
|
||||
|
||||
public Task DisposeAsync()
|
||||
{
|
||||
Redis.Dispose();
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
/// <summary>删除单个测试会话的固定缓存 key。</summary>
|
||||
public Task DeleteAsync(Guid conversationId)
|
||||
=> Redis.GetDatabase().KeyDeleteAsync(Key(conversationId));
|
||||
|
||||
/// <summary>返回生产缓存使用的固定 key。</summary>
|
||||
public static string Key(Guid conversationId) => $"chat:messages:{conversationId}";
|
||||
}
|
||||
101
tests/BuildFlow.Rag.Tests/Chat/RedisChatMessageCacheTests.cs
Normal file
101
tests/BuildFlow.Rag.Tests/Chat/RedisChatMessageCacheTests.cs
Normal file
@ -0,0 +1,101 @@
|
||||
using BuildFlow.Domain.Interfaces;
|
||||
using BuildFlow.Infrastructure.Cache;
|
||||
using FluentAssertions;
|
||||
using Microsoft.Extensions.Options;
|
||||
using StackExchange.Redis;
|
||||
using Xunit;
|
||||
|
||||
namespace BuildFlow.Rag.Tests.Chat;
|
||||
|
||||
/// <summary>Redis List 聊天缓存的并发追加、传输粒度、顺序和 TTL 集成测试。</summary>
|
||||
[Collection(RedisChatMessageCacheCollection.Name)]
|
||||
public sealed class RedisChatMessageCacheTests(RedisChatMessageCacheFixture fixture)
|
||||
{
|
||||
[Fact]
|
||||
public async Task ConcurrentAppendsPreserveEveryMessage()
|
||||
{
|
||||
var conversationId = Guid.NewGuid();
|
||||
await fixture.DeleteAsync(conversationId);
|
||||
var cache = CreateCache();
|
||||
var messages = Enumerable.Range(0, 100)
|
||||
.Select(index => CreateMessage(index))
|
||||
.ToArray();
|
||||
|
||||
await Task.WhenAll(messages.Select(message =>
|
||||
cache.AppendMessageAsync(conversationId, message, CancellationToken.None)));
|
||||
|
||||
var stored = await cache.GetMessagesAsync(conversationId, CancellationToken.None);
|
||||
var database = fixture.Redis.GetDatabase();
|
||||
stored.Should().NotBeNull();
|
||||
var storedMessages = stored!;
|
||||
storedMessages.Should().HaveCount(100);
|
||||
storedMessages.Select(x => x.Id).Should().OnlyHaveUniqueItems();
|
||||
(await database.KeyTypeAsync(RedisChatMessageCacheFixture.Key(conversationId)))
|
||||
.Should().Be(RedisType.List);
|
||||
(await database.ListLengthAsync(RedisChatMessageCacheFixture.Key(conversationId)))
|
||||
.Should().Be(100);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task AppendAddsOneListElementAndRefreshesConfiguredTtl()
|
||||
{
|
||||
var conversationId = Guid.NewGuid();
|
||||
await fixture.DeleteAsync(conversationId);
|
||||
var cache = CreateCache();
|
||||
var existing = Enumerable.Range(0, 500)
|
||||
.Select(CreateMessage)
|
||||
.ToList();
|
||||
await cache.SetMessagesAsync(conversationId, existing, CancellationToken.None);
|
||||
var database = fixture.Redis.GetDatabase();
|
||||
var key = RedisChatMessageCacheFixture.Key(conversationId);
|
||||
|
||||
await cache.AppendMessageAsync(
|
||||
conversationId,
|
||||
CreateMessage(500),
|
||||
CancellationToken.None);
|
||||
|
||||
(await database.KeyTypeAsync(key)).Should().Be(RedisType.List);
|
||||
(await database.ListLengthAsync(key)).Should().Be(501);
|
||||
var ttl = await database.KeyTimeToLiveAsync(key);
|
||||
ttl.Should().NotBeNull();
|
||||
ttl!.Value.Should().BeGreaterThan(TimeSpan.FromMinutes(4));
|
||||
ttl.Value.Should().BeLessThanOrEqualTo(TimeSpan.FromMinutes(5));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SetAndGetRoundTripCompleteHistoryInOrder()
|
||||
{
|
||||
var conversationId = Guid.NewGuid();
|
||||
await fixture.DeleteAsync(conversationId);
|
||||
var cache = CreateCache();
|
||||
var messages = Enumerable.Range(0, 20)
|
||||
.Select(CreateMessage)
|
||||
.ToList();
|
||||
|
||||
await cache.SetMessagesAsync(conversationId, messages, CancellationToken.None);
|
||||
var result = await cache.GetMessagesAsync(conversationId, CancellationToken.None);
|
||||
|
||||
result.Should().NotBeNull();
|
||||
var storedMessages = result!;
|
||||
storedMessages.Select(x => x.Id).Should().Equal(messages.Select(x => x.Id));
|
||||
storedMessages.Select(x => x.Content).Should().Equal(messages.Select(x => x.Content));
|
||||
}
|
||||
|
||||
private ChatMessageCache CreateCache()
|
||||
=> new(
|
||||
fixture.Redis,
|
||||
Options.Create(new ChatOptions
|
||||
{
|
||||
MaxContextMessages = 50,
|
||||
MaxContextTokens = 8000,
|
||||
CacheTtlMinutes = 5,
|
||||
}));
|
||||
|
||||
private static CachedChatMessage CreateMessage(int index)
|
||||
=> new(
|
||||
Guid.NewGuid(),
|
||||
index % 2 == 0 ? "User" : "Assistant",
|
||||
$"message-{index}",
|
||||
null,
|
||||
new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc).AddMinutes(index));
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user