Maomi.MQ

https://docs.whuanle.cn/zh/maomi_mq
Admin
Maomi.MQ is a RabbitMQ messaging framework that simplifies message publishing, consumption, routing,...
Tokens:52,765
Snippets:518
Trust Score:6.9
Update:6 months ago
Show doc for...
Tokens:
Raw
### Install Maomi.MQ Packages

Source: https://docs.whuanle.cn/zh/maomi_mq/10.other_support

Install the necessary Maomi.MQ packages for RabbitMQ, MediatR, and FastEndpoints integration.

```bash
dotnet add package Maomi.MQ.RabbitMQ
dotnet add package Maomi.MQ.MediatR
dotnet add package Maomi.MQ.FastEndpoints
```

--------------------------------

### EF Core Publisher Example with Delegated Publish

Source: https://docs.whuanle.cn/zh/maomi_mq/8.transcation

A comprehensive example of creating an order and publishing a message using `IEfCoreTransactionService.ExecuteAndRegisterAsync`. It includes adding the order to the database and registering the `OrderCreatedMessage` for asynchronous publishing.

```csharp
public async Task<long> CreateOrderAsync(CreateOrderRequest request, CancellationToken ct)
{
    var orderId = _idGen.NextId();
    var message = new OrderCreatedMessage
    {
        OrderId = orderId,
        UserId = request.UserId,
        Amount = request.Amount
    };

    var outbox = await _txService.ExecuteAndRegisterAsync(
        _db,
        async (db, cancellationToken) =>
        {
            db.Set<Order>().Add(new Order
            {
                Id = orderId,
                UserId = request.UserId,
                Amount = request.Amount,
                Status = OrderStatus.Created
            });
            await Task.CompletedTask;
        },
        exchange: "",
        routingKey: "order.created",
        message,
        cancellationToken: ct);

    await outbox.PublishAsync(ct);
    return orderId;
}

```

--------------------------------

### Dynamically Start Consumers

Source: https://docs.whuanle.cn/zh/maomi_mq/2.3.dynamic.md

Use the IDynamicConsumer service to dynamically start consumers with specified options. This method is suitable for starting consumers with defined model classes.

```csharp
private readonly IMessagePublisher _messagePublisher;
private readonly IDynamicConsumer _dynamicConsumer;

[HttpPost("create")]
public async Task<string> CreateConsumer([FromBody] ConsumerDto consumer)
{
	foreach (var item in consumer.Queues)
	{
		await _dynamicConsumer.ConsumerAsync<MyConsumer, TestEvent>(new ConsumerOptions(item));
	}

	return "ok";
}
```

--------------------------------

### Dynamically Start Consumers Using Functions

Source: https://docs.whuanle.cn/zh/maomi_mq/2.3.dynamic.md

Start consumers dynamically using function delegates for execute, faild, and fallback logic, without needing to define a separate model class.

```csharp
foreach (var item in consumer.Queues)
{
	var consumerTag = await _dynamicConsumer.ConsumerAsync<TestEvent>(
		consumerOptions: new ConsumerOptions(item),
		execute: async (header, message) =>
		{
			await Task.CompletedTask;
		},
		faild: async (header, ex, retryCount, message) => { },
		fallback: async (header, message, ex) => ConsumerState.Ack
		);
}
```

--------------------------------

### Install Transactional NuGet Packages

Source: https://docs.whuanle.cn/zh/maomi_mq/2.publisher

Required packages for implementing the local transaction table pattern.

```text
Maomi.MQ.Transaction.Mysql
Maomi.MQ.Transaction.Postgres
Maomi.MQ.Transaction.SqlServer
```

--------------------------------

### Add Maomi.MQ.Transaction Package

Source: https://docs.whuanle.cn/zh/maomi_mq/8.transcation.md

Install the necessary NuGet packages for Maomi.MQ.Transaction and its MySQL provider. Replace `Maomi.MQ.Transaction.Mysql` with the appropriate provider for PostgreSQL or SQL Server if needed.

```bash
dotnet add package Maomi.MQ.Transaction
dotnet add package Maomi.MQ.Transaction.Mysql

```

--------------------------------

### Complete Order Creation Workflow (Transaction + Outbox)

Source: https://docs.whuanle.cn/zh/maomi_mq/8.transcation.md

This example demonstrates the full workflow for creating an order: writing business data, registering an outbox message within the same transaction, committing the transaction, and then publishing the message. This ensures atomicity between data persistence and message intent.

```csharp
public async Task<long> CreateOrderAsync(CreateOrderRequest request, CancellationToken ct)
{
    await using var connection = new MySqlConnection(_connectionString);
    await connection.OpenAsync(ct);
    await using var tx = await connection.BeginTransactionAsync(ct);

    var orderId = _idGen.NextId();
    var message = new OrderCreatedMessage
    {
        OrderId = orderId,
        UserId = request.UserId,
        Amount = request.Amount
    };

    // 1) 写业务数据
    await InsertOrderAsync(connection, tx, orderId, request, ct);

    // 2) 同事务登记 outbox（这里会生成并持久化 MessageId）
    var outbox = await _outboxService.RegisterAsync(
        connection,
        tx,
        exchange: "",
        routingKey: "order.created",
        message,
        cancellationToken: ct);

    // 3) 提交本地事务
    await tx.CommitAsync(ct);

    // 4) 提交后立即发布（内部会处理成功/失败状态）
    await outbox.PublishAsync(ct);

    return orderId;
}

```

--------------------------------

### EF Core Publish Example with ExecuteAndRegisterAsync

Source: https://docs.whuanle.cn/zh/maomi_mq/8.transcation

Demonstrates publishing a message using `IEfCoreTransactionService.ExecuteAndRegisterAsync`. This method executes business logic within a transaction, registers the message to be sent (outbox), and returns an object to publish the message.

