主页 / 问题 / 1572534
Accepted
CrazyElf
Asked:2024-03-21 04:29:46 +0000 UTC2024-03-21 04:29:46 +0000 UTC

如何“折叠”调用以使被调用的代码不会“阻塞”？

  • 772

也许有一些现成的代码，以免重新发明轮子。

场景是这样的。该代码生成一些更新，然后通过某种方法进行处理。更新可能生成得太频繁，加上处理方法可能会变慢，结果更新没有时间被处理，它们会累积起来，然后处理器会做不必要的工作。我想在处理程序忙碌时以某种方式“累积”更新。而且，更新数据本身就已经在“累积”了，你不需要担心这个，你只需要把请求的“累积”解析到handler上，这样handler空闲的时候就不会处理所有的了单独累积的更新请求，但立即处理整个“更新”包。

像现在：

async Task UpdateFoo()
{
    ...
    await SaveAsync();
}

async Task UpdateBar()
{
    ...
    await SaveAsync();
}

async Task UpdateBaz()
{
    ...
    await SaveAsync();
}

async Task SaveAsync()
{
    // здесь нужно сделать так, 
    // чтобы одновременно обрабатывался только один запрос
    // по окончании которого проверялось бы не было ли ещё запросов
    // и если были, то обновление запускалось бы ещё один раз (сразу за всё "накопленное")
    // и так пока есть обновления по окончании очередной обработки
}

当然逻辑应该比较简单，脚本也应该很典型，但我不知道如何正常编写。为了登录一次 - 很明显SemaphoreSlim（带有WaitAsync）。但是完成剩下的事情最好的方法是什么？

c#

3 个回答

  1. Best Answer

    让我们举一个一般性的例子System.Threading.Channels.Channel<T>。

    private readonly Channel<T> _channel = Channel<T>.CreateUnbounded();

    编写一个 raking 方法

    private Task _workerTask;

private async Task WorkerAsync()
{
    ChannelReader<T> reader = _channel.Reader;
    List<T> list = new();

    // ждём здесь, если в канале пусто
    while (await reader.WaitToReadAsync())
    {
        // забираем всё что есть
        // ну или можно счётчиком ограничить максимальное количество выгребаемых данных за раз
        while (reader.TryRead(out T data))
        {
            list.Add(data);
        }
        // пачка собрана, погнали. это можно в try-catch завернуть, чтобы воркер не падал
        await UseAsync(list);
        list.Clear();
    }
}

    让我们启动工人

    _workerTask = WorkerAsync();

    把它扔进运河里

    _channel.Writer.TryWrite(x);

    要（永久）关闭通道并使该方法WorkerAsync完成，您需要调用

    _channel.Writer.Complete();
await _workerTask;

    这就是全部的魔力。

    这是基于您自己回答的代码，逻辑完全相同

    private readonly Channel<T> _channel = Channel<T>.CreateBounded(1); // максимум один в очереди

private async Task WorkerAsync()
{
    await foreach (var data in _channel.Reader.ReadAllAsync())
    {
        try
        {
            await _producer.SendAsync(data);
        }
        catch (Exception ex)
        {
            _logger.Error(ex, "Ошибка отсылки обновлений");
        }
    } 
}

/// <summary>
/// Отсылка с ограничениями: один поток отсылает, ещё один может ждать в очереди, остальные просто возвращаются
/// </summary>
/// <returns></returns>
private void SendUpdatedData()
{
    if (!channel.Writer.TryWrite(GetData())
    {
        _logger.Debug("Уже есть ожидание отсылки обновления...");
    }
}

    在这里，您只能在应用程序的整个持续时间内启动消费者一次，如何停止它如上所示 -_channel.Writer.Complete()

    Task task = WorkerAsync(); // ну или `_ = ...`

    这个任务的意义在于，如果你的应用程序结束了，你可以例如调用它Complete()并await task;等待直到所有请求都被worker处理完，然后正常关闭。但您提出的解决方案并没有提供这种可能性。

    • 2

  2. 如果所有 SaveAsync 调用都来自一个线程，那么您可以这样实现：

    1. 将更改排队

    2. 如果保护工作已经在进行中，请等待其完成

    3. 如果队列不为空

      1. 我们夺走了她的一切
      2. 将您的任务保存为待完成

      否则

      1. 等待保存完成

    如果调用可以是多线程的，那么您在放置锁时需要非常小心，以便它们覆盖所有操作，但不包括等待时间。

    像这样的东西（但我不保证任何事情，一般来说，我就写在这里）：

    object lobj = new();
List<...> changes = new();
Task active = Task.FromResult();

async Task SaveAsync()
{
  Task cur;

  lock (lobj)
  {
    changes.Add(...);
    cur = active;
  }

  await cur;

  lock (lobj)
  {
    if (changes.Count !== 0)
    {
      active = cur = SaveAsyncReal(changes.ToList());
      changes.Clear();
    }
    else
      cur = active;
  }

  await cur;
}

async Task SaveAsyncReal(List<...> diff)
{
  ...
}

    但如果这里还需要添加错误处理的话，那你还真需要考虑很久了……

    • 2

  3. 到目前为止，我已经使用一个任务和两个信号量自己完成了，尽管其他解决方案也很有趣，但不太适合我的任务。关键是一个更新正在工作，另一个更新正在队列中等待，所有其他更新都被丢弃 - 它们是不必要的，第二个挂起的更新就足够了。

    private void SendUpdatedData()
{
    _ = Task.Run(async () => { await SendAsync(); });
}

/// <summary>
/// Отсылка с ограничениями: один поток отсылает, ещё один может ждать в очереди, остальные просто возвращаются
/// </summary>
/// <returns></returns>
private async Task SendAsync()
{ 
    if (_semaphore2.CurrentCount == 0)
    {
        _logger.Debug("Уже есть ожидание отсылки обновления...");
        return;
    }
    await _semaphore2.WaitAsync();
    try
    {
        await _semaphore1.WaitAsync();
        try
        {
            var data = GetData();
            await _producer.SendAsync(data);
        }
        finally
        {
            _semaphore1.Release();
        }
    }
    catch (Exception ex)
    {
        _logger.Error(ex, "Ошибка отсылки обновлений");
    }
    finally
    {
        _semaphore2.Release();
    }
}

    请大家批评指正。）

    • 0

相关问题