SAEA.QueueSocket 26.4.23.1

SAEA.QueueSocket - 高性能内存消息队列 📮

English Version | 中文版

基于 SAEA.Sockets 的高性能内存消息队列，采用发布/订阅模式，支持万级并发消息分发。

快速导航 🧭

30秒快速开始 ⚡

最快上手方式，3 步即可运行消息队列系统：

Step 1: 安装 NuGet 包

dotnet add package SAEA.QueueSocket

Step 2: 启动队列服务器

using SAEA.QueueSocket;

var server = new QServer(port: 39654);
server.Start();
Console.WriteLine("消息队列服务器已启动!");

Step 3a: 创建生产者发布消息

var producer = new Producer("producer_1", "127.0.0.1:39654");
producer.Connect();
producer.Publish("orders", "订单数据: {id: 123}");

Step 3b: 创建消费者订阅消息

var consumer = new Consumer("consumer_1", "127.0.0.1:39654");
consumer.OnMessage += (msg) => 
    Console.WriteLine($"收到: {Encoding.UTF8.GetString(msg.Data)}");
consumer.Connect();
consumer.Subscribe("orders");

就这么简单！ 🎉 你已经实现了一个高性能的发布/订阅消息队列系统。

核心特性 🎯

架构设计 📐

Pub/Sub 架构图

┌─────────────────────────────────────────────────────────────┐
│                  SAEA.QueueSocket 架构                       │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌────────────┐                    ┌────────────┐           │
│  │  Producer  │                    │  Producer  │           │
│  │  (生产者)  │                    │  (生产者)  │           │
│  └─────┬──────┘                    └─────┬──────┘           │
│        │                                 │                  │
│        │ Publish(topic, msg)             │                  │
│        │                                 │                  │
│        └──────────────┬──────────────────┘                  │
│                       │                                     │
│                       ▼                                     │
│              ┌─────────────────┐                            │
│              │     QServer     │                            │
│              │   (队列服务器)   │                            │
│              └────────┬────────┘                            │
│                       │                                     │
│         ┌─────────────┼─────────────┐                       │
│         │             │             │                        │
│         ▼             ▼             ▼                        │
│   ┌──────────┐ ┌──────────┐ ┌──────────┐                    │
│   │ Exchange │ │ Exchange │ │ Exchange │                    │
│   │ (orders) │ │  (news)  │ │ (logs)   │                    │
│   └────┬─────┘ └────┬─────┘ └────┬─────┘                    │
│        │            │            │                           │
│        ▼            ▼            ▼                           │
│   ┌─────────┐ ┌─────────┐ ┌─────────┐                       │
│   │  Topic  │ │  Topic  │ │  Topic  │                       │
│   │  Queue  │ │  Queue  │ │  Queue  │                       │
│   └────┬────┘ └────┬────┘ └────┬────┘                       │
│        │            │            │                           │
│        ▼            ▼            ▼                           │
│   ┌────────┐  ┌────────┐  ┌────────┐                        │
│   │Consumer│  │Consumer│  │Consumer│                        │
│   │(消费者)│  │(消费者)│  │(消费者)│                        │
│   └────────┘  └────────┘  └────────┘                        │
│                                                             │
└─────────────────────────────────────────────────────────────┘

消息流转流程

消息发布流程:

Producer ──► Publish(topic, msg) ──► QServer 接收
                                           │
                                           ▼
                                    ┌─────────────┐
                                    │   Exchange   │
                                    │  路由分发    │
                                    └─────────────┘
                                           │
                                           ▼
                                    ┌─────────────┐
                                    │ MessageQueue│
                                    │  入队存储   │
                                    └─────────────┘
                                           │
                                           ▼
                                    ┌─────────────┐
                                    │  Topic 匹配 │
                                    │  订阅者列表 │
                                    └─────────────┘
                                           │
                        ┌──────────────────┼──────────────────┐
                        │                  │                  │
                        ▼                  ▼                  ▼
                   Consumer A        Consumer B        Consumer C
                   OnMessage()       OnMessage()       OnMessage()


消费者订阅流程:

Consumer ──► Connect() ──► QServer 建立连接
                                  │
                                  ▼
                           ┌─────────────┐
                           │  Subscribe  │
                           │  (topic)    │
                           └─────────────┘
                                  │
                                  ▼
                           ┌─────────────┐
                           │  Exchange   │
                           │  注册订阅   │
                           └─────────────┘
                                  │
                                  ▼
                           等待消息推送
                                  │
                                  ▼
                           OnMessage() 触发

应用场景 💡

✅ 适合使用 SAEA.QueueSocket 的场景

❌ 不适合的场景

性能对比 📊

与 Redis Pub/Sub 对比

性能测试数据

💡 提示: 测试环境为 Intel i7-10700, 32GB RAM, .NET 6.0

常见问题 ❓

Q1: SAEA.QueueSocket 与 Redis Pub/Sub 有什么区别？

A: 主要区别：

