重构NatsPublisher：IOptions依赖注入问题与优化建议

我为NATS Client开发了一款支持ASP.NET依赖注入（DI）的包装器，目前梳理出三处可优化内容，同时希望了解其他潜在优化方向：

一、已识别的优化项 #

1. 对_jetStream实现懒加载 #

当前NatsPublisher在构造时就初始化_jetStream，建议改为懒加载，仅在首次调用发布方法时才创建JetStream上下文，示例代码：

_jetStreamFactory = new Lazy<IJetStream>(() => connection.CreateJetStreamContext());

2. 调整NatsPublisher的依赖注入方式 #

本库采用装饰器模式扩展ConnectionFactory能力，但NatsPublisher直接依赖IOptions<ProducerConfiguration>不符合设计初衷，需要调整配置注入方式。

3. 寻求额外优化建议 #

希望了解代码中其他可优化的地方。

二、额外优化建议 #

1. 拆分配置职责 #

当前ProducerConfiguration同时包含NATS连接配置（Servers/Url/User/Password）和生产者专属配置（Stream/Subject），建议拆分为NatsConnectionOptions和NatsProducerOptions，职责更清晰，也避免配置混用。

2. 重构NatsPublisher的Stream初始化逻辑 #

当前在NatsPublisher构造函数中调用JetStreamUtils.CreateStreamOrUpdateSubjects，会导致DI容器初始化时就执行IO操作，建议将Stream初始化逻辑移到首次发布时执行，或者单独封装为初始化服务，在应用启动时按需执行。

3. 优化JetStreamUtils中的硬编码与异常处理 #

• CreateStream系列方法默认使用StorageType.Memory，建议设为可配置参数，增强灵活性；
• PublishInBackground方法中直接调用Environment.Exit(-1)过于粗暴，建议改为抛出自定义异常或通过日志记录错误，由上层决定处理方式；
• ReadMessagesAck方法返回IList<Msg>，可改为IReadOnlyList<Msg>，避免外部修改集合。

4. 增强DI注册的灵活性 #

AddNatsClient方法中直接注册单例的Options，建议新增重载方法AddNatsClient(IConfiguration configuration)，支持从配置绑定，简化使用示例中的重复配置读取逻辑。

5. 抽象JetStream创建逻辑 #

当前NatsPublisher直接通过connection.CreateJetStreamContext()创建JetStream，建议抽象出IJetStreamFactory接口，便于后续扩展或替换实现，同时符合依赖倒置原则。

6. 统一发布方法的实现 #

NatsPublisher中的多个Publish方法存在重复逻辑（创建Msg对象、调用_jetStream.PublishAsync），可提取公共方法减少重复代码，例如：

private async ValueTask<PublishAck> PublishInternal(byte[] payload, Action<Msg>? configureMsg = null)
{
    var msg = new Msg(_configuration.Subject, null, null, payload);
    configureMsg?.Invoke(msg);
    return await _jetStreamFactory.Value.PublishAsync(msg);
}

其他发布方法可直接调用该内部方法。

7. 配置验证 #

为ProducerConfiguration添加配置验证，比如在DI注册时使用AddOptions<ProducerConfiguration>().ValidateDataAnnotations()，或自定义验证逻辑，确保必填项（Stream/Subject）不为空。

三、相关代码 #

问题涉及的代码 #

public class ProducerConfiguration
{
    public string[]? Servers { get; init; }
    public string? Url { get; init; }
    public string? User { get; init; }
    public string? Password { get; init; }
    public required string Stream { get; init; }
    public required string Subject { get; init; }
}

public interface IPublisher
{
    ValueTask<PublishAck> Publish<T>(T payload) where T : class;

    ValueTask<PublishAck> PublishAsync(byte[] payload, IEnumerable<(string, string)> headers);

    ValueTask<PublishAck> PublishWithDeduplicationIdAsync(byte[] payload, string id);
}

public sealed class NatsPublisher : IPublisher
{
    private readonly ProducerConfiguration _configuration;
    private readonly IJetStream _jetStream;

    public NatsPublisher(IOptions<ProducerConfiguration> options, IConnection connection)
    {
        _configuration = options.Value;

        JetStreamUtils.CreateStreamOrUpdateSubjects(connection, _configuration.Stream, _configuration.Subject);
        _jetStream = connection.CreateJetStreamContext();
    }

    public async ValueTask<PublishAck> Publish<T>(T payload)
        where T : class
    {
        var data = JsonSerializer.SerializeToUtf8Bytes(payload);

        var msg = new Msg(_configuration.Subject, null, null, data);

        return await _jetStream.PublishAsync(msg);
    }

    public async ValueTask<PublishAck> PublishAsync(byte[] payload)
    {
        var msg = new Msg(_configuration.Subject, null, null, payload);
        return await _jetStream.PublishAsync(msg);
    }

