multithreading - QueueUtil 为 MessageHandler 生成太多线程，异步/等待问题

Question

更新:错误在于使用 Action<QueueMessage> onSuccess对于实际任务并在没有 await 的情况下调用它。我们实际上正在传递一个 async回调作为 onSuccess ，因此它不会阻塞，它只是被解雇而不等待它。将类型更改为 Func<QueueMessage, Task>并通过 await onSuccess(messageObj) 调用修复了该行为。

我正在接收消息，并且每条消息都由涉及 IO 和 CPU 密集型例程的长时间运行作业处理。我的问题是我注册的消息处理程序被调用的次数太多(基本上与消息的数量一样多)，这使得我的进程占用了 100% 的 CPU 资源，并且一切都严重减慢。

我有一个 QueueClient 的包装器，我想我因此遇到了麻烦。我知道我的代码是错误的，但我对 .NET 中的 async/await 如何工作缺乏足够的了解，无法自行修复它。

这是包装器代码，它获取 json 消息或指向 Azure 存储的链接，因此它处理此逻辑并将其传递给回调。

public static void RegisterReceiver(string queueName,
        Action<QueueMessage> onSuccess, Action<ExceptionReceivedEventArgs> onError)
        _queueClient.RegisterMessageHandler(
                async (message, cancellationToken) =>
                {
                    var messageObj = new QueueMessage(message);
                    var regex = new Regex(@"\[blob%([^\]]+)\](.+)");
                    if (regex.IsMatch(messageObj.Body))
                    {
                        // Pulling blob data from storage
                        var match = regex.Match(messageObj.Body);
                        messageObj.ContentType = match.Groups[1].Value;
                        messageObj.Body = await BlobUtil.GetAndDeleteBlobAsync(match.Groups[2].Value, settings.QueueBlobStorageName);
                    }
                    onSuccess(messageObj);
                }, 
                new MessageHandlerOptions((exceptionArgs) =>
                {
                    onError(exceptionArgs);
                    return Task.CompletedTask;
                })
                {
                    MaxConcurrentCalls = 1
                }
            );

这是带有最终回调的调用代码:

QueueUtil.RegisterReceiver(QueueUtil.QUEUE_NAME, 
                async message =>
                {
                    await processService.RunJobs(message);
                }, 
                exceptionArgs =>
                {
                    throw new Exception("Exception occurred while receiving message from " + QueueUtil.QUEUE_NAME, 
                        exceptionArgs.Exception);
                }
            );

因此，不知何故，有了所有这些异步/等待的事情，我的接收器就会在消息到达后立即生成线程。

Answer 1

错误在于使用 Action<QueueMessage> onSuccess对于实际任务并在没有 await 的情况下调用它。我们实际上正在传递一个 async回调作为 onSuccess ，因此它不会阻塞，它只是被解雇而不等待它。将类型更改为 Func<QueueMessage, Task>并通过 await onSuccess(messageObj) 调用修复了该行为。