IMq 接口 - Snet Docs
文档/API 参考/IMq 接口

📗 IMq 接口

命名空间: Snet.Core | 接口: IMq | 组合: 12 个接口 (IOn, IOff, IProducer, IConsumer, IGetStatus, IEvent, IGetParam, ICreateInstance, ILog, ILanguage, IDisposable, IAsyncDisposable)

IMq 是所有消息队列（中间件）组件的统一接口。它将连接生命周期、生产者、消费者、事件、日志和语言支持组合成一个统一的契约。

接口组成

public interface IMq : IOn, IOff, IProducer, IConsumer, IGetStatus, IEvent,
    IGetParam, ICreateInstance, ILog, ILanguage, IDisposable, IAsyncDisposable
{
}

IProducer

方法 类型 描述
Produce(string topic, string content, Encoding? encoding = null) 同步包装 向主题发布字符串消息（同步）
Produce(string topic, byte[] content) 同步包装 向主题发布原始字节数据（同步）
ProduceAsync(string topic, string content, Encoding? encoding = null, CancellationToken token = default) 虚方法 默认编码为 UTF8 → 字节后委托给抽象字节重载
ProduceAsync(string topic, byte[] content, CancellationToken token = default) 抽象方法 子类必须实现的发布方法

IConsumer

方法 描述
Consume(string topic) 订阅主题（同步）
ConsumeAsync(string topic, CancellationToken token = default) 订阅主题并开始接收消息（异步）
UnConsume(string topic) 取消订阅主题（同步）
UnConsumeAsync(string topic, CancellationToken token = default) 取消订阅主题并停止接收消息（异步）

快速入门

using Snet.Core;
using Snet.Mqtt;

IMq mq = new MqttClientOperate(new MqttClientData.Basics
{
    IpAddress = "127.0.0.1",
    Port = 1883,
    UserName = "shunnet",
    Password = "shunnet"
});

// 连接 (IOn 是 IMq 的一部分)
await mq.OnAsync();

// 发布字符串消息
await mq.ProduceAsync("sensors/temp", "25.3 °C", System.Text.Encoding.UTF8);

// 发布原始字节数据
byte[] payload = BitConverter.GetBytes(25.3f);
await mq.ProduceAsync("sensors/temp", payload);

// 订阅主题
await mq.ConsumeAsync("sensors/temp");

// 取消订阅
await mq.UnConsumeAsync("sensors/temp");

// 断开连接 (IOff 是 IMq 的一部分)
await mq.OffAsync();

发布消息

ProduceAsync 有两个重载：

字符串消息

// 使用显式编码
await mq.ProduceAsync("topic/test", "Hello World", System.Text.Encoding.UTF8);

// UTF8 是文本消息最常用的编码
await mq.ProduceAsync("topic/json", "{\"id\": 1, \"value\": 42}", System.Text.Encoding.UTF8);

原始字节数据

// 直接发送二进制数据
byte[] rawBytes = new byte[] { 0x01, 0x02, 0x03, 0x04 };
await mq.ProduceAsync("topic/binary", rawBytes);

// 将结构化数据以字节形式发送
var data = new { sensorId = 5, temperature = 23.7 };
string json = System.Text.Json.JsonSerializer.Serialize(data);
byte[] jsonBytes = System.Text.Encoding.UTF8.GetBytes(json);
await mq.ProduceAsync("sensors/data", jsonBytes);

消费消息

// 开始消费某个主题的消息
await mq.ConsumeAsync("topic/test");

// 消息通过 OnDataEvent 到达
mq.OnDataEventAsync += async (sender, e) =>
{
    if (e.Status)
        Console.WriteLine($"收到: {e.ResultData}");
};

// 停止消费
await mq.UnConsumeAsync("topic/test");

实现

所有中间件类都实现了 IMq：

类 包 协议
MqttClientOperate Snet.Mqtt MQTT 3.1.1
MqttServiceOperate Snet.Mqtt 嵌入式 MQTT Broker
MqttWebSocketServiceOperate Snet.Mqtt MQTT over WebSocket
KafkaOperate Snet.Kafka Apache Kafka
RabbitMQOperate Snet.RabbitMQ AMQP 0-9-1
NetMQOperate Snet.NetMQ ZeroMQ
NettyClientOperate Snet.Netty 原始 TCP (DotNetty)

参见