如何在azure服务总线主题接收器中处理取消令牌？

Question

我的问题是如何处理RegisterMessageHandler方法中的取消令牌，每当收到新消息时，该方法就会被调用？此外，我还想“优雅地”处理订阅客户端的关闭，也就是说，如果一条消息已经被处理，那么我希望该消息得到完全处理，然后我想关闭连接。下面是我正在使用的代码。

目前我们正按照以下方法进行：1。使用信号量锁锁定消息的进程，并在finally块中释放锁。2.调用CancellationToken.Register方法处理取消令牌。释放Register方法中的锁。

public class AzureServiceBusReceiver
{
  private SubscriptionClient subscriptionClient;
  private static Semaphore semaphoreLock;

public AzureServiceBusReceiver(ServiceBusReceiverSettings settings)
{
    semaphoreLock = new Semaphore(1, 1);
    subscriptionClient = new SubscriptionClient(
        settings.ConnectionString, settings.TopicName, settings.SubscriptionName, ReceiveMode.PeekLock);
}

public void Receive(
    CancellationToken cancellationToken)
{
    var options = new MessageHandlerOptions(e =>
    {

        return Task.CompletedTask;
    })
    {
        AutoComplete = false,

    };

    subscriptionClient.RegisterMessageHandler(
        async (message, token) =>
        {
            semaphoreLock.WaitOne();
            if (subscriptionClient.IsClosedOrClosing)
                return;
            CancellationToken combinedToken = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, token).Token;
            try
            {
                // message processing logic
            }
            catch (Exception ex)
            {
                await subscriptionClient.DeadLetterAsync(message.SystemProperties.LockToken);
            }
            finally
            {
                semaphoreLock.Release();
            }
        }, options);


    cancellationToken.Register(() =>
    {
        semaphoreLock.WaitOne();
        if (!subscriptionClient.IsClosedOrClosing)
            subscriptionClient.CloseAsync().GetAwaiter().GetResult();
        semaphoreLock.Release();
        return;
    });
} 
}

Answer 1

将消息客户端实现为ICommunicationListener，因此当服务关闭时，您可以阻止调用，直到消息处理完成。不要使用静态信号量，这样您可以在项目中安全地重用代码。

下面是一个示例，说明如何做到这一点。

下面是由该代码创建的Nuget包。