c# - ASB MessageReceiver ReceiveAsync 崩溃

Question

环境

1. Windows 10 专业版
2. .NET Core 控制台应用程序

代码

我有一个抽象的消息接收器，如下所示。在此代码中，实体是订阅的名称(例如用户)。

public class AzureMessageReceiver : ITdlMessageReceiver
{
    private readonly ServiceBusConnection serviceBusConnection;
    private readonly ILogger<AzureMessageReceiver> logger;

    public AzureMessageReceiver(ServiceBusConnection serviceBusConnection, ILogger<AzureMessageReceiver> logger)
    {
        this.serviceBusConnection = serviceBusConnection;
        this.logger = logger;
    }

    public async Task<TdlMessage<T>> ReceiveAsync<T>(string topic, string entity) where T : class
    {
        try
        {
            var subscriptionPath = EntityNameHelper.FormatSubscriptionPath(topic, entity);
            var messageReceiver = new MessageReceiver(serviceBusConnection, subscriptionPath, ReceiveMode.ReceiveAndDelete);
            var message = await messageReceiver.ReceiveAsync();

            if (message == null)
            {
                return null;
            }

            var messageString = Encoding.UTF8.GetString(message.Body);
            return JsonConvert.DeserializeObject<TdlMessage<T>>(messageString);
        }
        catch (Exception ex)
        {
            logger.LogError(ex, "Error receiving Azure message.");
            return null;
        }
    }
}

注入(inject)的ServiceBusConnection是这样构造的。 注意:相同的连接初始化可以将消息写入到相同的主题和订阅.

services.AddSingleton(serviceProvider =>
    new ServiceBusConnection(configuration[$"{DurableCommunicationKey}:AzureConnectionString"]));

更新:以下代码包装了对接收器类的调用，并且是接收消息的 Controller :

static async void Receive(ITdlMessageReceiver receiver, ILogger logger)
{
    while (true)
    {
        var message = await receiver.ReceiveAsync<TdlMessage<object>>(topic, entity);
        if (message != null)
        {
            logger.LogDebug($"Message received. Topic: {topic}. Action: {Enum.GetName(typeof(TopicActions), message.Action)}. Message: {JsonConvert.SerializeObject(message)}.");

        }

        Thread.Sleep(sleepTime);
    }
}

问题

每次执行此行 var message = wait messageReceiver.ReceiveAsync(); 时，它都会使控制台应用程序崩溃。没有异常，事件查看器中也没有任何内容。

我尝试过的

• 使用 ASB 中的辅助连接字符串
• 提供超时，例如 messageReceiver.ReceiveAsync(TimeSpan.FromMinutes(1));
• 将注入(inject)的topic从主题名称更改为主题的整个URL(例如https://{...}。 servicebus.windows.net/{topicName})
• 将 ReceiveMode 更改为 PeekLock
• 将 ConfigureAwait(false) 添加到 ReceiveAsync 调用中。
• 将超时更改为 TimeSpan.Zero。 注意:这不会使应用程序崩溃，但实际上会抛出一个被记录的异常。

Answer 1

async void 应转换为 async Task 并且您应该等待 Task.Delay 而不是调用 Thread . sleep 。如果要异步，您需要一直异步

static async Task Receive(ITdlMessageReceiver receiver, ILogger logger) {
    while (true) {
        var message = await receiver.ReceiveAsync<TdlMessage<object>>(topic, entity);
        if (message != null) {
            logger.LogDebug($"Message received. Topic: {topic}. Action: {Enum.GetName(typeof(TopicActions), message.Action)}. Message: {JsonConvert.SerializeObject(message)}.");    
        }    
        await Task.Delay(sleepTime);
    }
}

尝试使代码始终异步，是的，但作为控制台应用程序(单线程)，您将被允许在 Receive 上调用 Wait() Main 中的方法，因为它不会混合调用，从而导致异步流程出现问题。

public static void Main(string[] args) {

    //...
    //...
    //...


    Receive(receiver, logger).Wait();
}

引用Async/Await - Best Practices in Asynchronous Programming