    public async ValueTask<PublishAck> PublishAsync(byte[] payload, IEnumerable<(string, string)> headers)
    {
        var msg = new Msg(_configuration.Subject, null, null, payload);

        foreach (var (header, val) in headers)
        {
            msg.Header[header] = val;
        }

        return await _jetStream.PublishAsync(msg);
    }

    public async ValueTask<PublishAck> PublishWithDeduplicationIdAsync(byte[] payload, string id)
    {
        var msg = new Msg(_configuration.Subject, null, null, payload)
        {
            Header = { ["Nats-Msg-Id"] = id }
        };

        return await _jetStream.PublishAsync(msg);
    }
}

库本身代码 #

public static class ServiceCollectionExtensions
{
    public static IServiceCollection AddNatsClient(
        this IServiceCollection services,
        Action<Options>? configureOptions = null,
        ServiceLifetime connectionServiceLifeTime = ServiceLifetime.Transient)
    {
        ArgumentNullException.ThrowIfNull(services);

        var defaultOptions = ConnectionFactory.GetDefaultOptions();
        configureOptions?.Invoke(defaultOptions);
        services.AddSingleton(defaultOptions);

        services.AddSingleton<ConnectionFactory>();
        services.AddSingleton<INatsClientConnectionFactory, NatsClientConnectionFactoryDecorator>();

        services.TryAdd(new ServiceDescriptor(typeof(IConnection), provider =>
        {
            var options = provider.GetRequiredService<Options>();
            var connectionFactory = provider.GetRequiredService<INatsClientConnectionFactory>();
            return connectionFactory.CreateConnection(options);
        }, connectionServiceLifeTime));

        services.TryAdd(new ServiceDescriptor(typeof(IEncodedConnection), provider =>
        {
            var options = provider.GetRequiredService<Options>();
            var connectionFactory = provider.GetRequiredService<INatsClientConnectionFactory>();
            return connectionFactory.CreateEncodedConnection(options);
        }, connectionServiceLifeTime));

        return services;
    }
}

public interface INatsClientConnectionFactory
{
    IConnection CreateConnection(Action<Options>? configureOptions = null);

    IConnection CreateConnection(Options options);

    IEncodedConnection CreateEncodedConnection(Action<Options>? configureOptions = null);

    IEncodedConnection CreateEncodedConnection(Options options);
}

public sealed class NatsClientConnectionFactoryDecorator : INatsClientConnectionFactory
{
    private readonly ConnectionFactory _connectionFactory;

    public NatsClientConnectionFactoryDecorator(ConnectionFactory connectionFactory)
    {
        _connectionFactory = connectionFactory;
    }

    public IConnection CreateConnection(Action<Options>? configureOptions = default)
    {
        var options = ConnectionFactory.GetDefaultOptions();
        configureOptions?.Invoke(options);
        return CreateConnection(options);
    }

    public IConnection CreateConnection(Options options)
    {
        return _connectionFactory.CreateConnection(options);
    }

    public IEncodedConnection CreateEncodedConnection(Action<Options>? configureOptions = default)
    {
        var options = ConnectionFactory.GetDefaultOptions();
        configureOptions?.Invoke(options);
        return CreateEncodedConnection(options);
    }

    public IEncodedConnection CreateEncodedConnection(Options options)
    {
        return _connectionFactory.CreateEncodedConnection(options);
    }
}

public static class JetStreamUtils
{
    // ----------------------------------------------------------------------------------------------------
    // STREAM INFO / CREATE / UPDATE
    // ----------------------------------------------------------------------------------------------------
    public static StreamInfo? GetStreamInfoOrNullWhenNotExist(IJetStreamManagement jsm, string streamName)
    {
        try
        {
            return jsm.GetStreamInfo(streamName);
        }
        catch (NATSJetStreamException e)
        {
            if (e.ErrorCode == 404)
            {
                return null;
            }

            throw;
        }
    }

    public static bool StreamExists(IConnection c, string streamName)
    {
        return GetStreamInfoOrNullWhenNotExist(c.CreateJetStreamManagementContext(), streamName) != null;
    }

    public static bool StreamExists(IJetStreamManagement jsm, string streamName)
    {
        return GetStreamInfoOrNullWhenNotExist(jsm, streamName) != null;
    }

    public static void ExitIfStreamExists(IJetStreamManagement jsm, string streamName)
    {
        if (StreamExists(jsm, streamName))
        {
            Environment.Exit(-1);
        }
    }

    public static void ExitIfStreamNotExists(IConnection c, string streamName)
    {
        if (!StreamExists(c, streamName))
        {
            Environment.Exit(-1);
        }
    }

    public static StreamInfo CreateStream(IJetStreamManagement jsm, string streamName, StorageType storageType,
        params string[] subjects)
    {
        var sc = StreamConfiguration.Builder()
            .WithName(streamName)
            .WithStorageType(storageType)
            .WithSubjects(subjects)
            .Build();

        var si = jsm.AddStream(sc);

        return si;
    }

    public static StreamInfo CreateStream(IJetStreamManagement jsm, string stream, params string[] subjects)
    {
        return CreateStream(jsm, stream, StorageType.Memory, subjects);
    }

