mgzhenhong.ASW2.MessageQueue 2.1.3

ASW.MessageQueue

ASW.MessageQueue 是一个基于 .NET 8.0 的消息队列系统，提供了完整的消息发布/订阅功能。该系统基于 RDC（Request-response Data Communication）协议实现客户端与服务端之间的通信。

功能特性

• 消息发布/订阅模式：支持消息的发布和订阅功能
• 多种消息分发策略：支持忽略、精确匹配和通配符匹配三种分发模式
• 消息确认机制：支持消息确认、拒绝和丢弃操作
• 消息过期策略：支持按时间和按数量两种消息过期策略
• 高可用性设计：支持自动重连、消息持久化等特性
• 灵活的队列管理：支持动态创建和删除消息队列

核心组件

MessageQueueServer

消息队列服务端类，负责接收客户端连接、处理消息发布、订阅、确认和拒绝等操作。

主要功能：

• 接收和处理客户端连接
• 管理消息存储和分发
• 处理消息发布、订阅、确认、拒绝等操作
• 定时清理过期消息
• 处理超时消息

MessageQueueClient

消息队列客户端类，提供消息队列的客户端功能，包括消息发布、订阅、消费、确认和拒绝等操作。

主要功能：

• 连接消息队列服务端
• 发布消息到指定队列
• 订阅指定队列的消息
• 确认、拒绝或丢弃接收到的消息

IStorage

消息存储接口，定义了消息存储的基本操作。

IStorageFactory

消息存储工厂接口，用于创建和管理消息存储实例。

安装和使用

环境要求

• .NET 8.0 或更高版本
• 支持的IDE：Visual Studio 2022 或更高版本

安装步骤

1. 克隆项目到本地
2. 使用 Visual Studio 打开解决方案
3. 恢复 NuGet 包
4. 编译项目

基本使用

启动服务端

// 创建存储工厂
var storageFactory = new SqliteStorageFactory();

// 创建并启动服务端
var server = new MessageQueueServer(storageFactory, 2218);
server.Start();

// 等待服务端就绪
while (!server.IsReady) {
    await Task.Delay(100);
}

创建客户端

// 创建客户端并连接到服务端
var client = new MessageQueueClient("127.0.0.1", 2218);
client.Connect();

// 等待连接成功
while (client.State != TcpClientState.CONNECTED) {
    await Task.Delay(100);
}

发布消息

// 声明队列
await client.DeclareQueue("test_queue", 
    new MessageDistributePolicy(), 
    new MessageExpirationPolicy());

// 发布消息
await client.Publish("test_queue", "Hello, World!");

订阅消息

// 订阅队列消息
await client.Subscribe("test_queue", (message) => {
    Console.WriteLine($"收到消息: {message.MessageContent}");
    
    // 确认消息处理完成
    await client.ConfirmMessage(message);
});

架构设计

系统架构

+------------------+     +------------------+     +------------------+
|   客户端应用      |     |   消息队列服务端  |     |   消息存储层      |
|  MessageQueueClient |<--->|  MessageQueueServer |<--->|   IStorage      |
+------------------+     +------------------+     +------------------+

消息流

1. 客户端通过 RDC 协议连接到服务端
2. 客户端发布消息到服务端
3. 服务端将消息存储到存储层
4. 服务端根据队列的分发策略将消息分发到相应队列
5. 订阅了队列的客户端接收到消息
6. 客户端处理消息并发送确认、拒绝或丢弃指令

配置选项

服务端配置

• MessageExpirationCheckInterval：消息过期检查间隔（秒）
• TimeoutCheckInterval：超时检查间隔（秒）
• MessageConsumeTimeoutSeconds：消息消费超时时间（秒）
• ClientClearQueueIsEnabled：是否允许客户端创建队列
• ClientSetNameIsRequired：客户端是否必须设置名称

客户端配置

• AutoReconnectIsEnabled：连接断开后是否自动重连
• AutoReconnectDelayMs：自动重连前延时（毫秒）

贡献指南

欢迎提交 Issue 和 Pull Request 来改进这个项目。

许可证

本项目采用 MIT 许可证。