多个字段的 C# 不可变计数器

Question

我在消息计数器上具有相当高的吞吐量（每秒数万），并且正在寻找一种有效的方法来获取计数，而无需在任何地方放置锁，或者理想情况下在我每次更新时不锁定每个消息计数10 秒。

不可变计数器对象的使用

我正在使用不可变的计数器类：

public class Counter
{
    public Counter(int quotes, int trades)
    {
        Quotes = quotes;
        Trades = trades;
    }

    readonly public int Quotes;
    readonly public int Trades;
    // and some other counter fields snipped
}

并且会在每个消息处理循环中更新它：

class MyProcessor
{
    System.Timers.Timer timer;
    Counter counter = new Counter(0,0);

    public MyProcessor()
    {
       // update ever 10 seconds
       this.timer = new System.Timers.Timer(10000);

       timer.Elapsed += (sender, e) => {
          var quotesPerSecond = this.counter.Quotes / 10.0;
          var tradesPerSecond = this.counter.Trades / 10.0;
          this.Counter = new Counter(0,0);
       });
    }

    public void ProcessMessages(Messages messages)
    {
       foreach(var message in messages) { /* */ }

       var oldCounter = counter;
       this.counter = new Counter(oldCounter.Quotes, oldCounter.Trades);   
    }
}

我有很多计数器（未全部显示），因此意味着在各个计数器字段上会有很多单独的 Interlocked.Increment 调用。

我能想到的唯一另一种方法是锁定 ProcessMessages 的每一次运行（这将是广泛的），并且对于实用程序而不是关键的程序崩溃的东西来说很重。

当我们只需要每 10 秒更新一次时，是否可以以这种方式使用不可变计数器对象而无需硬互锁/线程机制？

标志检查思路以避免锁定

计时器线程能否设置一个标志以供ProcessMessages 进行检查，如果它看到它已设置，则再次从零开始计数，即

/* snipped the MyProcessor class, same as before */

System.Timers.Timer timer;
Counter counter = new Counter(0,0);
ManualResetEvent reset = new ManualResetEvent(false);

public MyProcessor()
{
   // update ever 10 seconds
   this.timer = new System.Timers.Timer(10000);

   timer.Elapsed += (sender, e) => {
      var quotesPerSecond = this.counter.Quotes / 10.0;
      var tradesPerSecond = this.counter.Trades / 10.0;
      // log
      this.reset.Set();
   });
}

// this should be called every second with a heartbeat message posted to queue
public void ProcessMessages(Messages messages)
{
   if (reset.WaitOne(0) == true)
   {
      this.counter = new Counter(this.counter.Quotes, this.counter.Trades, this.counter.Aggregates);
      reset.Reset();
   }
   else
   {
      this.counter = new Counter(
                        this.counter.Quotes + message.Quotes.Count,
                        this.counter.Trades + message.Trades.Count);
   }
}

/* end of MyProcessor class */

这会起作用，但是当进程消息停止时更新“停止”（尽管吞吐量非常高，但它确实会在晚上暂停几个小时，理想情况下应该显示实际值而不是最后一个值）。

解决此问题的一种方法是每秒向MyProcessor.ProcessMessages() 发布一条心跳消息，以强制在设置reset ManualResetEvent 时对消息计数器进行内部更新并随后重置。

Answer 1

以下是您的Counter 类的三个新方法。一种用于从特定位置读取最新值，一种用于安全更新特定位置，另一种用于在现有位置的基础上轻松创建新的Counter：

public static Counter Read(ref Counter counter)
{
    return Interlocked.CompareExchange(ref counter, null, null);
}

public static void Update(ref Counter counter, Func<Counter, Counter> updateFactory)
{
    var counter1 = counter;
    while (true)
    {
        var newCounter = updateFactory(counter1);
        var counter2 = Interlocked.CompareExchange(ref counter, newCounter, counter1);
        if (counter2 == counter1) break;
        counter1 = counter2;
    }
}

public Counter Add(int quotesDelta, int tradesDelta)
{
    return new Counter(Quotes + quotesDelta, Trades + tradesDelta);
}

使用示例：

Counter latest = Counter.Read(ref this.counter);

Counter.Update(ref this.counter, existing => existing.Add(1, 1));

多个线程同时直接访问MyProcessor.counter 字段不是线程安全的，因为它既不受volatile 保护，也不受lock 保护。上述方法使用安全，因为它们是通过interlocked 操作访问字段。

Answer 2

我想用我的想法更新每个人，计数器更新是在线程本身内推送的。

一切都由DequeueThread 循环驱动，特别是this.queue.ReceiveAsync(TimeSpan.FromSeconds(UpdateFrequencySeconds)) 函数。

这将要么从队列中返回一个项目，处理它并更新计数器，要么超时然后更新计数器 - 没有其他线程参与所有事情，包括更新消息速率，都在线程内完成。

总而言之，没有什么是并行运行的（就数据包的出队而言），它一次获取一个项目并处理它，然后处理计数器。然后最后循环回来处理队列中的下一项。

这消除了同步的需要：

