EasyNats.Core 1.0.1

EasyNats

EasyNats 旨在为 .NET 开发者提供一套透明、高性能、且对云原生环境（Native AOT）极度友好的高可用中间件矩阵。通过将 NATS 的高性能底层封装，开发者无需再部署 Redis、RabbitMQ 或复杂的服务网格。"在云原生时代，我们不应该为了简单的缓存和消息而维护一堆笨重的中间件。EasyNats 的目标是利用 NATS 强大的底层能力，为 .NET 开发者打造一个‘轻量、全能、极速’的微服务基座。"

✨ 核心特性

🚀 极致性能：基于 NATS .NET v2 官方驱动，采用 System.IO.Pipelines 和 StringValues 零拷贝技术。

📦 高效序列化：原生支持 MessagePack，相比 JSON 拥有更小的体积和更快的编解码速度。

🛡️ Native AOT 兼容：消除所有运行时反射警告，支持构建超小体积、秒级启动的二进制文件。

⚖️ 资源友好：针对低配置环境优化内存管理，支持拉取式消费者 (Pull Consumer) 的背压控制。

🛠️ 开箱即用：自动初始化 Stream、自动管理连接生命周期、简单的强类型 RPC 支持。

🛠️ 拓展延迟消息(DelayMessage)支持

🗺️ 更新计划 (Roadmap)

🟡 正在进行 (In Progress)

• EasyNats.Caching: 一个基于NATS KV兼容Redis协议的高可用分布式缓存数据库。 具备混合二级缓存架构 (L1+L2 Hybrid Storage),强一致性集群同步,缓存击穿与雪崩保护性能.
• EasyNats.Outbox: 保证数据库与消息发送的原子性。
• EasyNats.DistributedLock: 基于 NATS KV 的高性能分布式锁。
• EasyNats.Dashboard: 轻量化 UI，实时监控 Stream 流量与消费者健康度。

🚀 快速开始

1. 安装包

dotnet add package EasyNats.DependencyInjection

2. 配置并启用 EasyNats 在 Program.cs 中，通过简单的扩展方法注入服务：

var builder = WebApplication.CreateEmptyBuilder(args);

// 1. 注入基础服务
builder.Services.AddEasyNats(builder.Configuration, options => {
    options.ClientName = "OrderService";
   options.Url = "nats://127.0.0.1:4222";
});

// 2. 注册处理器 (Native AOT 安全)
builder.Services
    .AddConsumer<OrderMessage, OrderConsumer>()
    .AddOrderedConsumer<RoundMessage, RoundConsumer>()
    .AddBroadcastConsumer<NoticeMessage, NotifyConsumer>()
   .AddResponder<NodeStatusRequest, NodeStatusResponse, NodeStatusResponder>();

var app = builder.Build();
app.Run();

3. 定义消息与处理器 消息必须实现 IMessagePackMessage 契约（以便 Source Generator 识别）：

[MessagePackObject]
public class OrderMessage : IMessagePackMessage
{
    [Key(0)]
    public string OrderId { get; set; }
}

//工作队列模式 (Work Queue)：竞争消费，一条消息只处理一次。
public class OrderConsumer : IConsumer<OrderMessage>
{
    public async Task Handle(OrderMessage message, CancellationToken ct)
    {
        // 处理逻辑
       Console.WriteLine($"收到订单: {message.OrderId}");
    }
}

[MessagePackObject]
public class RoundMessage : IMessagePackMessage
{
    [Key(0)]
    public int RoundIndex { get; set; }
}

//顺序队列模式 : 严格保证消息的投递顺序与发送顺序一致。
public class RoundConsumer : IOrderedConsumer<RoundMessage>
{
    public async Task Handle(OrderMessage message, CancellationToken ct)
    {
        // 处理逻辑
       Console.WriteLine($"收到回合: {message.RoundIndex}");
    }
}

[MessagePackObject]
public class NoticeMessage : IMessagePackMessage
{
    [Key(0)]
    public string News { get; set; }
}

//广播模式 (Broadcast)：全员消费，每个实例一份。
public class NotifyConsumer : IBroadcastSubscriber<NoticeMessage>
{
    public async Task Handle(NoticeMessage message, CancellationToken ct)
    {
        // 处理逻辑
       Console.WriteLine($"收到通知: {message.News}");
    }
}


//远程过程调用模式：

[MessagePackObject]
public class NodeStatusRequest : IMessagePackMessage
{
    [Key(0)]
    public Guid NodeId { get; set; }
}

[MessagePackObject]
public class NodeStatusResponse : IMessagePackMessage
{
    [Key(0)]
    public Guid NodeId { get; set; }
    
    [Key(1)]
    public bool IsActive { get; set; }
}

public class NodeStatusResponder : IResponder<GetUserRequest, RpcResponse<NodeStatusResponse>>
{
    private readonly ILogger<NodeStatusResponder> _logger;

    public NodeStatusResponder(ILogger<NodeStatusResponder> logger) 
    {
        _logger = logger;
    }

    public async Task<RpcResponse<NodeStatusResponse>> RespondAsync(NodeStatusRequest request, CancellationToken ct)
    {
        _logger.LogInformation("正在查询节点: {Id}", request.NodeId);

        //耗时操作
        await Task.Delay(10, ct); 

        return RpcResponse<NodeStatusResponse>.Ok(new NodeStatusResponse
        {
            NodeId = Uuid.CreatV7(),
            IsActive = true
        });;
    }
}

4. 定义发布者

//客户端/发布者

//发布消息
public MyService
{
    private readonly IMsgBus _bus;
    private readonly IRpcClient _rpc;

    public MyService(IMsgBus bus, IRpcClient rpc)
    {
        _bus = bus;
        _rpc = rpc;
    }

    public Task test()
    {
        //发布消息
        await _bus.PublishAsync(new RoundMessage{ RoundIndex = 1 });

        //发布延迟消息
        await _bus.PublishDelayAsync(new OrderMessage{ OrderId = "202603219999"} ,TimeSpan.FromSeconds(5));

        //发布延迟消息
        await _bus.EmitAsync(new NoticeMessage{ news = "hello,world!"} ,TimeSpan.FromSeconds(5));

        //远程过程调用
        var response = await _natsBus.CallAsync<NodeStatusRequest, RpcResponse<NodeStatusResponse>>(
            new NodeStatusRequest { NodeId = Guid.NewGuid() });
        Console.WriteLine($"拿到结果: {response.UserName}");
    }
}

🏗️ 架构设计 EasyNats 将复杂的 NATS 协议抽象为三个层次：

Abstractions: 定义核心契约，不依赖驱动，便于单元测试。

Core: 负责物理连接、JetStream 状态机管理、高性能序列化适配。

DependencyInjection: 提供 AOT 安全的扫描与注册机制，管理 Worker 生命周期。

⚠️ Native AOT 发布建议

<PropertyGroup>
  <PublishAot>true</PublishAot>
  <OptimizationPreference>Size</OptimizationPreference>
  <StackTraceSupport>false</StackTraceSupport>
</PropertyGroup>

注意：在 Native AOT 模式下，请避免使用 RegisterAllHandlers 自动扫描程序集，转而使用 AddConsumer<T, H> 显式注册，以确保编译器不会裁剪掉关键的构造函数。

🛠️ 配置参考 (appsettings.json)

{
  "MQSettings": {
    "NatsConnectionString": "nats://localhost:4222",
    "User": "admin",
    "Pass": "123456",
    "ClientName": "EasyClient",
    "StreamPrefix": "EasyStream"
  }
}

📜 开源协议 本项目遵循 MIT 开源协议。