去重复流记录

Question

我用Redis来传输数据。我有多个生产者实例产生相同的数据，目的是事件一致性。

现在，生产者用1到2之间的随机交易id生成交易。我想要一个去重复服务或基于交易id的东西来分散复制。我该怎么做？

消费者

using System.Text.Json;
using Shared;
using StackExchange.Redis;

var tokenSource = new CancellationTokenSource();
var token = tokenSource.Token;

var muxer = ConnectionMultiplexer.Connect("localhost:6379");
var db = muxer.GetDatabase();

const string streamName = "positions";
const string groupName = "avg";

if (!await db.KeyExistsAsync(streamName) ||
    (await db.StreamGroupInfoAsync(streamName)).All(x => x.Name != groupName))
{
    await db.StreamCreateConsumerGroupAsync(streamName, groupName, "0-0");
}

var consumerGroupReadTask = Task.Run(async () =>
{
    var id = string.Empty;
    while (!token.IsCancellationRequested)
    {
        if (!string.IsNullOrEmpty(id))
        {
            await db.StreamAcknowledgeAsync(streamName, groupName, id);
            id = string.Empty;
        }

        var result = await db.StreamReadGroupAsync(streamName, groupName, "avg-1", ">", 1);
        if (result.Any())
        {
            id = result.First().Id;
            var dict = ParseResult(result.First());

            var trade = JsonSerializer.Deserialize<Trade>(dict["trade"]);

            Console.WriteLine($"Group read result: trade: {dict["trade"]}, time: {dict["time"]}");
        }

        await Task.Delay(1000);
    }
});

Console.ReadLine();

static Dictionary<string, string> ParseResult(StreamEntry entry)
{
    return entry.Values.ToDictionary(x => x.Name.ToString(), x => x.Value.ToString());
}

生产者

using System.Text.Json;
using Shared;
using StackExchange.Redis;

var tokenSource = new CancellationTokenSource();
var token = tokenSource.Token;

var muxer = ConnectionMultiplexer.Connect("localhost:6379");
var db = muxer.GetDatabase();

const string streamName = "positions";

var producerTask = Task.Run(async () =>
{
    var random = new Random();
    while (!token.IsCancellationRequested)
    {
        var trade = new Trade(random.Next(1, 3), "btcusdt", 25000, 2);

        var entry = new List<NameValueEntry>
        {
            new("trade", JsonSerializer.Serialize(trade)),
            new("time", DateTimeOffset.Now.ToUnixTimeSeconds())
        };

        await db.StreamAddAsync(streamName, entry.ToArray());

        await Task.Delay(2000);
    }
});

Console.ReadLine();

Answer 1

您可以在这里使用一些策略，这取决于所需的分发级别和处理来自流的丢失消息的程度。下面是使用Redis的几个可行的解决方案：

当您可以容忍事件中1%的失分时，使用Bloom过滤器。

您可以在Redis中使用BloomFilter，这将是一种非常紧凑、非常快速的方法来确定是否还没有记录特定的记录。如果你跑：

var hasBeenAdded = ((int)await db.ExecuteAsync("BF.ADD", "bf:trades",dict["trade"])) == 1;

如果hasBeenAdded为真，您可以明确地说，记录不是重复的，如果它是假的，则取决于您如何使用BF.RESERVE设置布卢姆过滤器

如果您想使用Bloom过滤器，则需要将RedisBloom加载到您的Redis实例中，或者只需使用红宝石堆。

在错误不可接受时使用排序集

如果你的应用程序不能容忍失误，你可能会更明智地使用一个集合或一个排序集，一般来说，我建议你使用一个集合，因为它们更容易清理。

基本上，如果您使用的是排序集，您将使用ZSCORE zset:trades trade-id检查记录是否已经记录在平均值中，如果分数回来，您知道记录已经被使用过，否则可以将其添加到排序集中。重要的是，由于排序集线性增长，所以希望定期清理它，因此，如果将消息id设置为得分的时间戳，则可能会确定返回的可行间隔，并执行ZREMRANGEBYSCORE清除旧记录。