```csharp
var outbox = await _efCoreTransactionService.ExecuteAndRegisterAsync(
    dbContext,
    async (db, ct) =>
    {
        db.Set<Order>().Add(order);
        await Task.CompletedTask;
    },
    exchange: "",
    routingKey: "order.created",
    message,
    cancellationToken: ct);

await outbox.PublishAsync(ct);

```

--------------------------------

### Basic Consumer Implementation

Source: https://docs.whuanle.cn/zh/maomi_mq/2.0.consumer

Example of a basic consumer implementing IConsumer<TMessage>. Configure queue binding and retry behavior using attributes. This pattern includes ExecuteAsync, FaildAsync, and FallbackAsync for message processing and error handling.

```csharp
public class TestEvent
{
    public int Id { get; set; }
}

[Consumer("PublisherWeb", Qos = 1, RetryFaildRequeue = true)]
public class MyConsumer : IConsumer<TestEvent>
{
    private static int _retryCount = 0;

    // 消费
    public async Task ExecuteAsync(MessageHeader messageHeader, TestEvent message)
    {
        _retryCount++;
        Console.WriteLine($"执行次数:{_retryCount} 事件 id: {message.Id} {DateTime.Now}");
        await Task.CompletedTask;
    }

    // 每次消费失败时执行
    public Task FaildAsync(MessageHeader messageHeader, Exception ex, int retryCount, TestEvent message)
        => Task.CompletedTask;

    // 补偿
    public Task<ConsumerState> FallbackAsync(MessageHeader messageHeader, TestEvent? message, Exception? ex)
        => Task.FromResult(ConsumerState.Ack);
}

```

--------------------------------

### Configure Maomi.MQ with Basic Options

Source: https://docs.whuanle.cn/zh/maomi_mq/3.configuration

Configure Maomi.MQ with essential options like WorkId, AutoQueueDeclare, AppName, and RabbitMQ connection details. The WorkId is crucial for distributed unique ID generation, and AppName identifies the application. Ensure the RabbitMQ host is correctly set, for example, via an environment variable.

```csharp
builder.Services.AddMaomiMQ((MqOptionsBuilder options) =>
{
    // 必填，当前程序节点，用于配置分布式雪花 id，
    // 配置 WorkId 可以避免高并发情况下同一个消息的 id 重复。
	options.WorkId = 1;
    
    // 是否自动创建队列
	options.AutoQueueDeclare = true;
    
    // 当前应用名称，用于标识消息的发布者和消费者程序
	options.AppName = "myapp";
    
    // 必填，RabbitMQ 配置
	options.Rabbit = (ConnectionFactory options) =>
	{
        options.HostName = Environment.GetEnvironmentVariable("RABBITMQ")!;
        options.Port = 5672;
		options.ClientProvidedName = Assembly.GetExecutingAssembly().GetName().Name;
	};
}, [typeof(Program).Assembly]);  // 要被扫描的程序集

```

--------------------------------

### 配置服务注册

Source: https://docs.whuanle.cn/zh/maomi_mq/8.transcation

在 Startup 或 Program.cs 中配置 Maomi.MQ.Transaction 及数据库连接。

```csharp
using Maomi.MQ.Transaction.Mysql;
using MySqlConnector;

builder.Services.AddMaomiMQTransactionMySql();

builder.Services.AddMaomiMQTransaction(options =>
{
    options.ProviderName = "mysql";
    options.Connection = _ => new MySqlConnection(connectionString);
    options.AutoCreateTable = true;
});

builder.Services.AddMaomiMQ(
    (MqOptionsBuilder options) =>
    {
        // RabbitMQ 配置
    },
    [typeof(Program).Assembly],
    Maomi.MQ.Extensions.CreateTransactionFilters());
```

--------------------------------

### Consumer Type Filter Example

Source: https://docs.whuanle.cn/zh/maomi_mq/3.configuration.md

Example of using a `ConsumerTypeFilter` to intercept and modify consumer options, such as prefixing queue names.

```APIDOC
## Consumer Type Filter

### Description
Allows intercepting and modifying `IConsumerOptions` before they are registered.

### Method
```csharp
new ConsumerTypeFilter((consumerOptions, type) => { ... })
```

### Endpoint
N/A (This is a code construct, not an HTTP endpoint)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```csharp
new ConsumerTypeFilter((consumerOptions, type) =>
{
    var newConsumerOptions = new ConsumerOptions();
    newConsumerOptions.CopyFrom(consumerOptions);

    newConsumerOptions.Queue = "app1_" + newConsumerOptions.Queue;

    // true indicates to register the consumer; false indicates to ignore it.
    return new RegisterQueue(true, newConsumerOptions);
});
```

### Response
#### Success Response (200)
Returns a `RegisterQueue` object indicating whether to register the consumer and its modified options.

#### Response Example
```csharp
// Example of RegisterQueue object
public record RegisterQueue(bool Register, ConsumerOptions Options);
```
```

--------------------------------

### IRoutingProvider Basic Implementation

Source: https://docs.whuanle.cn/zh/maomi_mq/3.configuration.md

Demonstrates a basic implementation of `IRoutingProvider` for modifying consumer options and router key options, such as adding environment-specific prefixes.

```APIDOC
## IRoutingProvider Basic Implementation

### Description
Provides a mechanism to dynamically adjust routing configurations for consumers and message publishing based on runtime conditions.

### Method
```csharp
services.AddSingleton<IRoutingProvider, MyRoutingProvider>();
```

### Endpoint
N/A (This is a service registration, not an HTTP endpoint)

### Parameters
None

### Request Example
```csharp
// Service registration in Startup.cs or Program.cs
builder.Services.AddSingleton<IRoutingProvider, MyRoutingProvider>();
```

### Response
#### Success Response (200)
N/A

#### Response Example
```csharp
using Maomi.MQ.Consumer;

