Galosys.Foundation.RabbitMQ.Client 26.7.7.2

Galosys.Foundation.RabbitMQ.Client

目标框架:
et8.0\ ·
et10.0\（#if NET10_0\ 条件编译）

定位: RabbitMQ 高性能客户端，连接池 + 批量发布 + Publisher Confirms + OTel 可观测

锟斤拷锟斤拷 RabbitMQ.Client 锟斤拷锟斤拷锟斤拷锟斤拷实锟街ｏ拷锟教筹拷 Galosys.Foundation.Amqp 锟斤拷锟斤拷悖拷峁╋拷锟斤拷映亍锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷艿锟斤拷锟絇ublisher Confirms锟斤拷锟斤拷锟剿伙拷锟芥、DLX/DLQ 锟斤拷锟脚★拷OpenTelemetry 追锟斤拷/指锟斤拷锟斤拷锟揭碉拷锟斤拷锟斤拷浴锟?

锟杰癸拷

Galosys.Foundation.Amqp                锟斤拷 锟斤拷锟斤拷悖ˋmqpTemplate / AmqpPublisher / AmqpConsumer锟斤拷
    锟斤拷
    锟斤拷
Galosys.Foundation.RabbitMQ.Client     锟斤拷 锟斤拷模锟介（RabbitTemplate / RabbitPublisher / RabbitConsumer锟斤拷

锟斤拷锟斤拷锟斤拷锟斤拷

• 锟斤拷锟接筹拷 锟斤拷 锟斤拷锟斤拷锟斤拷/锟斤拷锟斤拷锟竭讹拷锟斤拷锟斤拷锟接池ｏ拷Round-Robin Channel 锟斤拷锟戒，锟脚猴拷锟斤拷锟斤拷锟斤拷
• 直锟斤拷通锟斤拷锟斤拷锟斤拷 锟斤拷 同通锟斤拷锟斤拷锟矫★拷Properties 锟斤拷锟矫★拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷搿ublisher Confirm 追锟斤拷 + 锟斤拷时锟斤拷锟斤拷
• 锟皆讹拷锟斤拷锟剿癸拷锟斤拷 锟斤拷 TopologyCache 锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟侥讹拷锟斤拷/锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟皆讹拷锟截斤拷
• DLX/DLQ 锟斤拷锟斤拷 锟斤拷 锟斤拷锟斤拷失锟斤拷锟皆讹拷路锟缴碉拷锟斤拷锟脚斤拷锟斤拷锟斤拷 + 锟斤拷锟脚讹拷锟斤拷
• 指锟斤拷锟剿憋拷锟斤拷锟斤拷 锟斤拷 锟斤拷锟窖讹拷失锟斤拷锟皆讹拷锟斤拷锟皆ｏ拷x-delayed-message 锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟阶拷锟?DLQ
• OTel 追踪 — W3C traceparent 注入，Producer/Consumer Span
• OTel 指标 — 7 个 Counter/Histogram/Gauge，覆盖发送、消费、延迟、批量大小、连接/通道活跃数
• 健康检查 — RabbitMQHealthCheck 聚合发布者/消费者连接池状态：Healthy/Degraded/Unhealthy
• IMessageBus 锟斤拷 RabbitMessageBus 实锟斤拷 Core 锟斤拷 IMessageBus 锟接口ｏ拷Send + Publish锟斤拷\n

前锟斤拷锟斤拷锟斤拷

• RabbitMQ 3.x+ 锟斤拷锟斤拷锟?
• 锟斤拷锟斤拷 rabbitmq_delayed_message_exchange 锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟揭拷锟?
• 锟斤拷锟介开锟斤拷 rabbitmq_management 锟斤拷锟斤拷锟斤拷锟轿拷锟截ｏ拷

锟斤拷装

<PackageReference Include="Galosys.Foundation.RabbitMQ.Client" Version="x.x.x" />

锟斤拷锟劫匡拷始

1. 锟斤拷小锟斤拷锟斤拷

{
  "RabbitMQ": {
    "HostName": "localhost",
    "Port": 5672,
    "UserName": "guest",
    "Password": "guest",
    "VirtualHost": "/"
  }
}

AppName 默锟斤拷取应锟斤拷锟斤拷锟狡ｏ拷AutoCreateEnabled 默锟斤拷 true 锟皆讹拷锟斤拷锟斤拷锟斤拷锟斤拷/锟斤拷锟斤拷锟斤拷锟斤拷

2. 注锟斤拷锟斤拷锟?

// 锟斤拷式一锟斤拷模锟斤拷锟皆讹拷锟斤拷锟街ｏ拷锟狡硷拷锟斤拷
await Host.CreateDefaultBuilder()
    .UseModularization()
    .Locate()
    .RunAsync();

// 锟斤拷式锟斤拷锟斤拷锟街讹拷注锟斤拷
services.AddRabbitMQ(configuration);

3. 锟斤拷锟斤拷锟斤拷息

using RabbitMQ.Client;

public class OrderCreatedMessage
{
    public string OrderNo { get; set; }
}

public class OrderService
{
    private readonly AmqpTemplate _amqp;

    public OrderService(AmqpTemplate amqp) => _amqp = amqp;

    // 锟斤拷锟斤拷锟斤拷锟斤拷
    public async Task SendAsync() =>
        await _amqp.SendAsync("order_created", new[] { new OrderCreatedMessage { OrderNo = "ORD001" } });

    // 锟斤拷锟斤拷锟斤拷锟斤拷
    public async Task SendBatchAsync() =>
        await _amqp.SendAsync("order_created", new[]
        {
            new OrderCreatedMessage { OrderNo = "ORD001" },
            new OrderCreatedMessage { OrderNo = "ORD002" }
        });