internal class Counter
{
    public Counter(Action<int,int,int,int> updateCallback, double updateEvery)
    {
        this.updateCallback = updateCallback;
        this.UpdateEvery = updateEvery;
    }

    public void Poll()
    {
        if (nextUpdate < DateTimeOffset.UtcNow)
        {
            // post the stats, and reset
            this.updateCallback(this.quotes, this.trades, this.aggregates, this.statuses);
            this.quotes = 0;
            this.trades = 0;
            this.aggregates = 0;
            this.statuses = 0;
            nextUpdate = DateTimeOffset.UtcNow.AddSeconds(this.UpdateEvery);
        }
    }

    public void AddQuotes(int count) => this.quotes += count;
    public void AddTrades(int count) => this.trades += count;
    public void AddAggregates(int count) => this.aggregates += count;
    public void AddStatuses(int count) => this.statuses += count;

    private int quotes;
    private int trades;
    private int aggregates;
    private int statuses;

    private readonly Action<int,int,int,int> updateCallback;
    public double UpdateEvery { get; private set; }
    private DateTimeOffset nextUpdate;
}

public class DeserializeWorker
{
    private readonly BufferBlock<byte[]> queue = new BufferBlock<byte[]>();
    private readonly IPolygonDeserializer polygonDeserializer;
    private readonly ILogger<DeserializeWorker> logger;

    private readonly Counter counter; 
    const double UpdateFrequencySeconds = 5.0;        
    long maxBacklog = 0;

    public DeserializeWorker(IPolygonDeserializer polygonDeserializer, ILogger<DeserializeWorker> logger)
    {
        this.polygonDeserializer = polygonDeserializer ?? throw new ArgumentNullException(nameof(polygonDeserializer));
        this.logger = logger;
        this.counter = new Counter(ProcesCounterUpdateCallback, UpdateFrequencySeconds);
    }

    public void Add(byte[] data)
    {
        this.queue.Post(data);
    }

    public Task Run(CancellationToken stoppingToken)
    {
        return Task
                .Factory
                .StartNew(
                    async () => await DequeueThread(stoppingToken),
                    stoppingToken,
                    TaskCreationOptions.LongRunning,
                    TaskScheduler.Default)
                .Unwrap();
    }

    private async Task DequeueThread(CancellationToken stoppingToken)
    {
        while (stoppingToken.IsCancellationRequested == false)
        {
            try
            {
                var item = await this.queue.ReceiveAsync(TimeSpan.FromSeconds(UpdateFrequencySeconds), stoppingToken);
                await ProcessAsync(item);
            }
            catch (TimeoutException)
            {
                // this is ok, timeout expired 
            }
            catch(TaskCanceledException)
            {
                break; // task cancelled, break from loop
            }
            catch (Exception e)
            {
                this.logger.LogError(e.ToString());
            }

            UpdateCounters();
        }

        await StopAsync();
    }


    protected async Task StopAsync()
    {
        this.queue.Complete();
        await this.queue.Completion;
    }

    protected void ProcessStatuses(IEnumerable<Status> statuses)
    {
        Parallel.ForEach(statuses, (current) =>
        {
            if (current.Result != "success")
                this.logger.LogInformation($"{current.Result}: {current.Message}");
        });
    }

    protected void ProcessMessages<T>(IEnumerable<T> messages)
    {
        Parallel.ForEach(messages, (current) =>
        {
            // serialize by type T
            // dispatch
        });
    }

    async Task ProcessAsync(byte[] item)
    {
        try
        {
            var memoryStream = new MemoryStream(item);
            var message = await this.polygonDeserializer.DeserializeAsync(memoryStream);

            var messagesTask = Task.Run(() => ProcessStatuses(message.Statuses));
            var quotesTask = Task.Run(() => ProcessMessages(message.Quotes));
            var tradesTask = Task.Run(() => ProcessMessages(message.Trades));
            var aggregatesTask = Task.Run(() => ProcessMessages(message.Aggregates));

            this.counter.AddStatuses(message.Statuses.Count);
            this.counter.AddQuotes(message.Quotes.Count);
            this.counter.AddTrades(message.Trades.Count);
            this.counter.AddAggregates(message.Aggregates.Count);

            Task.WaitAll(messagesTask, quotesTask, aggregatesTask, tradesTask);                                
        }
        catch (Exception e)
        {
            this.logger.LogError(e.ToString());
        }
    }

    void UpdateCounters()
    {
        var currentCount = this.queue.Count;
        if (currentCount > this.maxBacklog)
            this.maxBacklog = currentCount;

        this.counter.Poll();
    }

    void ProcesCounterUpdateCallback(int quotes, int trades, int aggregates, int statuses)
    {
        var updateFrequency = this.counter.UpdateEvery;
        logger.LogInformation(
            $"Queue current {this.queue.Count} (max {this.maxBacklog }), {quotes / updateFrequency} quotes/sec, {trades / updateFrequency} trades/sec, {aggregates / updateFrequency} aggregates/sec, {statuses / updateFrequency} status/sec");
    }
}