public sealed class MyRoutingProvider : IRoutingProvider
{
    private readonly string _prefix;

    public MyRoutingProvider(IHostEnvironment env)
    {
        _prefix = env.IsDevelopment() ? "dev." : "prod.";
    }

    public IConsumerOptions Get(IConsumerOptions options)
    {
        var newOptions = new ConsumerOptions();
        newOptions.CopyFrom(options);
        newOptions.Queue = _prefix + options.Queue;
        return newOptions;
    }

    public IRouterKeyOptions Get(IRouterKeyOptions options)
    {
        return new RouterKeyOptions(_prefix + options.RoutingKey, options.Exchange);
    }

    private sealed class RouterKeyOptions : IRouterKeyOptions
    {
        public RouterKeyOptions(string routingKey, string? exchange)
        {
            RoutingKey = routingKey;
            Exchange = exchange;
        }

        public string RoutingKey { get; }
        public string? Exchange { get; }
    }
}
```
```

--------------------------------

### Inject EF Core Transaction Services

Source: https://docs.whuanle.cn/zh/maomi_mq/8.transcation.md

Example of injecting IEfCoreTransactionService and the application DbContext into a service class.

```csharp
public sealed class OrderEfCoreService
{
    private readonly IEfCoreTransactionService _txService;
    private readonly AppDbContext _db;

    public OrderEfCoreService(IEfCoreTransactionService txService, AppDbContext db)
    {
        _txService = txService;
        _db = db;
    }
}
```

--------------------------------

### 安装 NuGet 包

Source: https://docs.whuanle.cn/zh/maomi_mq/8.transcation

通过 dotnet CLI 安装核心库及对应的数据库驱动。

```bash
dotnet add package Maomi.MQ.Transaction
dotnet add package Maomi.MQ.Transaction.Mysql
```

--------------------------------

### Define TestEvent Class

Source: https://docs.whuanle.cn/zh/maomi_mq/4.qos

Defines the structure for messages to be published and consumed. This class is used across both the publisher and consumer examples.

```csharp
public class TestEvent
{
    public int Id { get; set; }
    public string Message { get; set; }
    public int[] Data { get; set; }

    public override string ToString()
    {
        return Id.ToString();
    }
}
```

--------------------------------

### Customize Consumer Options with Type Filter

Source: https://docs.whuanle.cn/zh/maomi_mq/3.configuration

Use a ConsumerTypeFilter to intercept and modify IConsumerOptions before registration. This example demonstrates adding a prefix to the queue name.

```csharp
new ConsumerTypeFilter((consumerOptions, type) =>
{
    var newConsumerOptions = new ConsumerOptions();
    newConsumerOptions.CopyFrom(consumerOptions);

    newConsumerOptions.Queue = "app1_" + newConsumerOptions.Queue;

    // true 表示注册该消费者；false 表示忽略该消费者
    return new RegisterQueue(true, newConsumerOptions);
});
```

--------------------------------

### Dead-Letter Queue Consumer Example

Source: https://docs.whuanle.cn/zh/maomi_mq/2.1.consumer.md

A consumer designed to process messages from a dead-letter queue. It logs the event ID and returns Ack to acknowledge processing.

```csharp
// ConsumerWeb_dead 消费失败的消息会被此消费者消费。
[Consumer("consumerWeb_dead_queue", Qos = 1)]
public class Dead_QueueConsumer : IConsumer<DeadQueueEvent>
{
    // 消费
    public Task ExecuteAsync(MessageHeader messageHeader, DeadQueueEvent message)
    {
        Console.WriteLine($"死信队列，事件 id:{message.Id}");
        return Task.CompletedTask;
    }

    // 每次失败时被执行
    public Task FaildAsync(MessageHeader messageHeader, Exception ex, int retryCount, DeadQueueEvent message) => Task.CompletedTask;

    // 最后一次失败时执行
    public Task<ConsumerState> FallbackAsync(MessageHeader messageHeader, DeadQueueEvent? message, Exception? ex)
        => Task.FromResult(ConsumerState.Ack);
}
```

--------------------------------

### 配置 Maomi.MQ 服务

Source: https://docs.whuanle.cn/zh/maomi_mq/1.start.md

在 Web 项目中注入 Maomi.MQ 服务并配置 RabbitMQ 连接参数。

```csharp
// using Maomi.MQ;
// using RabbitMQ.Client;

builder.Services.AddMaomiMQ((MqOptionsBuilder options) =>
{
    options.WorkId = 1;
    options.AppName = "myapp";
    options.Rabbit = (ConnectionFactory options) =>
    {
        options.HostName = Environment.GetEnvironmentVariable("RABBITMQ")!;
        options.Port = 5672;
    };
}, [typeof(Program).Assembly]);

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

--------------------------------

### Dynamically Stop Consumers by Consumer Tag

Source: https://docs.whuanle.cn/zh/maomi_mq/2.3.dynamic.md

Stop a specific consumer dynamically using its unique consumer tag. This requires obtaining the consumer tag when the consumer is initially started.

```csharp
var consumerTag = await _dynamicConsumer.ConsumerAsync<MyConsumer, TestEvent>(new ConsumerOptions(item));
await _dynamicConsumer.StopConsumerTagAsync(consumerTag);
```

--------------------------------

### 引入 Maomi.MQ 可观测性依赖包

Source: https://docs.whuanle.cn/zh/maomi_mq/7.opentelemetry

在项目中添加必要的 NuGet 包引用。

```text
Maomi.MQ.Instrumentation
OpenTelemetry.Exporter.Console
OpenTelemetry.Exporter.OpenTelemetryProtocol
OpenTelemetry.Extensions.Hosting
OpenTelemetry.Instrumentation.AspNetCore
```

--------------------------------

### EF Core Barrier Consume Example

Source: https://docs.whuanle.cn/zh/maomi_mq/8.transcation

Shows how to consume messages within a transactional barrier using `IEfCoreTransactionService.ExecuteInBarrierAsync`. This ensures that message processing is part of the EF Core transaction.

```csharp
await _efCoreTransactionService.ExecuteInBarrierAsync(
    dbContext,
    consumerName: "order.created",
    messageHeader,
    async (db, ct) =>
    {
        await Task.CompletedTask;
    },
    ct);

