c# - 围绕同时处理单个和批量请求的架构

Question

我在 Windows 服务中托管了一个 WCF 服务。该服务公开了 2 个方法:

1. bool ProcessClaim(string options, ref string xml); 将一些数据作为输入，进行一些处理(包括 IO 绑定(bind)操作，如数据库查询)，然后返回结果。
2. void RunJob(string ticket); 立即返回。根据ticket，从存储(如DB或文件系统)读取输入数据，对每个数据元素做同样的处理，并将结果存回存储。批处理通常包含许多声明。

用户可以调用ProcessClaim来处理单个请求，调用RunJob来运行批处理。多个批处理可以同时运行。每个处理请求都包装为 Task，因此所有请求都是并行执行的。问题是不允许批处理通过调度大量请求来塞满处理队列。换句话说，如果用户执行大批量，它会在很长一段时间内阻塞小批量和单个处理请求。所以我想出了以下架构，Albahari 对此进行了很好的描述(非常简短):

public sealed class ProcessingQueue : IDisposable
{
    private class WorkItem
    {
        public readonly TaskCompletionSource<string> TaskSource;
        public readonly string Options;
        public readonly string Claim;
        public readonly CancellationToken? CancelToken;

        public WorkItem(
            TaskCompletionSource<string> taskSource,
            string options,
            string claim,
            CancellationToken? cancelToken)
        {
            TaskSource = taskSource;
            Options = options;
            Claim = claim;
            CancelToken = cancelToken;
        }
    }

    public ProcessingQueue()
        : this(Environment.ProcessorCount)
    {
    }

    public ProcessingQueue(int workerCount)
    {
        _taskQ = new BlockingCollection<WorkItem>(workerCount * 2);

        for (var i = 0; i < workerCount; i++)
            Task.Factory.StartNew(Consume);
    }

    public void Dispose()
    {
        _taskQ.CompleteAdding();
    }

    private readonly BlockingCollection<WorkItem> _taskQ;

    public Task<string> EnqueueTask(string options, string claim, CancellationToken? cancelToken = null)
    {
        var tcs = new TaskCompletionSource<string>();
        _taskQ.Add(new WorkItem(tcs, options, claim, cancelToken));
        return tcs.Task;
    }

    public static Task<string> ProcessRequest(string options, string claim, CancellationToken? cancelToken = null)
    {
        return Task<string>.Factory.StartNew(() => ProcessItem(options, claim));
    }

    private void Consume()
    {
        foreach (var workItem in _taskQ.GetConsumingEnumerable())
        {
            if (workItem.CancelToken.HasValue && workItem.CancelToken.Value.IsCancellationRequested)
                workItem.TaskSource.SetCanceled();
            else
            {
                try
                {
                    workItem.TaskSource.SetResult(ProcessItem(workItem.Options, workItem.Claim));
                }
                catch (Exception ex)
                {
                    workItem.TaskSource.SetException(ex);
                }
            }
        }
    }

    private static string ProcessItem(string options, string claim)
    {
        // do some actual work here
        Thread.Sleep(2000); // simulate work;
        return options + claim; // return final result
    }
}

静态方法 ProcessRequest 可用于处理单个请求，而实例方法 EnqueueTask - 用于批处理。当然，所有批处理都必须使用 ProcessingQueue 的单个共享实例。尽管这种方法效果很好并且可以控制同时运行的多个批处理的速度，但我觉得有些地方不对:

• 必须手动维护一个工作线程池
• 很难猜测最佳工作线程数(我默认使用处理器核心数)
• 当没有批处理在运行时，线程束仍然处于阻塞状态，浪费系统资源
• 处理 block 工作线程的 IO 绑定(bind)部分降低了 CPU 使用效率

我想知道，有没有更好的方法来处理这种情况？

更新:其中一项要求是为批处理提供全部功能，这意味着当用户执行一个批处理并且没有其他传入请求时，所有资源都必须专用于处理该批处理。

Answer 1

我会说，使用单一服务接口(interface)和单一托管容器来处理这两种截然不同的需求可能是错误的。

您应该将您的服务分离为两个 - 一个按需返回对单个请求的响应，另一个排队批量查询并在单个线程上处理它们。

通过这种方式，您可以为实时消费者提供高可用性 channel ，并为批量消费者提供离线 channel 。这些可以作为单独的关注点进行部署和管理，允许您在每个服务接口(interface)上提供不同的服务级别。

只是我对提议的架构的想法。

更新

事实是你的批量处理 channel 是线下 channel 。这种方式意味着消费者将不得不排队等待，并且等待他们的请求返回的时间不确定。

那么作业队列呢？每个作业在处理时都会获得所有可用资源。处理作业后，调用者会收到作业已完成的通知。