mgzhenhong.ASW2.MessageQueue.SqliteStorage 2.1.2

ASW.MessageQueue.SqliteStorage

ASW.MessageQueue.SqliteStorage 是 ASW.MessageQueue 消息队列系统的 SQLite 存储实现。它基于 SqlSugar ORM 提供消息的持久化存储和管理功能。

功能特性

• SQLite 数据库存储：使用 SQLite 作为消息存储介质，确保数据持久化
• 基于 SqlSugar ORM：利用 SqlSugar 简化数据库操作
• 消息持久化：确保消息在系统重启后不会丢失
• 队列管理：支持队列的创建、删除和管理
• 消息分发：实现消息到队列的分发机制
• 过期策略：支持消息的过期清理功能

核心组件

SqliteMessageStorage

SQLite 消息存储实现类，实现了 IStorage 接口。

主要功能：

• 消息的存储和检索
• 队列的定义和管理
• 消息分发到队列
• 消息确认、拒绝和丢弃处理
• 过期消息清理
• 消费中消息的超时处理

SqliteStorageFactory

SQLite 存储工厂类，实现了 IStorageFactory 接口。

主要功能：

• 创建和管理 SQLite 存储实例
• 支持多个存储区的管理

QueueDistributeState

队列分发状态类，负责消息分发到队列的逻辑。

主要功能：

• 管理队列分发状态
• 实现消息到队列的分发机制
• 使用线程持续分发消息

数据库设计

存储消息表 (StorageMessageEntity)

存储所有发布的消息，包含以下字段：

• Uid: 消息唯一标识
• RoutingKey: 路由键
• Content: 消息内容
• PublishTime: 发布时间
• Producer: 生产者

队列表 (QueueEntity)

存储队列信息，包含以下字段：

• Name: 队列名称
• DistributePolicy: 分发策略
• ExpirationPolicy: 过期策略
• CreateTime: 创建时间
• LastDistributedMessageUid: 最后分发的消息UID

队列消息表 (QueueMessageEntity)

每个队列对应一个消息表，存储队列中的消息，包含以下字段：

• Uid: 消息唯一标识
• StorageMessageUid: 存储消息UID
• Producer: 生产者
• RoutingKey: 路由键
• Content: 消息内容
• PublishTime: 发布时间
• State: 消息状态（空闲、消费中、已确认、已拒绝、已丢弃）
• Consumer: 消费者
• ConfirmTime: 确认时间
• RejectedCount: 拒绝次数
• DropReason: 丢弃原因
• DropTime: 丢弃时间
• EnqueueTime: 入队时间

安装和使用

环境要求

• .NET 8.0 或更高版本
• SQLite 数据库
• SqlSugar ORM

安装步骤

1. 确保已安装 ASW.MessageQueue 项目
2. 克隆项目到本地
3. 使用 Visual Studio 打开解决方案
4. 恢复 NuGet 包
5. 编译项目

基本使用

创建存储工厂

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

// 声明存储区
var storageResult = storageFactory.DeclearStorage("default");
if (storageResult) {
    var storage = storageResult.Data;
    // 设置存储区过期策略
    storage.ExpirationPolicy = new MessageExpirationPolicy {
        ExpirationType = MessageExpirationType.Never
    };
}

在服务端使用 SQLite 存储

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

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

配置选项

存储配置

• 数据库文件路径：指定 SQLite 数据库文件的存储位置
• 过期策略：设置存储区和队列的消息过期策略

性能优化

• 分发线程：每个队列使用独立的分发线程
• 批量操作：支持消息的批量存储和更新

架构设计

类关系图

classDiagram
    IStorage <|-- SqliteMessageStorage
    IStorageFactory <|-- SqliteStorageFactory
    SqliteMessageStorage --> QueueDistributeState
    SqliteMessageStorage --> DbHelper
    SqliteMessageStorage --> StoragePropertyManager
    
    class IStorage {
        <<interface>>
        +StoreMessage()
        +DeclearQueue()
        +FetchQueueMessages()
        +ConfirmMessage()
        +RejectMessage()
        +DropMessage()
    }
    
    class IStorageFactory {
        <<interface>>
        +DeclearStorage()
        +GetStorages()
        +TryGetStorage()
    }
    
    class SqliteMessageStorage {
        -queueDistributeStateDic
        -dbHelper
        -propertyManager
        +StartDistribute()
        +CleanExpiredStorageMessages()
        +ResetAllConsumingMessagesToIdle()
    }
    
    class SqliteStorageFactory {
        -storageDic
        +DeclearStorage()
        +GetStorages()
    }
    
    class QueueDistributeState {
        -messageForDistributeList
        -dbHelper
        +DistributeMessage()
    }

贡献指南

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

许可证

本项目采用 MIT 许可证。