为什么 ICommunicationListener.Abort() 不断被调用？答案

【问题标题】：Why is ICommunicationListener.Abort() continuously invoked?为什么 ICommunicationListener.Abort() 不断被调用？
【发布时间】：2017-11-17 03:19:33
【问题描述】：

以下是在本地 SF 集群中运行并侦听服务总线队列的无状态服务侦听器代码。

我想要的只是在一个永远在线的 SF 服务中持续监听队列命令。欢迎提供代码改进提示！

问题 #1

Abort() 被连续调用，有效地关闭了我的连接。是什么导致了这种行为，我该如何解决？在我的理解中，Abort() 应该只在未处理的异常或强制关闭的情况下被调用，我都不知道。

奖金问题

假设我们从 Abort() 中注释掉 CloseClient 调用，这样可以正确处理我们的队列。在第一条消息之后，CancellationToken 的 WaitHandle 被标记为已释放，并将其传递给我的回调会引发异常。这是什么原因造成的？

感谢您的帮助！

using System;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.ServiceBus.Messaging;
using Microsoft.ServiceFabric.Services.Communication.Runtime;

namespace Common
{
    public class QueueListener : ICommunicationListener
    {
        private readonly string _connectionString;
        private readonly string _path;
        private readonly Action<BrokeredMessage> _callback;

        private QueueClient _client;

        public QueueListener(string connectionString, string path, Action<BrokeredMessage> callback)
        {
            // Set field values
            _connectionString = connectionString;
            _path = path;

            // Save callback action
            _callback = callback;
        }

        public Task<string> OpenAsync(CancellationToken cancellationToken)
        {
            // Connect to subscription
            _client = QueueClient.CreateFromConnectionString(_connectionString, _path);

            // Configure the callback options
            var options = new OnMessageOptions
            {
                AutoComplete = false
            };

            // Catch and throw exceptions
            options.ExceptionReceived += (sender, args) => throw args.Exception;

            // Wire callback on message receipt
            _client.OnMessageAsync(message =>
            {
                return Task.Run(() => _callback(message), cancellationToken)
                    .ContinueWith(task =>
                    {
                        if (task.Status == TaskStatus.RanToCompletion)
                            message.CompleteAsync();
                        else
                            message.AbandonAsync();
                    }, cancellationToken);
            }, options);

            return Task.FromResult(_client.Path);
        }

        public Task CloseAsync(CancellationToken cancellationToken)
        {
            CloseClient();
            return Task.FromResult(_client.Path);
        }

        public void Abort()
        {
            CloseClient();
        }

        private void CloseClient()
        {
            // Make sure client is still open
            if (_client == null || _client.IsClosed)
                return;

            // Close connection
            _client.Close();
            _client = null;
        }
    }
}

【问题讨论】：

标签： c# azure azure-service-fabric azureservicebus azure-servicebus-queues

【解决方案1】：

传递给OpenAsync 的取消令牌不打算在方法范围之外使用，就像现在一样。您将其传递给OnMessage。 OnMessage 将在调用 OpenAsync 后调用，但令牌在 OpenAsyn 完成后被释放。

创建一个新的CancellationToken，或者使用这个nuget包应该对here有帮助。

【讨论】：

谢谢！完全有道理。对主要问题有什么想法吗？
我怀疑 OnMessage 会因为释放的令牌而产生异常，这会导致事件 ExceptionReceived 触发，从而导致未处理的异常，从而触发 Abort。但是您需要对其进行调试以确保。
不是这个，但我想通了。再次感谢！

【解决方案2】：

问题在于创建 ICommunicationListener 以与服务总线接口背后的整个概念；听众什么都没听！

将 Service Bus 连接重构为 StatelessService 实现中的默认 RunAsync() 方法修复了问题并且允许按预期监控取消令牌。

【讨论】：