```

--------------------------------

### Basic Consumer Pattern

Source: https://docs.whuanle.cn/zh/maomi_mq/2.0.consumer.md

Demonstrates the implementation of a basic consumer using the `IConsumer<TMessage>` interface and the `[Consumer]` attribute for configuration.

```APIDOC
## Basic Consumer Pattern (`IConsumer<TMessage>`)

### Description
This pattern involves implementing the `IConsumer<TMessage>` interface and using the `[Consumer("queue_name")]` attribute to bind to a specific queue. Consumers manage consumption behavior and include failure notification and compensation mechanisms.

### Interface
`IConsumer<TMessage>`

### Methods
- **ExecuteAsync**(MessageHeader messageHeader, TMessage message) - Task
  - Description: Processes the received message after successful deserialization.
- **FaildAsync**(MessageHeader messageHeader, Exception ex, int retryCount, TMessage message) - Task
  - Description: Executed immediately when an exception occurs during `ExecuteAsync`.
- **FallbackAsync**(MessageHeader messageHeader, TMessage? message, Exception? ex) - Task<ConsumerState>
  - Description: Executed when all retries fail or an exception prevents `ExecuteAsync` from being accessed.

### Example Implementation
```csharp
public class TestEvent
{
    public int Id { get; set; }
}

[Consumer("PublisherWeb", Qos = 1, RetryFaildRequeue = true)]
public class MyConsumer : IConsumer<TestEvent>
{
    private static int _retryCount = 0;

    // Consume the message
    public async Task ExecuteAsync(MessageHeader messageHeader, TestEvent message)
    {
        _retryCount++;
        Console.WriteLine($"Execution Count:{_retryCount} Event Id: {message.Id} {DateTime.Now}");
        await Task.CompletedTask;
    }

    // Executed on each consumption failure
    public Task FaildAsync(MessageHeader messageHeader, Exception ex, int retryCount, TestEvent message)
        => Task.CompletedTask;

    // Compensation logic
    public Task<ConsumerState> FallbackAsync(MessageHeader messageHeader, TestEvent? message, Exception? ex)
        => Task.FromResult(ConsumerState.Ack);
}
```
```

--------------------------------

### Define Event Interceptor Function

Source: https://docs.whuanle.cn/zh/maomi_mq/3.configuration.md

Implement an event interceptor function to customize event consumer options. This example shows how to modify the queue name based on the event type.

```csharp
private static RegisterQueue EventInterceptor(IConsumerOptions consumerOptions, Type eventType)
{
	if (eventType == typeof(TestEvent))
	{
		var newConsumerOptions = new ConsumerOptions();
		newConsumerOptions.CopyFrom(consumerOptions);
		newConsumerOptions.Queue = newConsumerOptions.Queue + "_1";

		return new RegisterQueue(true, newConsumerOptions);
	}
	return new RegisterQueue(true, consumerOptions);
}

```

--------------------------------

### 部署可观测性服务

Source: https://docs.whuanle.cn/zh/maomi_mq/7.opentelemetry

使用 Docker Compose 启动已配置好的基础设施服务。

```bash
docker compose up -d
```

--------------------------------

### 自动发布配置

Source: https://docs.whuanle.cn/zh/maomi_mq/1.start.md

通过模型类上的 [RouterKey] 特性简化消息发布流程。

```csharp
[RouterKey("scenario.quickstart")]
public sealed class QuickStartMessage
{
}
```

```csharp
await _publisher.AutoPublishAsync(message);
```

```csharp
await _publisher.PublishAsync(string.Empty, request.Queue, message);
```

```csharp
[Consumer("scenario.quickstart")]
public sealed class QuickStartConsumer : IConsumer<QuickStartMessage>
{
}
```

--------------------------------

### Forwarding Unroutable Messages to a Failure Queue

Source: https://docs.whuanle.cn/zh/maomi_mq/2.publisher

Implement the BasicReturnAsync method to handle messages that cannot be routed. This example forwards such messages to a new queue named with a '.faild' suffix for further inspection or processing.

```csharp
public class MyDefaultBreakdown : IBreakdown
{
    private readonly ConnectionPool _connectionPool;

    public MyDefaultBreakdown(ConnectionPool connectionPool)
    {
        _connectionPool = connectionPool;
    }

    /// <inheritdoc />
    public async Task BasicReturnAsync(object sender, BasicReturnEventArgs @event)
    {
        var connectionObject = _connectionPool.Get();
        await connectionObject.DefaultChannel.BasicPublishAsync<BasicProperties>(
            @event.Exchange, 
            @event.RoutingKey + ".faild", 
            true, 
            new BasicProperties(@event.BasicProperties), 
            @event.Body);
    }