• 部署: SAEA.QueueSocket 无需外部服务，内嵌于应用；Redis 需要单独部署
• 延迟: SAEA.QueueSocket 为内存直连，延迟更低（~0.5ms vs ~2-5ms）
• 持久化: 两者均为内存队列，重启后消息丢失
• 适用场景: SAEA.QueueSocket 适合单机高性能场景；Redis 适合分布式场景

Q2: 如何保证消息不丢失？

A: SAEA.QueueSocket 设计为高性能内存队列，消息存储在内存中。如需消息持久化：

• 使用 RabbitMQ、Kafka 等支持持久化的消息队列
• 或在业务层实现消息确认和重发机制

Q3: Topic 数量有限制吗？

A: Topic 数量没有硬性限制，由系统内存决定。建议：

• 单 Topic 订阅者数量不超过 100
• Topic 总数不超过 1000
• 定期清理无用的 Topic

Q4: 如何处理消费者断线重连？

A: Consumer 支持事件通知：

consumer.OnDisconnected += () => 
{
    Console.WriteLine("连接断开，正在重连...");
    Task.Delay(1000).Wait();
    consumer.Connect();
    consumer.Subscribe("orders");
};

Q5: 消息堆积如何处理？

A:

• QServer 支持配置最大消息堆积数量
• 监控 CalcInfo 回调获取队列状态
• 增加消费者数量提高消费速度
• 使用 Batcher 批量处理提升效率

Q6: 支持消息过滤吗？

A: 支持 Topic 级别的消息过滤：

• 消费者只订阅感兴趣的 Topic
• 一个消费者可订阅多个 Topic
• 使用 Unsubscribe() 取消订阅

Q7: 如何查看服务器状态？

A: 使用 CalcInfo 方法：

server.CalcInfo((sessionCount, topicCount, queuedCount) => 
{
    Console.WriteLine($"连接数: {sessionCount}");
    Console.WriteLine($"Topic数: {topicCount}");
    Console.WriteLine($"待分发消息: {queuedCount}");
});

核心类 🔧

使用示例 📝

完整服务器示例

using SAEA.QueueSocket;

var server = new QServer(port: 39654, maxConnections: 1000, maxQueued: 10000);
server.Start();

server.CalcInfo((sessionCount, topicCount, queuedCount) => 
{
    Console.WriteLine($"[统计] 连接: {sessionCount}, Topic: {topicCount}, 队列: {queuedCount}");
});

Console.WriteLine("队列服务器已启动，端口: 39654");
Console.ReadLine();

完整生产者示例

using SAEA.QueueSocket;

var producer = new Producer("producer_001", "127.0.0.1:39654");

producer.OnError += (ex) => Console.WriteLine($"错误: {ex.Message}");
producer.OnDisconnected += () => Console.WriteLine("连接断开");

producer.Connect();

while (true)
{
    var message = Console.ReadLine();
    if (message == "exit") break;
    
    producer.Publish("orders", message);
    Console.WriteLine("消息已发布");
}

producer.Close();

完整消费者示例

using SAEA.QueueSocket;
using System.Text;

var consumer = new Consumer("consumer_001", "127.0.0.1:39654");

consumer.OnMessage += (msg) => 
{
    var content = Encoding.UTF8.GetString(msg.Data);
    Console.WriteLine($"[收到] Topic: {msg.Topic}, 消息: {content}");
};

consumer.OnError += (ex) => Console.WriteLine($"错误: {ex.Message}");
consumer.OnDisconnected += () => 
{
    Console.WriteLine("连接断开，5秒后重连...");
    Task.Delay(5000).Wait();
    consumer.Connect();
    consumer.Subscribe("orders");
};

consumer.Connect();
consumer.Subscribe("orders");
consumer.Subscribe("notifications");

Console.WriteLine("消费者已启动，等待消息...");
Console.ReadLine();

consumer.Close();

使用 QClient 基类（同时发布和订阅）

using SAEA.QueueSocket;
using System.Text;

var client = new QClient("client_001", "127.0.0.1:39654");

client.OnMessage += (msg) => 
    Console.WriteLine($"收到: {Encoding.UTF8.GetString(msg.Data)}");
client.OnError += (ex) => Console.WriteLine($"错误: {ex.Message}");
client.OnDisconnected += () => Console.WriteLine("断开连接");

client.Connect();

client.Subscribe("topic1");
client.Publish("topic2", "Hello from client");

client.Unsubscribe("topic1");
client.Close();

消息类型

public enum QueueSocketMsgType : byte
{
    Ping = 1,       // 心跳请求
    Pong = 2,       // 心跳响应
    Publish = 3,    // 发布消息
    Subcribe = 4,   // 订阅请求
    Unsubcribe = 5, // 取消订阅
    Close = 6,      // 关闭连接
    Data = 7        // 数据消息
}

Topic 特性

依赖项

更多资源

• GitHub 仓库
• NuGet 包
• 作者博客

许可证

Apache License 2.0