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