    /// <inheritdoc />
    public Task NotFoundConsumerAsync(string queue, Type messageType, Type consumerType)
    {
        return Task.CompletedTask;
    }
}
```

--------------------------------

### Configure MaomiMQ Transaction with Cleanup Strategy

Source: https://docs.whuanle.cn/zh/maomi_mq/8.transcation

Configure MaomiMQ transaction services with automatic cleanup options for message tables. This example sets up cleanup based on retention days and completed record count.

```csharp
builder.Services.AddMaomiMQTransaction(options =>
{
    options.ProviderName = "mysql";
    options.Connection = _ => new MySqlConnection(connectionString);

    options.Cleanup = new MQTransactionCleanupOptions
    {
        Enabled = true,
        ScanInterval = TimeSpan.FromMinutes(2),
        KeepCompletedDays = 7,
        MaxCompletedCount = 100000,
        DeleteBatchSize = 1000
    };
});
```

--------------------------------

### Basic IRoutingProvider Implementation

Source: https://docs.whuanle.cn/zh/maomi_mq/3.configuration

A base implementation of IRoutingProvider that prefixes queue and routing keys based on the host environment (development or production).

```csharp
using Maomi.MQ.Consumer;

public sealed class MyRoutingProvider : IRoutingProvider
{
    private readonly string _prefix;

    public MyRoutingProvider(IHostEnvironment env)
    {
        _prefix = env.IsDevelopment() ? "dev." : "prod.";
    }

    public IConsumerOptions Get(IConsumerOptions options)
    {
        var newOptions = new ConsumerOptions();
        newOptions.CopyFrom(options);
        newOptions.Queue = _prefix + options.Queue;
        return newOptions;
    }

    public IRouterKeyOptions Get(IRouterKeyOptions options)
    {
        return new RouterKeyOptions(_prefix + options.RoutingKey, options.Exchange);
    }

    private sealed class RouterKeyOptions : IRouterKeyOptions
    {
        public RouterKeyOptions(string routingKey, string? exchange)
        {
            RoutingKey = routingKey;
            Exchange = exchange;
        }

        public string RoutingKey { get; }

        public string? Exchange { get; }
    }
}
```

--------------------------------

### Implement a Basic Consumer

Source: https://docs.whuanle.cn/zh/maomi_mq/1.start

Implement the IConsumer<TEvent> interface and use the [Consumer] attribute to configure consumer properties. The ExecuteAsync method handles messages, FaildAsync is called on exceptions, and FallbackAsync handles final fallback logic.

```csharp
[Consumer("test")]
public class MyConsumer : IConsumer<TestEvent>
{
    // 消费
    public async Task ExecuteAsync(MessageHeader messageHeader, TestEvent message)
    {
        Console.WriteLine($"事件 id: {message.Id} {DateTime.Now}");
        await Task.CompletedTask;
    }

    // 每次消费失败时执行
    public Task FaildAsync(MessageHeader messageHeader, Exception ex, int retryCount, TestEvent message) 
        => Task.CompletedTask;

    // 补偿
    public Task<ConsumerState> FallbackAsync(MessageHeader messageHeader, TestEvent? message, Exception? ex) 
        => Task.FromResult( ConsumerState.Ack);
}

```

--------------------------------

### Transactional Consumer Implementation

Source: https://docs.whuanle.cn/zh/maomi_mq/8.transcation

Example of a consumer that uses `ITransactionBarrierService` to execute business logic within a transaction barrier. This pattern centralizes transactional logic and handles framework-level concerns like state changes and rollbacks.

```csharp
public sealed class TransactionOrderCreatedConsumer : IConsumer<OrderCreatedMessage>
{
    private readonly ITransactionBarrierService _barrierService;

    public TransactionOrderCreatedConsumer(ITransactionBarrierService barrierService)
    {
        _barrierService = barrierService;
    }

    public Task ConsumeAsync(MessageHeader header, OrderCreatedMessage message, CancellationToken cancellationToken)
    {
        return _barrierService.ExecuteInBarrierAsync(
            consumerName: "order.created.consumer",
            header,
            async (connection, transaction, ct) =>
            {
                // 业务逻辑：建议保持幂等，例如按 OrderId 更新状态/写流水
                await HandleOrderCreatedAsync(connection, transaction, message, ct);
            },
            cancellationToken);
    }
}

```

--------------------------------

### 集成 MediatR 消费与发布

Source: https://docs.whuanle.cn/zh/maomi_mq/1.start.md

通过 MediatRConsumer 特性将 MQ 消息映射为 MediatR 命令，并使用 MediatRMqCommand 发布消息。

```csharp
using Maomi.MQ.MediatR;
using MediatR;

[MediatRConsumer("biz.mediatr.order", Qos = 1)]
public sealed class SyncOrderCommand : IRequest
{
    public string OrderNo { get; set; } = string.Empty;
}

public sealed class SyncOrderCommandHandler : IRequestHandler<SyncOrderCommand>
{
    public Task Handle(SyncOrderCommand request, CancellationToken cancellationToken)
        => Task.CompletedTask;
}

// 通过 MediatR 触发 MQ 发布
await mediator.Send(new MediatRMqCommand<SyncOrderCommand>
{
    Message = new SyncOrderCommand { OrderNo = "SO-MED-001" }
});
```

--------------------------------

### Configure Dead Letter Queue Properties in Consumer

Source: https://docs.whuanle.cn/zh/maomi_mq/1.start

Configure dead letter queue properties directly on the consumer using attributes like RetryFaildRequeue, DeadExchange, and DeadRoutingKey. This simplifies the setup for retry and dead-lettering mechanisms.

```csharp
[Consumer(
    "example.retry.main",
    RetryFaildRequeue = false,
    DeadExchange = "",
    DeadRoutingKey = "example.retry.dead")]
public sealed class RetryConsumer : IConsumer<RetryMessage>
{
}

