本地运行Azure Service Fabric有状态TCP服务时安装报错但可正常运行求助

问题描述 #

基于Azure Service Fabric的StatefulService构建TCP服务器，本地运行时出现部署状态异常：

• Visual Studio提示：

fabric:/TCPTestServerFabric/TCPTestService is not ready, 1 partitions remaining.

• Service Fabric Explorer显示：

Partition is below target replica or instance count.
fabric:/TCPTestServerFabric/TCPTestService 1 1 61eee22a-c5d4-46e8-ab57-8d41009abc3d
  N/P InBuild _Node_0 133444395888851522

实际服务可正常接收TCP客户端连接、收发消息，移除OpenAsync代码后仍报同样错误。已排查磁盘空间（剩余250GB）和Service Fabric配置选项，均未解决。

问题根源 #

核心问题出在**ICommunicationListener的OpenAsync方法实现不符合Service Fabric生命周期规范**：

• OpenAsync的设计目的是完成通信监听初始化，返回服务可访问地址后立即结束，而非长期阻塞在连接处理循环中。你的代码中OpenAsync包含无限循环和连接处理逻辑，导致Service Fabric一直认为监听未就绪，标记服务为InBuild状态。
• CloseAsync方法直接返回Task.FromCanceled(cancellationToken)，未正确释放Socket资源，可能引发资源泄漏。

解决方案 #

1. 拆分逻辑：OpenAsync只负责初始化Socket监听并返回地址，将连接处理逻辑放到后台独立任务中执行。
2. 正确实现资源释放：在CloseAsync和Abort中确保Socket资源被关闭、释放。
3. 保持服务生命周期正常：RunAsync方法仅处理状态管理逻辑（或空实现），不阻塞服务运行。

修改后的代码 #

TCPTestServer 实现 #

public class TCPTestServer : ICommunicationListener
{
    private readonly StatefulServiceContext _context;
    private readonly IReliableStateManager _stateManager;
    private Socket _listenerSocket;
    private CancellationTokenSource _cts;

    public TCPTestServer(StatefulServiceContext context, IReliableStateManager stateManager)
    {
        _context = context;
        _stateManager = stateManager;
        _cts = new CancellationTokenSource();
    }

    public void Abort()
    {
        _cts.Cancel();
        _listenerSocket?.Close();
        ServiceEventSource.Current.ServiceMessage(_context, $"Request aborted");
    }

    public async Task CloseAsync(CancellationToken cancellationToken)
    {
        _cts.Cancel();
        if (_listenerSocket != null)
        {
            _listenerSocket.Close();
            _listenerSocket.Dispose();
        }
        ServiceEventSource.Current.ServiceMessage(_context, $"Server closed");
        await Task.CompletedTask;
    }

    public async Task<string> OpenAsync(CancellationToken cancellationToken)
    {
        EndpointResourceDescription serviceEndpoint = _context.CodePackageActivationContext.GetEndpoint("ServiceEndpoint");
        int port = serviceEndpoint.Port;

        IPEndPoint ipEndPoint = new IPEndPoint(IPAddress.Parse("127.0.0.1"), port);
        _listenerSocket = new Socket(ipEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);

        _listenerSocket.Bind(ipEndPoint);
        _listenerSocket.Listen();
        ServiceEventSource.Current.ServiceMessage(_context, $"Server started on port {port}");

        // 启动后台任务处理连接，不阻塞OpenAsync
        _ = HandleIncomingConnectionsAsync(_cts.Token);

        // 返回服务地址，告知Service Fabric监听已就绪
        return $"tcp://127.0.0.1:{port}";
    }

    private async Task HandleIncomingConnectionsAsync(CancellationToken cancellationToken)
    {
        while (!cancellationToken.IsCancellationRequested)
        {
            try
            {
                Socket handler = await _listenerSocket.AcceptAsync(cancellationToken);
                ServiceEventSource.Current.ServiceMessage(_context, $"Incoming connection: {handler.RemoteEndPoint}");

                // 每个连接启动独立任务处理，避免阻塞其他连接
                _ = HandleClientCommunicationAsync(handler, cancellationToken);
            }
            catch (OperationCanceledException)
            {
                ServiceEventSource.Current.ServiceMessage(_context, $"Connection handling canceled");
                break;
            }
            catch (Exception ex)
            {
                ServiceEventSource.Current.ServiceMessage(_context, $"Error accepting connection: {ex.Message}");
            }
        }
    }

    private async Task HandleClientCommunicationAsync(Socket handler, CancellationToken cancellationToken)
    {
        try
        {
            while (!cancellationToken.IsCancellationRequested)
            {
                var buffer = new byte[1_024];
                var received = await handler.ReceiveAsync(buffer, SocketFlags.None, cancellationToken);
                if (received == 0)
                {
                    ServiceEventSource.Current.ServiceMessage(_context, $"Disconnected: {handler.RemoteEndPoint}");
                    break;
                }

                var response = Encoding.UTF8.GetString(buffer, 0, received);
                ServiceEventSource.Current.ServiceMessage(_context, $"Incoming message from {handler.RemoteEndPoint}: {response}");

                var ackMessage = "<ACK/>";
                var echoBytes = Encoding.UTF8.GetBytes(ackMessage);
                int bytesSent = await handler.SendAsync(echoBytes, 0, cancellationToken);

                ServiceEventSource.Current.ServiceMessage(_context, $"Sent ACK to {handler.RemoteEndPoint}: {ackMessage}");
            }
        }
        catch (OperationCanceledException)
        {
            ServiceEventSource.Current.ServiceMessage(_context, $"Client communication canceled: {handler.RemoteEndPoint}");
        }
        catch (Exception ex)
        {
            ServiceEventSource.Current.ServiceMessage(_context, $"Error handling client {handler.RemoteEndPoint}: {ex.Message}");
        }
        finally
        {
            handler.Close();
            handler.Dispose();
        }
    }
}

TCPTestService 实现（无需修改） #

internal sealed class TCPTestService : StatefulService
{
    public TCPTestService(StatefulServiceContext context) : base(context) { }

    protected override IEnumerable<ServiceReplicaListener> CreateServiceReplicaListeners()
    {
        return new[] { new ServiceReplicaListener(context => 
                       new TCPTestServer(context, StateManager), "ServiceEndpoint") };
    }

    protected override async Task RunAsync(CancellationToken cancellationToken)
    {
        // 可添加状态管理相关逻辑，空实现也不影响服务运行
        await Task.CompletedTask;
    }
}

内容的提问来源于stack exchange，提问作者Inoel