Galosys.Foundation.RabbitMQ.Client 26.7.2.1

Galosys.Foundation.RabbitMQ.Client

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

**锟斤拷锟斤拷锟?*: ?? 锟饺讹拷 锟斤拷 锟斤拷锟斤拷锟斤拷 RabbitMQ 锟酵伙拷锟剿o拷锟斤拷锟接筹拷 + 锟斤拷锟斤拷锟杰碉拷 + Publisher Confirms + OTel 锟缴观诧拷锟斤拷

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

锟杰癸拷

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

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

• 锟斤拷锟接筹拷 锟斤拷 锟斤拷锟斤拷锟斤拷/锟斤拷锟斤拷锟竭讹拷锟斤拷锟斤拷锟接池o拷Round-Robin Channel 锟斤拷锟戒,锟脚猴拷锟斤拷锟斤拷锟斤拷
• 直锟斤拷通锟斤拷锟斤拷锟斤拷 锟斤拷 同通锟斤拷锟斤拷锟矫★拷Properties 锟斤拷锟矫★拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷搿ublisher Confirm 追锟斤拷 + 锟斤拷时锟斤拷锟斤拷
• 锟皆讹拷锟斤拷锟剿癸拷锟斤拷 锟斤拷 TopologyCache 锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟侥讹拷锟斤拷/锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟皆讹拷锟截斤拷
• DLX/DLQ 锟斤拷锟斤拷 锟斤拷 锟斤拷锟斤拷失锟斤拷锟皆讹拷路锟缴碉拷锟斤拷锟脚斤拷锟斤拷锟斤拷 + 锟斤拷锟脚讹拷锟斤拷
• 指锟斤拷锟剿憋拷锟斤拷锟斤拷 锟斤拷 锟斤拷锟窖讹拷失锟斤拷锟皆讹拷锟斤拷锟皆o拷x-delayed-message 锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟阶拷锟?DLQ
• 锟斤拷锟斤拷锟斤拷支锟斤拷 锟斤拷 Ping 选锟斤拷锟脚节点,锟皆讹拷锟斤拷锟斤拷指锟?
• OTel 追锟斤拷 锟斤拷 W3C traceparent 锟斤拷锟斤拷锟斤拷Producer/Consumer Span
• OTel 指锟斤拷 锟斤拷 7 锟斤拷 Counter/Histogram/Gauge锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟杰撅拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟斤拷锟接迟★拷锟斤拷锟斤拷锟斤拷小锟斤拷锟斤拷跃锟斤拷锟斤拷锟斤拷锟斤拷
• *锟斤拷锟斤拷锟斤拷锟? 锟斤拷 RabbitMQHealthCheck 锟桔猴拷锟斤拷锟斤拷锟斤拷/锟斤拷锟斤拷锟斤拷锟斤拷锟接筹拷状态锟斤拷Healthy/Degraded/Unhealthy锟斤拷
• IMessageBus 锟斤拷 RabbitMessageBus 实锟斤拷 Core 锟斤拷 IMessageBus 锟接口o拷Send + Publish锟斤拷\n

前锟斤拷锟斤拷锟斤拷

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

锟斤拷装

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

锟斤拷锟劫匡拷始

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

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

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

2. 注锟斤拷锟斤拷锟?

// 锟斤拷式一锟斤拷模锟斤拷锟皆讹拷锟斤拷锟街o拷锟狡硷拷锟斤拷
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);
 // 锟皆讹拷锟斤拷锟斤拷息锟斤拷锟皆o拷锟斤拷 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
 }
 // 锟姐播锟斤拷锟窖o拷fanout + 锟街久伙拷锟斤拷锟叫o拷
 [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 锟斤拷锟狡o拷Galosys.Foundation.RabbitMQ

指锟斤拷

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

锟斤拷锟斤拷锟斤拷

锟斤拷锟斤拷

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