```

--------------------------------

### Create Order with Transactional Outbox

Source: https://docs.whuanle.cn/zh/maomi_mq/8.transcation.md

Demonstrates creating an order and publishing a message atomically using the EF Core transaction service.

```csharp
public async Task<long> CreateOrderAsync(CreateOrderRequest request, CancellationToken ct)
{
    var orderId = _idGen.NextId();
    var message = new OrderCreatedMessage
    {
        OrderId = orderId,
        UserId = request.UserId,
        Amount = request.Amount
    };

    var outbox = await _txService.ExecuteAndRegisterAsync(
        _db,
        async (db, cancellationToken) =>
        {
            db.Set<Order>().Add(new Order
            {
                Id = orderId,
                UserId = request.UserId,
                Amount = request.Amount,
                Status = OrderStatus.Created
            });
            await Task.CompletedTask;
        },
        exchange: "",
        routingKey: "order.created",
        message,
        cancellationToken: ct);

    await outbox.PublishAsync(ct);
    return orderId;
}
```

--------------------------------

### 配置 OpenTelemetry 链路追踪与监控

Source: https://docs.whuanle.cn/zh/maomi_mq/7.opentelemetry

在依赖注入容器中配置 OpenTelemetry，包括 Maomi.MQ 的追踪和指标导出设置。

```csharp
builder.Services.AddOpenTelemetry()
	  .ConfigureResource(resource => resource.AddService(serviceName))
	  .WithTracing(tracing =>
	  {
		  tracing.AddMaomiMQInstrumentation(options =>
		  {
			  options.Sources.AddRange(MaomiMQDiagnostic.Sources);
			  options.RecordException = true;
		  })
		  .AddAspNetCoreInstrumentation()
		  .AddOtlpExporter(options =>
		  {
			  options.Endpoint = new Uri(Environment.GetEnvironmentVariable("OTLPEndpoint")! + "/v1/traces");
			  options.Protocol = OtlpExportProtocol.HttpProtobuf;
		  });
	  })
	  .WithMetrics(metrices =>
	  {
		  metrices.AddAspNetCoreInstrumentation()
		  .AddMaomiMQInstrumentation()
		  .AddOtlpExporter(options =>
		  {
			  options.Endpoint = new Uri(Environment.GetEnvironmentVariable("OTLPEndpoint")! + "/v1/metrics");
			  options.Protocol = OtlpExportProtocol.HttpProtobuf;
		  });
	  });
```

--------------------------------

### 创建新的连接对象

Source: https://docs.whuanle.cn/zh/maomi_mq/2.publisher

使用 ConnectionPool 创建新的连接和通道，注意管理生命周期以避免内存泄漏。

```csharp
using var newConnectionObject = _connectionPool.Create();
using var newConnection = newConnectionObject.Connection;
using var newChannel = newConnection.CreateChannelAsync();
```

--------------------------------

### 动态消费者操作

Source: https://docs.whuanle.cn/zh/maomi_mq/2.1.consumer.md

使用 IDynamicConsumer 在运行时动态绑定或停止消费者。

```csharp
var consumerTag = await _dynamicConsumer.ConsumerAsync<MyConsumer, TestEvent>(new ConsumerOptions("myqueue")
{
	Qos = 10
});
```

```csharp
await _dynamicConsumer.StopConsumerTagAsync(consumerTag);
await _dynamicConsumer.StopConsumerAsync(queueName);
```

--------------------------------

### 配置延迟队列和死信队列

Source: https://docs.whuanle.cn/zh/maomi_mq/6.dead_queue.md

设置队列消息过期时间和绑定死信队列。继承 EmptyConsumer 的队列会在程序启动时创建但不会进行消费。

```csharp
[Consumer("ConsumerWeb_dead_2", Expiration = 6000, DeadQueue = "ConsumerWeb_dead_queue_2")]
public class EmptyDeadConsumer : EmptyConsumer<DeadEvent>
{
}

// ConsumerWeb_dead 消费失败的消息会被此消费者消费。
[Consumer("ConsumerWeb_dead_queue_2", Qos = 1)]
public class Dead_2_QueueConsumer : IConsumer<DeadQueueEvent>
{
    // 消费
    public Task ExecuteAsync(MessageHeader messageHeader, TestEvent message)
    {
        Console.WriteLine($"事件 id:{message.Id} 已到期");
        return Task.CompletedTask;
    }
}
```

--------------------------------

### 定义 protobuf-net 消息模型

Source: https://docs.whuanle.cn/zh/maomi_mq/9.serializer.md

使用 ProtoContract 和 ProtoMember 特性标注模型以支持 protobuf-net 序列化。

```csharp
[ProtoContract]
public sealed class PersonMessage
{
    [ProtoMember(1)] public Guid Id { get; set; }
    [ProtoMember(2)] public string Name { get; set; } = string.Empty;
}
```

--------------------------------

### 使用 RouterKey 特性自动路由

Source: https://docs.whuanle.cn/zh/maomi_mq/2.publisher

在模型类上使用 RouterKey 特性，发送消息时无需手动指定路由键。

```csharp
[RouterKey("scenario.quickstart")]
public sealed class QuickStartMessage
```

```csharp
var message = new QuickStartMessage
{
	Text = request.Text,
	At = DateTimeOffset.UtcNow
};

await _publisher.AutoPublishAsync(message);
```

--------------------------------

### 设置消费者 Qos 为 1

Source: https://docs.whuanle.cn/zh/maomi_mq/4.qos

将 Qos 设置为 1 以确保消息被逐个推送，从而实现严格的顺序消费。

```csharp
[Consumer("web1", Qos = 1)]
public class MyConsumer : IConsumer<TestEvent>
{
}
```

--------------------------------

### 实现消费者服务

Source: https://docs.whuanle.cn/zh/maomi_mq/2.1.consumer.md

通过实现 IConsumer 接口并添加 Consumer 特性来定义消费者逻辑。

```csharp
[Consumer("ConsumerWeb", Qos = 1)]
public class MyConsumer : IConsumer<TestEvent>
{
    private readonly ILogger<MyConsumer> _logger;

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

    // 消费
    public async Task ExecuteAsync(MessageHeader messageHeader, TestEvent message)
    {
        Console.WriteLine($"事件 id:{message.Id}");
    }