    public static StreamInfo CreateStream(IConnection c, string stream, params string[] subjects)
    {
        return CreateStream(c.CreateJetStreamManagementContext(), stream, StorageType.Memory, subjects);
    }

    public static StreamInfo CreateStreamExitWhenExists(IConnection c, string streamName, params string[] subjects)
    {
        return CreateStreamExitWhenExists(c.CreateJetStreamManagementContext(), streamName, subjects);
    }

    public static StreamInfo CreateStreamExitWhenExists(IJetStreamManagement jsm, string streamName,
        params string[] subjects)
    {
        ExitIfStreamExists(jsm, streamName);
        return CreateStream(jsm, streamName, StorageType.Memory, subjects);
    }

    public static void CreateStreamWhenDoesNotExist(IJetStreamManagement jsm, string stream, params string[] subjects)
    {
        try
        {
            jsm.GetStreamInfo(stream);
            return;
        }
        catch (NATSJetStreamException)
        {
        }

        var sc = StreamConfiguration.Builder()
            .WithName(stream)
            .WithStorageType(StorageType.Memory)
            .WithSubjects(subjects)
            .Build();
        jsm.AddStream(sc);
    }

    public static void CreateStreamWhenDoesNotExist(IConnection c, string stream, params string[] subjects)
    {
        CreateStreamWhenDoesNotExist(c.CreateJetStreamManagementContext(), stream, subjects);
    }

    public static StreamInfo CreateStreamOrUpdateSubjects(IJetStreamManagement jsm, string streamName,
        StorageType storageType, params string[] subjects)
    {
        var si = GetStreamInfoOrNullWhenNotExist(jsm, streamName);
        if (si == null)
        {
            return CreateStream(jsm, streamName, storageType, subjects);
        }

        var sc = si.Config;
        var needToUpdate = false;
        foreach (var sub in subjects)
        {
            if (!sc.Subjects.Contains(sub))
            {
                needToUpdate = true;
                sc.Subjects.Add(sub);
            }
        }

        if (needToUpdate)
        {
            si = jsm.UpdateStream(sc);
        }

        return si;
    }

    public static StreamInfo CreateStreamOrUpdateSubjects(IJetStreamManagement jsm, string streamName,
        params string[] subjects)
    {
        return CreateStreamOrUpdateSubjects(jsm, streamName, StorageType.Memory, subjects);
    }

    public static StreamInfo CreateStreamOrUpdateSubjects(IConnection c, string stream, params string[] subjects)
    {
        return CreateStreamOrUpdateSubjects(c.CreateJetStreamManagementContext(), stream, StorageType.Memory,
            subjects);
    }

    // ----------------------------------------------------------------------------------------------------
    // PUBLISH
    // ----------------------------------------------------------------------------------------------------
    public static void Publish(IConnection c, string subject, int count)
    {
        Publish(c.CreateJetStreamContext(), subject, "data", count);
    }

    public static void Publish(IJetStream js, string subject, int count)
    {
        Publish(js, subject, "data", count);
    }

    public static void Publish(IJetStream js, string subject, string prefix, int count)
    {
        for (var x = 1; x <= count; x++)
        {
            var data = prefix + x;
            js.Publish(subject, Encoding.UTF8.GetBytes(data));
        }
    }

    public static void PublishInBackground(IJetStream js, string subject, string prefix, int count)
    {
        new Thread(() =>
        {
            try
            {
                for (var x = 1; x <= count; x++)
                {
                    js.Publish(subject, Encoding.ASCII.GetBytes(prefix + "-" + x));
                }
            }
            catch (Exception)
            {
                Environment.Exit(-1);
            }
        }).Start();
        Thread.Sleep(100); // give the publish thread a little time to get going
    }

    // ----------------------------------------------------------------------------------------------------
    // READ MESSAGES
    // ----------------------------------------------------------------------------------------------------
    public static IList<Msg> ReadMessagesAck(ISyncSubscription sub, int timeout = 1000)
    {
        IList<Msg> messages = new List<Msg>();
        var keepGoing = true;
        while (keepGoing)
        {
            try
            {
                var msg = sub.NextMessage(timeout);
                messages.Add(msg);
                msg.Ack();
            }
            catch (NATSTimeoutException)
            {
                keepGoing = false;
            }
        }

        return messages;
    }
}

使用示例 #

builder.Services.Configure<NatsProducerConfiguration>(options =>
    builder.Configuration.GetSection("Nats").Bind(options));
var natsConfiguration = builder.Configuration.GetSection("Nats").Get<NatsProducerConfiguration>();

builder.Services.AddNatsClient(options =>
{
    options.Servers = natsConfiguration?.Servers;
    options.Url = natsConfiguration?.Url;
    options.User = natsConfiguration?.User;
    options.Password = natsConfiguration?.Password;

    options.MaxReconnect = 5;
    options.ReconnectWait = 5000;
});

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