Galosys.Foundation.Amqp 26.5.20.1

Galosys.Foundation.Amqp

成熟度: 🟢 稳定 — AMQP 抽象层，已被 RabbitMQ.Client 模块正式继承

AMQP 消息抽象层模块，定义统一的发送/消费抽象基类、配置模型、[AmqpHandler] 注解、Handler 自动扫描。用户代码仅依赖此模块的 API，切换底层实现只需改 PackageReference。

架构

Galosys.Foundation.Amqp ← 本模块（抽象层）
 ▲
 │
Galosys.Foundation.RabbitMQ.Client ← RabbitMQ 实现（ConnectionPool + BatchPublishPipeline + Publisher Confirms）

功能特性

• 统一 API — AmqpTemplate + [AmqpHandler] + IMessageHandler<T>，切换实现零改动
• 声明式消费 — [AmqpHandler("routingKey")] 方法级注解，启动时自动扫描注册
• 广播消息 — BroadcastAsync fanout 发布
• 延迟消息 — DelayAsync 支持 x-delayed-message 插件
• 消息属性扩展 — AmqpProperties 流式 API：自定义 MessageId、Header、延迟、TTL
• 生命周期管理 — AmqpBootstrapper（IHostedLifecycleService，自动扫描 Handler + 优雅关闭）
• 每消息 DI Scope — 消费端自动创建独立 DI Scope，正确处理 Scoped 生命周期

安装

通常不直接安装此包，而是选择实现包：

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

快速开始

1. 最小配置

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

AppName 默认取应用名称，通常无需显式配置。

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" } });

 // 延迟消息（需 x-delayed-message 插件）
 public async Task DelayAsync() =>
 await _amqp.DelayAsync("order_created", new[] { new OrderCreatedMessage { OrderNo = "ORD001" } }, delayMillionSeconds: 5000);
}

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
 }

 // 广播消费
 [AmqpHandler("order_events", Broadcast = true, Durable = true)]
 public async Task<bool> OnBroadcastAsync(IMessage msg)
 {
 return true;
 }
}

注册机制： 类标注 [Handler] → DI 自动注册；方法标注 [AmqpHandler] → 启动时自动扫描订阅。类需实现 IMessageHandler 或 IMessageHandler<T>。

详细配置

完整配置示例

{
 "RabbitMQ": {
 "HostName": "localhost",
 "UserName": "guest",
 "Password": "guest",
 "VirtualHost": "/",
 "AutoCreateEnabled": true,
 "ConsumerDispatchConcurrency": 8,
 "NetworkRecoveryInterval": 10,
 "RequestedHeartbeat": 45,
 "ContinuationTimeout": 20,
 "Limit": {
 "ProducerMaxConnections": 4,
 "ConsumerMaxConnections": 4,
 "MaxMessageSize": 65536,
 "BatchSize": 100,
 "BatchTimeoutMs": 50,
 "PublishConfirmTimeoutMs": 3000,
 "PipelineCapacity": 5000
 },
 "Handlers": {
 "order_created": {
 "Concurrency": 4,
 "MaxRetryCount": 3,
 "RetryBaseDelayMs": 1000,
 "PrefetchMultiplier": 2,
 "DurableQueue": false,
 "QueueName": null
 }
 }
 }
}

根配置（AmqpOptions）

限流配置（AmqpLimitOptions）

Handler 配置（AmqpHandlerProperty）

核心类

依赖

• Galosys.Foundation.Core
• Microsoft.Extensions.Hosting
• Microsoft.Extensions.Options