    // 每次失败时被执行
    public async Task FaildAsync(MessageHeader messageHeader, Exception ex, int retryCount, TestEvent message)
    {
        _logger.LogError(ex, "Consumer exception,event id: {Id},retry count: {retryCount}", message!.Id, retryCount);
    }

    // 最后一次失败时执行
    public async Task<ConsumerState> FallbackAsync(MessageHeader messageHeader, TestEvent? message, Exception? ex)
    {
        return ConsumerState.Ack;
    }
}
```

--------------------------------

### 引入 OpenTelemetry 命名空间

Source: https://docs.whuanle.cn/zh/maomi_mq/7.opentelemetry

在代码文件中添加必要的 OpenTelemetry 和 Maomi.MQ 命名空间引用。

```csharp
using OpenTelemetry.Logs;
using OpenTelemetry.Metrics;
using OpenTelemetry.Resources;
using OpenTelemetry.Trace;
using Maomi.MQ;
using OpenTelemetry.Exporter;
using RabbitMQ.Client;
using System.Reflection;
using OpenTelemetry;
```

--------------------------------

### 消费异常处理与补偿

Source: https://docs.whuanle.cn/zh/maomi_mq/2.1.consumer.md

处理消费失败时的日志记录及最终补偿逻辑。

```csharp
// 每次失败时被执行
public async Task FaildAsync(MessageHeader messageHeader, Exception ex, int retryCount, TestEvent message)
{
	_logger.LogError(ex, "Consumer exception,event id: {Id},retry count: {retryCount}", message!.Id, retryCount);
}
```

```csharp
// 最后一次失败时执行
public async Task<ConsumerState> FallbackAsync(MessageHeader messageHeader, TestEvent? message, Exception? ex)
{
	return ConsumerState.Ack;
}
```

--------------------------------

### 自定义事件中间件基础实现

Source: https://docs.whuanle.cn/zh/maomi_mq/2.2.eventbus

实现 IEventMiddleware 接口以拦截事件并控制执行链路。next 委托用于触发后续的事件处理器。

```csharp
public class TestEventMiddleware : IEventMiddleware<TestEvent>
{
    public async Task ExecuteAsync(MessageHeader messageHeader,TestEvent message, EventHandlerDelegate<TestEvent> next)
    {
        await next(messageHeader, message, CancellationToken.None);
    }
    public Task FaildAsync(MessageHeader messageHeader, Exception ex, int retryCount, TestEvent? message) => Task.CompletedTask;
    public Task<ConsumerState> FallbackAsync(MessageHeader messageHeader, TestEvent? message, Exception? ex) => Task.FromResult(ConsumerState.Ack);
}
```

--------------------------------

### 配置消息过期时间

Source: https://docs.whuanle.cn/zh/maomi_mq/2.publisher

通过 BasicProperties 配置单条消息的过期时间。

```csharp
await _messagePublisher.PublishAsync(exchange: string.Empty, routingKey: "publish", message: new TestEvent
{
	Id = i
}, (BasicProperties p) =>
{
	p.Expiration = "1000";
});
```

--------------------------------

### Implement Event Bus Pipeline

Source: https://docs.whuanle.cn/zh/maomi_mq/1.start.md

Define execution chains using IEventMiddleware and order steps with [EventOrder] for complex business workflows.

```csharp
using Maomi.MQ.EventBus;

[RouterKey("biz.order.pipeline.v1")]
public sealed class OrderPipelineEvent
{
    public Guid OrderId { get; set; } = Guid.NewGuid();

    public decimal Amount { get; set; }
}

[Consumer("biz.order.pipeline.v1")]
public sealed class OrderPipelineMiddleware : IEventMiddleware<OrderPipelineEvent>
{
    public Task ExecuteAsync(MessageHeader messageHeader, OrderPipelineEvent message, EventHandlerDelegate<OrderPipelineEvent> next)
    {
        return next(messageHeader, message, CancellationToken.None);
    }

    public Task FaildAsync(MessageHeader messageHeader, Exception ex, int retryCount, OrderPipelineEvent? message)
        => Task.CompletedTask;

    public Task<ConsumerState> FallbackAsync(MessageHeader messageHeader, OrderPipelineEvent? message, Exception? ex)
        => Task.FromResult(ConsumerState.Ack);
}

[EventOrder(1)]
public sealed class ReserveInventoryHandler : IEventHandler<OrderPipelineEvent>
{
    public Task ExecuteAsync(OrderPipelineEvent message, CancellationToken cancellationToken)
        => Task.CompletedTask;

    public Task CancelAsync(OrderPipelineEvent message, CancellationToken cancellationToken)
        => Task.CompletedTask;
}

[EventOrder(2)]
public sealed class CreateBillHandler : IEventHandler<OrderPipelineEvent>
{
    public Task ExecuteAsync(OrderPipelineEvent message, CancellationToken cancellationToken)
        => Task.CompletedTask;

    public Task CancelAsync(OrderPipelineEvent message, CancellationToken cancellationToken)
        => Task.CompletedTask;
}
```

--------------------------------

### 实现消费者

Source: https://docs.whuanle.cn/zh/maomi_mq/1.start.md

实现 IConsumer 接口并使用 [Consumer] 特性注解来定义消费者逻辑。

```csharp
[Consumer("test")]
public class MyConsumer : IConsumer<TestEvent>
{
    // 消费
    public async Task ExecuteAsync(MessageHeader messageHeader, TestEvent message)
    {
        Console.WriteLine($"事件 id: {message.Id} {DateTime.Now}");
        await Task.CompletedTask;
    }

    // 每次消费失败时执行
    public Task FaildAsync(MessageHeader messageHeader, Exception ex, int retryCount, TestEvent message) 
        => Task.CompletedTask;

