Csharp go 重流记录

我正在使用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.StreamCreate消费者GroupAsync(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();