.NET Core中单例已关闭的Stan(NATS Streaming)连接如何续期重连？

你遇到的15秒阈值问题是STAN默认的客户端心跳超时时间导致的：断开时间短于15秒时NATS重连后STAN连接的心跳还未被服务器判定为失效，所以可以正常使用；超过15秒服务器已经将该STAN客户端下线，旧STAN连接会被永久标记为关闭，无法复用。同时你原有代码存在三个核心错误：

1. 事件注册顺序错误：先创建STAN连接，再配置NATS事件、赋值STAN选项的NatsConn，前面创建的连接根本没有用到后续的配置
2. DI注册逻辑无效：在重连事件中新增的IStanConnection单例不会替换已经注入到其他业务服务中的旧连接实例，相当于无效注册
3. STAN连接本身不支持底层NATS连接替换：0.3.0版本的IStanConnection一旦关闭就无法恢复，必须重建完整的STAN连接

通过连接包装器动态管理有效STAN连接，不要直接注册IStanConnection单例。

第一步：实现连接提供器包装类 #

public interface IStanConnectionProvider
{
    IStanConnection GetCurrentConnection();
}

public class StanConnectionProvider : IStanConnectionProvider, IDisposable
{
    private readonly StanSettings _settings;
    private IStanConnection _currentConnection;
    private readonly object _lock = new();
    private bool _disposed;

    public StanConnectionProvider(StanSettings settings)
    {
        _settings = settings;
        _currentConnection = CreateStanConnection();
    }

    private IStanConnection CreateStanConnection()
    {
        var natsOptions = ConnectionFactory.GetDefaultOptions();
        natsOptions.Servers = _settings.NatsServers.ToArray();
        natsOptions.AllowReconnect = true;
        natsOptions.MaxReconnect = Options.ReconnectForever;
        natsOptions.PingInterval = 2000;
        natsOptions.MaxPingsOut = 2;
        natsOptions.ReconnectWait = 1000;
        natsOptions.Timeout = 4000;

        // NATS重连时重建STAN连接
        natsOptions.ReconnectedEventHandler += (sender, args) =>
        {
            lock (_lock)
            {
                if (_disposed) return;
                // 释放旧STAN连接
                _currentConnection?.Dispose();
                // 用新NATS连接创建STAN连接
                var stanOptions = StanOptions.GetDefaultOptions();
                stanOptions.NatsConn = args.Conn;
                var newClientId = $"{_settings.ClientId}-{Guid.NewGuid()}";
                _currentConnection = new StanConnectionFactory().CreateConnection(_settings.ClusterId, newClientId, stanOptions);
                Console.WriteLine($"STAN连接重建完成，当前NATS连接ID：{args.Conn.ConnectedId}");
            }
        };

        var stanOpts = StanOptions.GetDefaultOptions();
        stanOpts.NatsConn = new ConnectionFactory().CreateConnection(natsOptions);
        // STAN连接丢失时主动重建
        stanOpts.ConnectionLostEventHandler += (sender, args) =>
        {
            lock (_lock)
            {
                if (_disposed) return;
                Console.WriteLine($"STAN连接丢失：{args.ConnectionException.Message}，开始重建");
                _currentConnection?.Dispose();
                // 创建全新的NATS+STAN连接
                var newNatsConn = new ConnectionFactory().CreateConnection(natsOptions);
                stanOpts.NatsConn = newNatsConn;
                var newClientId = $"{_settings.ClientId}-{Guid.NewGuid()}";
                _currentConnection = new StanConnectionFactory().CreateConnection(_settings.ClusterId, newClientId, stanOpts);
            }
        };

        var initClientId = $"{_settings.ClientId}-{Guid.NewGuid()}";
        return new StanConnectionFactory().CreateConnection(_settings.ClusterId, initClientId, stanOpts);
    }

    public IStanConnection GetCurrentConnection()
    {
        lock (_lock)
        {
            if (_disposed) throw new ObjectDisposedException(nameof(StanConnectionProvider));
            return _currentConnection;
        }
    }

    public void Dispose()
    {
        lock (_lock)
        {
            if (_disposed) return;
            _currentConnection?.Dispose();
            _disposed = true;
        }
    }
}

第二步：修改DI注册逻辑 #

public static IServiceCollection AddStanClient(this IServiceCollection services, StanSettings settings)
{
    services.AddSingleton(settings);
    services.AddSingleton<IStanConnectionProvider, StanConnectionProvider>();
    return services;
}

第三步：业务代码使用方式 #

不要直接注入IStanConnection，改为注入IStanConnectionProvider，每次发消息时获取当前有效的连接：

public class StanMessageService
{
    private readonly IStanConnectionProvider _connProvider;

    public StanMessageService(IStanConnectionProvider connProvider)
    {
        _connProvider = connProvider;
    }

    public async Task SendMessageAsync(string subject, byte[] payload)
    {
        var validConn = _connProvider.GetCurrentConnection();
        await validConn.PublishAsync(subject, payload);
    }
}

• 每次重建STAN连接必须生成新的客户端ID，复用旧ID会被STAN服务器判定为重复客户端拒绝连接
• 如果是消费者场景，重建STAN连接后需要重新注册主题订阅，旧订阅会随着旧STAN连接失效
• 加锁逻辑是为了避免多线程同时触发重建导致的资源冲突

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