    // 补偿
    public Task<ConsumerState> FallbackAsync(MessageHeader messageHeader, TestEvent? message, Exception? ex) 
        => Task.FromResult( ConsumerState.Ack);
}
```

--------------------------------

### Enable Publisher Confirmations

Source: https://docs.whuanle.cn/zh/maomi_mq/2.publisher

Create a message publisher with confirmation tracking enabled to ensure reliable message delivery.

```csharp
using var confirmPublisher = _messagePublisher.CreateSingle(
	new CreateChannelOptions(publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true));

for (var i = 0; i < 5; i++)
{
	await confirmPublisher.PublishAsync(exchange: string.Empty, routingKey: "publish", message: new TestEvent
	{
		Id = 666
	});
}
```

--------------------------------

### Register Maomi.MQ.Transaction Services (ADO.NET)

Source: https://docs.whuanle.cn/zh/maomi_mq/8.transcation.md

Register the Maomi.MQ.Transaction services with the dependency injection container. Configure the database provider, connection string, and whether to auto-create tables. Note: `AutoCreateTable=true` is suitable for development; production environments should use migration tools or DBA scripts.

```csharp
using Maomi.MQ.Transaction.Mysql;
using MySqlConnector;

builder.Services.AddMaomiMQTransactionMySql();

builder.Services.AddMaomiMQTransaction(options =>
{
    options.ProviderName = "mysql";
    options.Connection = _ => new MySqlConnection(connectionString);
    options.AutoCreateTable = true;
});

builder.Services.AddMaomiMQ(
    (MqOptionsBuilder options) =>
    {
        // RabbitMQ 配置
    },
    [typeof(Program).Assembly],
    Maomi.MQ.Extensions.CreateTransactionFilters());

```

--------------------------------

### IRoutingProvider with Caching Optimization

Source: https://docs.whuanle.cn/zh/maomi_mq/3.configuration.md

An optimized implementation of `IRoutingProvider` using `ConcurrentDictionary` for local caching to reduce allocation and GC pressure in high-throughput scenarios.

```APIDOC
## IRoutingProvider with Caching Optimization

### Description
An optimized `IRoutingProvider` implementation that utilizes local caching with `ConcurrentDictionary` to improve performance by reducing object allocations and GC pressure.

### Method
```csharp
services.AddSingleton<IRoutingProvider, CachedRoutingProvider>();
```

### Endpoint
N/A (This is a service registration, not an HTTP endpoint)

### Parameters
None

### Request Example
```csharp
// Service registration in Startup.cs or Program.cs
builder.Services.AddSingleton<IRoutingProvider, CachedRoutingProvider>();
```

### Response
#### Success Response (200)
N/A

#### Response Example
```csharp
using Maomi.MQ.Consumer;
using System.Collections.Concurrent;

public sealed class CachedRoutingProvider : IRoutingProvider
{
    private readonly string _prefix;
    private readonly ConcurrentDictionary<string, IConsumerOptions> _consumerCache = new();
    private readonly ConcurrentDictionary<string, IRouterKeyOptions> _publishCache = new();

    public CachedRoutingProvider(IHostEnvironment env)
    {
        _prefix = env.IsDevelopment() ? "dev." : "prod.";
    }

    public IConsumerOptions Get(IConsumerOptions options)
    {
        var key = $"{options.Queue}|{options.BindExchange}|{options.RoutingKey}|{options.ExchangeType}|{options.IsBroadcast}";
        return _consumerCache.GetOrAdd(key, _ =>
        {
            var newOptions = new ConsumerOptions();
            newOptions.CopyFrom(options);
            newOptions.Queue = _prefix + options.Queue;
            return newOptions;
        });
    }

    public IRouterKeyOptions Get(IRouterKeyOptions options)
    {
        var key = $"{options.Exchange}|{options.RoutingKey}";
        return _publishCache.GetOrAdd(key, _ =>
            new RouterKeyOptions(_prefix + options.RoutingKey, options.Exchange));
    }

    private sealed class RouterKeyOptions : IRouterKeyOptions
    {
        public RouterKeyOptions(string routingKey, string? exchange)
        {
            RoutingKey = routingKey;
            Exchange = exchange;
        }

        public string RoutingKey { get; }
        public string? Exchange { get; }
    }
}
```
```

--------------------------------

### Configure Custom Serializer

Source: https://docs.whuanle.cn/zh/maomi_mq/1.start.md

Register alternative serializers like Protobuf in the Maomi.MQ options to optimize message transmission performance.

```csharp
using ProtoBuf;

builder.Services.AddMaomiMQ(options =>
{
    options.MessageSerializers = serializers =>
    {
        // 添加 Protobuf 序列化器
        serializers.Insert(0, new ProtobufMessageSerializer());
    };
}, [typeof(Program).Assembly]);

[ProtoContract]
public sealed class PersonMessage
{
    [ProtoMember(1)]
    public Guid Id { get; set; } = Guid.NewGuid();

    [ProtoMember(2)]
    public string Name { get; set; } = string.Empty;

    [ProtoMember(3)]
    public int Age { get; set; }
}
```

--------------------------------

### Configure MySQL for Maomi.MQ Transactions

Source: https://docs.whuanle.cn/zh/maomi_mq/1.start

Set up MySQL as the transaction provider for Maomi.MQ. Ensure the connection string is correctly configured and set AutoCreateTable to true if the table does not exist.

```csharp
using Maomi.MQ.Transaction.Mysql;
using MySqlConnector;

builder.Services.AddMaomiMQTransactionMySql();
builder.Services.AddMaomiMQTransaction(options =>
{
    options.ProviderName = TransactionProviderNames.MySql;
    options.Connection = _ => new MySqlConnection(builder.Configuration.GetConnectionString("Default"));
    options.AutoCreateTable = true;
});

```
Tokens:
Raw