控制 Redis stream 的消息数量

Posted 2021-10-05 dotNET跨平台

控制 Redis stream 的消息数量

Intro

Redis Stream 是 Redis 5.0 引入的一个新的类型，之前我们介绍过使用 Redis Stream 来实现消息队列，可以参考之前的文章 使用 Redis Stream 实现消息队列，而 Stream 的消息会持久化地内存中，如果我们不控制消息数量的话，可能会出现大量的消息存在内存里导致过大的内存占用，Redis Stream 5.0 开始支持根据 Max Length 来控制 Stream 的长度（消息数量），从 6.2 开始支持根据消息 Id 来控制 Stream 的长度，默认地消息 Id 是一个时间戳，所以使用默认地 Id 也可以理解为按时间来控制 Stream 长度，下面我们来看使用示例吧

Redis 语法

控制 Stream 消息长度有两个 Redis 命令，一个是 XTRIM 只做 Trim 操作，把不满足要求的消息去除，另外一个是 XADD 在添加 Stream 消息的同时做 Trim 操作，简化还要多一步 Trim 的操作，将添加消息和控制消息长度可以合并为一个操作

XTRIM 语法：

XTRIM key MAXLEN|MINID [=|~] threshold [LIMIT count]

使用示例：

XTRIM mystream MAXLEN 1000
XTRIM mystream MINID 649085820

XTRIM mystream MAXLEN ~ 1000（Nearly trim，不准确，可能有些消息该删掉的会保留下来，但是执行效率会比 `=`(Exactly Trim) 高一些）
XTRIM mystream MINID = 649085820

Trimming the stream can be done using one of these strategies:

• MAXLEN: Evicts entries as long as the stream's length exceeds the specified threshold, where threshold is a positive integer.

• MINID: Evicts entries with IDs lower than threshold, where threshold is a stream ID.

XADD 语法：

XADD key [NOMKSTREAM] [MAXLEN|MINID [=|~] threshold [LIMIT count]] *|ID field value [field value ...]

使用示例：

redis> XADD mystream * name Sara surname OConnor
"1631546114612-0"
redis> XADD mystream * field1 value1 field2 value2 field3 value3
"1631546114612-1"

redis> XADD mystream MAXLEN ~ 1000 field1 value1
redis> XADD mystream MINID ~ 1631546460687 field1 value1

XADD 允许用户在向 Stream 里添加消息的时候控制消息的长度返回值是消息ID，默认是一个时间戳，语法如上，可以 Trim 也可以 不Trim，可以根据需要选择

Prepare

先来准备一些帮助类和公共方法，下面的示例是基于 StackExchange.Redis 来实现的

RedisHelper，获取 Redis 连接

internal static class RedisHelper
{
    private static readonly IConnectionMultiplexer ConnectionMultiplexer = StackExchange.Redis.ConnectionMultiplexer.Connect("127.0.0.1:6379");

    public static IDatabase GetRedisDb(int dbIndex = 0)
    {
        return ConnectionMultiplexer.GetDatabase(dbIndex);
    }
}

AddStreamMessage，向指定 stream 中添加若干条消息

private static async Task AddStreamMessage(string key, int msgCount, Action action=null)
{
    var redis = RedisHelper.GetRedisDb();
    for (var i = 0; i < msgCount; i++)
    {
        await redis.StreamAddAsync(key, "messages", $"val-{i}");
        action?.Invoke();
    }
}

Max-Length

根据 MaxLength 来控制 Stream 长度示例

var streamKey = $"stream-{nameof(MaxLengthTrim)}";
await AddStreamMessage(streamKey, 10);
var redis = RedisHelper.GetRedisDb();
Console.WriteLine(await redis.StreamLengthAsync(streamKey));

// trim directly
await redis.StreamTrimAsync(streamKey, 5);
Console.WriteLine(await redis.StreamLengthAsync(streamKey));

// add with trim
await redis.StreamAddAsync(streamKey, StreamMessageField, "Test", maxLength: 3);
Console.WriteLine(await redis.StreamLengthAsync(streamKey));

await redis.KeyDeleteAsync(streamKey);

输出结果如下：

Min-ID

根据 Min-ID 来控制 Stream 消息长度，是 Redis 6.2 新引入的功能，目前 StackExchange.Redis 还没有专门的 API 来支持这个功能，不过我们可以通过 Execute 来执行 Redis 命令，通常这些 Redis 客户端库都会支持直接调用 Redis 命令，根据 MinID 控制 Stream 长度示例如下：

private const string StreamAddCommandName = "XADD";
private const string StreamTrimCommandName = "XTRIM";

private const string StreamAddAutoMsgId = "*";

private const string StreamTrimByMinIdName = "MINID";

private const string StreamTrimOperator = "=";

private const string StreamMessageField = "message";

private static async Task MinMsgIdTrim()
{
    var streamKey = $"stream-{nameof(MaxLengthTrim)}";
    await AddStreamMessage(streamKey, 10, () => Thread.Sleep(1000));

    var redis = RedisHelper.GetRedisDb();
    var minId = DateTimeOffset.UtcNow.Subtract(TimeSpan.FromSeconds(5)).ToUnixTimeMilliseconds();
    Console.WriteLine(await redis.StreamLengthAsync(streamKey));

    // https://redis.io/commands/xtrim
    // trim directly
    await redis.ExecuteAsync(
        StreamTrimCommandName, 
        streamKey,
        StreamTrimByMinIdName,
        StreamTrimOperator, // optional
        minId
    );
    Console.WriteLine(await redis.StreamLengthAsync(streamKey));
    minId = DateTimeOffset.UtcNow.Subtract(TimeSpan.FromSeconds(2)).ToUnixTimeMilliseconds();

    // https://redis.io/commands/xadd
    // add with trim
    var result = redis.Execute(
        StreamAddCommandName, 
        streamKey, 
        StreamTrimByMinIdName, 
        StreamTrimOperator, // optional
        minId,
        StreamAddAutoMsgId, 
        StreamMessageField, 
        "Test"
    );
    Console.WriteLine(await redis.StreamLengthAsync(streamKey));

    await redis.KeyDeleteAsync(streamKey);
}

上述代码输出结果如下：

More

本文主要介绍了控制 Redis Stream 的消息长度，除了介绍 Redis 本身的命令之外，也是介绍一下如何使用 StackExchange.Redis 实现调用没有 API 支持的 Redis 命令，Redis 6.2 之后支持了很多新的特性，但是很多库都还太支持，了解如何原生调用 Redis 命令有些时候会很有帮助

References

• https://redis.io/commands/xadd

• https://redis.io/commands/xtrim

• https://github.com/WeihanLi/SamplesInPractice/blob/master/RedisSample/StreamTrimSample.cs