    // 锟姐播锟斤拷息锟斤拷fanout锟斤拷
    public async Task BroadcastAsync() =>
        await _amqp.BroadcastAsync("order_events", new[] { new OrderCreatedMessage { OrderNo = "ORD001" } });

    // 锟接筹拷锟斤拷息
    public async Task DelayAsync() =>
        await _amqp.DelayAsync("order_created", new[] { new OrderCreatedMessage { OrderNo = "ORD001" } }, delayMillionSeconds: 5000);

    // 锟皆讹拷锟斤拷锟斤拷息锟斤拷锟皆ｏ拷锟斤拷 MessageId锟斤拷
    public async Task SendWithPropsAsync() =>
        await _amqp.SendAsync("order_created", new[] { new OrderCreatedMessage { OrderNo = "ORD001" } },
            postMessageProcessor: props =>
            {
                props.MessageId = Guid.NewGuid().ToString();
                props.SetCustomProperty("x-correlation-id", "corr-123");
            });
}

4. 锟斤拷锟斤拷锟斤拷息

using RabbitMQ.Client;

[Handler]
public class OrderHandler : IMessageHandler
{
    [AmqpHandler("order_created")]
    public async Task<bool> HandleAsync(IMessage msg)
    {
        // 锟斤拷锟斤拷锟斤拷息
        return true; // true = ACK, false = Nack 锟斤拷 锟斤拷锟皆伙拷锟斤拷锟?DLQ
    }

    // 锟姐播锟斤拷锟窖ｏ拷fanout + 锟街久伙拷锟斤拷锟叫ｏ拷
    [AmqpHandler("order_events", Broadcast = true, Durable = true)]
    public async Task<bool> OnEventAsync(IMessage msg)
    {
        return true;
    }
}

*注锟斤拷锟斤拷疲锟? 锟斤拷锟阶?[Handler] 锟斤拷 DI 锟皆讹拷注锟结；锟斤拷锟斤拷锟斤拷注 [AmqpHandler] 锟斤拷 锟斤拷锟斤拷时锟皆讹拷扫锟借订锟侥★拷

5. 使锟斤拷 IMessageBus

Outbox 妯″紡涓嬶紝MessageBusOutboxDecorator 鏀堕泦娑堟伅鍒?PendingMsgCol锛圓syncLocal锛夛紝OutboxSaveChangesInterceptor 鍦?SaveChanges 鏃跺悓浜嬪姟鍐欏叆 base_outbox_msg銆?

public class MyService
{
    private readonly IMessageBus _bus;

    public MyService(IMessageBus bus) => _bus = bus;

    public async Task PublishAsync(MyEvent evt) =>
        await _bus.PublishAsync("my_event", evt);
}

锟斤拷细锟斤拷锟斤拷

锟斤拷锟斤拷锟斤拷锟斤拷示锟斤拷

{
  "RabbitMQ": {
    "HostName": "rabbit1, rabbit2, rabbit3",
    "UserName": "user",
    "Password": "pass",
    "VirtualHost": "/",
    "AutoCreateEnabled": true,
    "ConsumerDispatchConcurrency": 16,
    "NetworkRecoveryInterval": 10,
    "RequestedHeartbeat": 45,
    "ContinuationTimeout": 20,
    "Limit": {
      "ProducerMaxConnections": 4,
      "ProducerChannelMaxPerConnection": 10,
      "ConsumerMaxConnections": 4,
      "ConsumerChannelMaxPerConnection": 50,
      "MaxMessageSize": 65536,
      "MaxTps": 0,
      "BatchSize": 100,
      "BatchTimeoutMs": 50,
      "PublishConfirmTimeoutMs": 3000,
      "RetryCount": 3,
      "RetryBaseDelayMs": 1000,
      "PipelineCapacity": 5000
    },
    "Handlers": {
      "order_created": {
        "Concurrency": 4,
        "MaxRetryCount": 3,
        "RetryBaseDelayMs": 1000,
        "PrefetchMultiplier": 2,
        "DurableQueue": false,
        "QueueName": null
      }
    }
  }
}

锟斤拷锟斤拷锟斤拷锟斤拷细说锟斤拷锟斤拷 Amqp 模锟斤拷锟侥碉拷锟斤拷

锟缴观诧拷锟斤拷

锟斤拷锟斤拷锟斤拷锟?

锟皆讹拷注锟结健锟斤拷锟斤拷椋拷锟斤拷锟?/health 锟斤拷锟缴查看 RabbitMQ 状态锟斤拷

锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟轿?"rabbitmq"锟斤拷锟斤拷签 ["rabbitmq", "messaging"]锟斤拷

锟斤拷态锟斤拷Healthy锟斤拷全锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷/ Degraded锟斤拷锟斤拷锟斤拷锟斤拷锟接断匡拷锟斤拷/ Unhealthy锟斤拷全锟斤拷锟较匡拷锟斤拷锟斤拷

OpenTelemetry 追锟斤拷

ActivitySource 锟斤拷锟狡ｏ拷Galosys.Foundation.RabbitMQ

指锟斤拷

Meter 锟斤拷锟狡ｏ拷Galosys.Foundation.RabbitMQ

锟斤拷锟斤拷锟斤拷

锟斤拷锟斤拷

• Galosys.Foundation.Amqp锟斤拷锟斤拷锟斤拷悖?
• Galosys.Foundation.Core锟斤拷IMessageBus锟斤拷IObjectSerializer锟斤拷
• RabbitMQ.Client锟斤拷NuGet锟斤拷