feat:增加项目实战模块
This commit is contained in:
56
docs/Web/Network/Mqtt.md
Normal file
56
docs/Web/Network/Mqtt.md
Normal file
@@ -0,0 +1,56 @@
|
||||
---
|
||||
title: Mqtt简介与使用
|
||||
date: 2026-05-27
|
||||
---
|
||||
|
||||
# 一、简介
|
||||
  MQTT（Message Queuing Telemetry Transport） 是一种 轻量级物联网通信协议，基于 发布 / 订阅（Pub/Sub） 模式。
|
||||
|
||||
## 1.1 核心概念
|
||||
1. Broker：消息服务器
|
||||
2. Client：发布或订阅消息的设备 / 程序
|
||||
3. Topic: 消息主题（路径形式）
|
||||
4. Publish：发布消息
|
||||
5. Subscribe：订阅消息
|
||||
6. QoS：服务质量等级（0 / 1 / 2）
|
||||
|
||||
## 1.2 工作方式
|
||||
```text
|
||||
[设备A] --发布--> TopicA --> [Broker] --> TopicA --> [设备B]
|
||||
```
|
||||
|
||||
# 二、安装
|
||||
  [EMQX](https://www.emqx.com/zh) Enterprise 是基于开源 EMQX 扩展的企业级 MQTT 物联网消息接入平台，完全兼容 MQTT v3.1.1 / v5.0。
|
||||
|
||||
## 2.1 docker安装
|
||||
```yml
|
||||
services:
|
||||
emqx:
|
||||
image: emqx/emqx-enterprise:5.10.0
|
||||
container_name: emqx
|
||||
ports:
|
||||
- "1883:1883"
|
||||
- "8083:8083"
|
||||
- "8084:8084"
|
||||
- "8883:8883"
|
||||
- "18083:18083"
|
||||
restart: always
|
||||
```
|
||||
|
||||
  启动容器后，输入ip:18083，默认账号密码为admin/public。
|
||||
|
||||
# 三、消息规范
|
||||
## 3.1 Topic
|
||||
```text
|
||||
<domain>/<product>/<device_id>/<msg_type>/<resource>
|
||||
```
|
||||
|
||||
  例如`iot/thermostat/dev001/status/online`、`iot/thermostat/dev001/cmd/set_mode`。
|
||||
|
||||
## 通配符
|
||||
1. +：单层通配符，例如：iot/+/dev001/telemetry/+
|
||||
2. #：多层通配符，只能放在topic末尾，例如：iot/thermostat/dev001/#
|
||||
|
||||
::: tip
|
||||
使用+通配符时，必须严格满足层级要求，例如，如果topic是5层，那么iot/+将无法匹配到消息
|
||||
:::
|
||||
311
docs/Web/Network/NetCore.md
Normal file
311
docs/Web/Network/NetCore.md
Normal file
@@ -0,0 +1,311 @@
|
||||
---
|
||||
title: .Net Core实现
|
||||
date: 2026-05-27
|
||||
---
|
||||
|
||||
# 一、.Net Core实现
|
||||
## 1.1 SuperSocket
|
||||
  [SuperSocket](https://www.supersocket.net/)是一个轻量级, 跨平台而且可扩展的 .Net/Mono Socket 服务器程序框架。可以轻松构建TCP、UDP、WebSocket服务器。
|
||||
|
||||
## 1.2 安装依赖
|
||||
  NuGut安装SuperSocket、SuperSocket.WebSocket和SuperSocket.WebSocket.Server 2.0及以上版本。
|
||||
|
||||
## 1.3 配置文件
|
||||
  appsettings.json
|
||||
```json
|
||||
{
|
||||
"serverOptions": {
|
||||
"TcpServer": {
|
||||
"name": "TcpServer",
|
||||
"listeners": [
|
||||
{
|
||||
"ip": "Any",
|
||||
"port": 4040
|
||||
}
|
||||
]
|
||||
},
|
||||
"WebSocketServer": {
|
||||
"name": "WebSocket",
|
||||
"listeners": [
|
||||
{
|
||||
"ip": "Any",
|
||||
"port": 5050
|
||||
}
|
||||
]
|
||||
}
|
||||
},
|
||||
"Mqtt": {
|
||||
"Host": "127.0.0.1",
|
||||
"Port": 1883,
|
||||
"ClientId": "MqttClient",
|
||||
"Topics": [
|
||||
"test/topic1",
|
||||
"test/topic2"
|
||||
]
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## 1.4 主程序
|
||||
  program.cs
|
||||
```cs
|
||||
var host = Host.CreateDefaultBuilder(args)
|
||||
.ConfigureServices((context, services) =>
|
||||
{
|
||||
services.AddSingleton<IPackageHandler<ProtocolFrame>, TcpPackageHandler>();
|
||||
services.AddSingleton<WebSocketMessageHandler>();
|
||||
|
||||
// 1. 注入 MQTT 配置
|
||||
services.Configure<MqttSettings>(context.Configuration.GetSection("Mqtt"));
|
||||
// 2. 注册 MQTT 客户端
|
||||
services.AddSingleton<IMqttClient>(serviceProvider => new MqttClientFactory().CreateMqttClient());
|
||||
// 3. 注册 MQTT 后台服务
|
||||
services.AddHostedService<MqttService>();
|
||||
})
|
||||
.AsMultipleServerHostBuilder()
|
||||
.AddServer<TcpService, ProtocolFrame, BinaryPipelineFilter>(builder =>
|
||||
{
|
||||
builder.ConfigureServerOptions((ctx, config) => config.GetSection("TcpServer"));
|
||||
})
|
||||
.AddWebSocketServer(builder =>
|
||||
{
|
||||
builder
|
||||
.UseWebSocketMessageHandler(async (session, package) =>
|
||||
{
|
||||
using var scope = session.Server.ServiceProvider.CreateScope();
|
||||
var handler = scope.ServiceProvider.GetRequiredService<WebSocketMessageHandler>();
|
||||
|
||||
await handler.HandleAsync(session, package);
|
||||
})
|
||||
.ConfigureServerOptions((ctx, config) => config.GetSection("WebSocketServer"));
|
||||
})
|
||||
.ConfigureLogging(logging => logging.ClearProviders())
|
||||
.UseNLog()
|
||||
.Build();
|
||||
|
||||
await host.RunAsync();
|
||||
```
|
||||
|
||||
  通过.AsMultipleServerHostBuilder()可以构造多服务器实例。
|
||||
|
||||
# 二、 TCP服务器
|
||||
## 2.1 协议数据
|
||||
```cs
|
||||
/// <summary>
|
||||
/// 协议数据结构 示例：65 6D 00 05 00 01 68 65 6C 6C 6F
|
||||
/// </summary>
|
||||
/// <param name="Magic">帧头标识</param>
|
||||
/// <param name="Length">数据长度</param>
|
||||
/// <param name="Type">数据类型</param>
|
||||
/// <param name="Payload">数据内容</param>
|
||||
///
|
||||
record ProtocolFrame(
|
||||
ushort Magic,
|
||||
ushort Length,
|
||||
ushort Type,
|
||||
byte[] Payload
|
||||
);
|
||||
```
|
||||
|
||||
  可以根据实际情况自定义消息格式。
|
||||
|
||||
## 2.2 协议解析
|
||||
```cs
|
||||
class BinaryPipelineFilter : FixedHeaderPipelineFilter<ProtocolFrame>
|
||||
{
|
||||
/// <summary>
|
||||
/// 固定头长度
|
||||
/// </summary>
|
||||
public BinaryPipelineFilter() : base(6)
|
||||
{
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 从包头中解析出 Body 长度
|
||||
/// </summary>
|
||||
/// <param name="buffer">字节流</param>
|
||||
/// <returns>Body 长度</returns>
|
||||
protected override int GetBodyLengthFromHeader(ref ReadOnlySequence<byte> buffer)
|
||||
{
|
||||
var reader = new SequenceReader<byte>(buffer);
|
||||
|
||||
// 读取前2字节 → magic
|
||||
reader.TryReadBigEndian(out ushort magic);
|
||||
|
||||
// 再读2字节 → Length
|
||||
reader.TryReadBigEndian(out ushort length);
|
||||
|
||||
// 再读2字节 → Type
|
||||
reader.TryReadBigEndian(out ushort type);
|
||||
|
||||
// 校验帧头
|
||||
if (magic != 0x656D)
|
||||
{
|
||||
throw new Exception("非法帧头");
|
||||
}
|
||||
|
||||
// 限制长度
|
||||
if (length == 0 || length > 8192)
|
||||
{
|
||||
throw new Exception("非法长度");
|
||||
}
|
||||
|
||||
// 返回 Payload 长度
|
||||
return length;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 把完整字节包 → 转换成 ProtocolFrame
|
||||
/// </summary>
|
||||
/// <param name="buffer">字节流</param>
|
||||
/// <returns>ProtocolFrame</returns>
|
||||
protected override ProtocolFrame DecodePackage(ref ReadOnlySequence<byte> buffer)
|
||||
{
|
||||
var reader = new SequenceReader<byte>(buffer);
|
||||
|
||||
// 读取前2字节 → magic
|
||||
reader.TryReadBigEndian(out ushort magic);
|
||||
|
||||
// 再读2字节 → Length
|
||||
reader.TryReadBigEndian(out ushort length);
|
||||
|
||||
// 再读2字节 → Type
|
||||
reader.TryReadBigEndian(out ushort type);
|
||||
|
||||
var payload = buffer.Slice(6, length).ToArray();
|
||||
|
||||
// 构造ProtocolFrame
|
||||
return new ProtocolFrame(magic, length, type, payload);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
  根据协议数据ProtocolFrame来解包。
|
||||
|
||||
## 2.3 消息处理
|
||||
```cs
|
||||
class TcpPackageHandler(ILogger<TcpPackageHandler> logger) : IPackageHandler<ProtocolFrame>
|
||||
{
|
||||
private readonly ILogger<TcpPackageHandler> _logger = logger;
|
||||
|
||||
public async ValueTask Handle(IAppSession session, ProtocolFrame package, CancellationToken cancellationToken)
|
||||
{
|
||||
_logger.LogInformation($"Magic={package.Magic:X4}, Type={package.Type}, Len={package.Length}");
|
||||
await session.SendAsync(package.Payload, cancellationToken);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## 2.4 TCPService
|
||||
```cs
|
||||
class TcpService(IServiceProvider serviceProvider, IOptions<ServerOptions> serverOptions) : SuperSocketService<ProtocolFrame>(serviceProvider, serverOptions)
|
||||
{
|
||||
}
|
||||
```
|
||||
|
||||
# 三、Websocket服务器
|
||||
## 3.1 消息处理
|
||||
```cs
|
||||
class WebSocketMessageHandler(ILogger<WebSocketMessageHandler> logger)
|
||||
{
|
||||
private readonly ILogger<WebSocketMessageHandler> _logger = logger;
|
||||
|
||||
public async ValueTask HandleAsync(WebSocketSession session, WebSocketPackage package)
|
||||
{
|
||||
_logger.LogInformation($"[WebSocket] {package.Message}");
|
||||
await session.SendAsync("ok");
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
# 四、 MQTT服务器
|
||||
## 4.1 MqttService
|
||||
```cs
|
||||
class MqttService(IMqttClient mqttClient, IOptions<MqttSettings> settings, ILogger<TcpPackageHandler> logger) : BackgroundService
|
||||
{
|
||||
private readonly IMqttClient _mqttClient = mqttClient;
|
||||
private readonly MqttSettings _settings = settings.Value;
|
||||
private readonly ILogger<TcpPackageHandler> _logger = logger;
|
||||
private MqttClientOptions? _mqttOptions;
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
// 构建连接配置
|
||||
_mqttOptions = new MqttClientOptionsBuilder()
|
||||
.WithClientId($"{_settings.ClientId}_{Guid.NewGuid():N}")
|
||||
.WithTcpServer(_settings.Host, _settings.Port)
|
||||
.WithCleanSession()
|
||||
.Build();
|
||||
|
||||
_mqttClient.ConnectedAsync += OnConnectedAsync;
|
||||
_mqttClient.DisconnectedAsync += OnDisconnectedAsync;
|
||||
_mqttClient.ApplicationMessageReceivedAsync += HandleMessage;
|
||||
|
||||
_logger.LogInformation("[MQTT] 正在连接到服务器 {Host}:{Port}...", _settings.Host, _settings.Port);
|
||||
|
||||
await _mqttClient.ConnectAsync(_mqttOptions, stoppingToken);
|
||||
|
||||
// 等待程序停止
|
||||
await Task.Delay(Timeout.Infinite, stoppingToken);
|
||||
}
|
||||
|
||||
public async Task OnConnectedAsync(MqttClientConnectedEventArgs arg)
|
||||
{
|
||||
_logger.LogInformation("[MQTT] 已成功连接到服务器");
|
||||
|
||||
// 连接成功后订阅主题
|
||||
foreach (var topic in _settings.Topics)
|
||||
{
|
||||
await _mqttClient.SubscribeAsync(topic, MqttQualityOfServiceLevel.AtLeastOnce);
|
||||
_logger.LogInformation($"[MQTT] 已订阅主题：{topic}");
|
||||
}
|
||||
|
||||
await Task.CompletedTask;
|
||||
}
|
||||
|
||||
public async Task OnDisconnectedAsync(MqttClientDisconnectedEventArgs arg)
|
||||
{
|
||||
_logger.LogError("[MQTT] 连接失败，原因：{Reason}", arg.Reason);
|
||||
|
||||
await Task.CompletedTask;
|
||||
}
|
||||
|
||||
public async Task HandleMessage(MqttApplicationMessageReceivedEventArgs arg)
|
||||
{
|
||||
var topic = arg.ApplicationMessage.Topic;
|
||||
var payload = Encoding.UTF8.GetString(arg.ApplicationMessage.Payload);
|
||||
|
||||
_logger.LogInformation($"\n[MQTT] 收到消息\n主题：{topic}\n内容：{payload}\n");
|
||||
|
||||
await Task.CompletedTask;
|
||||
}
|
||||
|
||||
public async Task<bool> PublishAsync(string topic, string payload, CancellationToken cancellationToken = default)
|
||||
{
|
||||
try
|
||||
{
|
||||
var message = new MqttApplicationMessageBuilder()
|
||||
.WithTopic(topic)
|
||||
.WithPayload(payload)
|
||||
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
|
||||
.Build();
|
||||
|
||||
await _mqttClient.PublishAsync(message, cancellationToken);
|
||||
return true;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "[MQTT] 发布消息到主题 {Topic} 失败", topic);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
public override async Task StopAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
await _mqttClient.DisconnectAsync();
|
||||
await base.StopAsync(stoppingToken);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
  需要实现后台服务接口一直运行。
|
||||
@@ -634,310 +634,3 @@ Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo= # 服务器加密后的密
|
||||
| 固定头部 | 必选，2 字节起，包含消息类型（如发布、订阅）、QoS 等级、是否保留消息等标识。 |
|
||||
| 可变头部 | 可选，仅部分消息类型（如发布、订阅）需要，包含主题名、消息 ID 等信息。 |
|
||||
| 负载 | 可选，消息的实际内容（如 JSON 字符串、二进制数据），例如 {"temperature": 25}。 |
|
||||
|
||||
# 五、.Net Core实现
|
||||
## 5.1 SuperSocket
|
||||
  [SuperSocket](https://www.supersocket.net/)是一个轻量级, 跨平台而且可扩展的 .Net/Mono Socket 服务器程序框架。可以轻松构建TCP、UDP、WebSocket服务器。
|
||||
|
||||
## 5.2 安装依赖
|
||||
  NuGut安装SuperSocket、SuperSocket.WebSocket和SuperSocket.WebSocket.Server 2.0及以上版本。
|
||||
|
||||
## 5.3 配置文件
|
||||
  appsettings.json
|
||||
```json
|
||||
{
|
||||
"serverOptions": {
|
||||
"TcpServer": {
|
||||
"name": "TcpServer",
|
||||
"listeners": [
|
||||
{
|
||||
"ip": "Any",
|
||||
"port": 4040
|
||||
}
|
||||
]
|
||||
},
|
||||
"WebSocketServer": {
|
||||
"name": "WebSocket",
|
||||
"listeners": [
|
||||
{
|
||||
"ip": "Any",
|
||||
"port": 5050
|
||||
}
|
||||
]
|
||||
}
|
||||
},
|
||||
"Mqtt": {
|
||||
"Host": "127.0.0.1",
|
||||
"Port": 1883,
|
||||
"ClientId": "MqttClient",
|
||||
"Topics": [
|
||||
"test/topic1",
|
||||
"test/topic2"
|
||||
]
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## 5.3 主程序
|
||||
  program.cs
|
||||
```cs
|
||||
var host = Host.CreateDefaultBuilder(args)
|
||||
.ConfigureServices((context, services) =>
|
||||
{
|
||||
services.AddSingleton<IPackageHandler<ProtocolFrame>, TcpPackageHandler>();
|
||||
services.AddSingleton<WebSocketMessageHandler>();
|
||||
|
||||
// 1. 注入 MQTT 配置
|
||||
services.Configure<MqttSettings>(context.Configuration.GetSection("Mqtt"));
|
||||
// 2. 注册 MQTT 客户端
|
||||
services.AddSingleton<IMqttClient>(serviceProvider => new MqttClientFactory().CreateMqttClient());
|
||||
// 3. 注册 MQTT 后台服务
|
||||
services.AddHostedService<MqttService>();
|
||||
})
|
||||
.AsMultipleServerHostBuilder()
|
||||
.AddServer<TcpService, ProtocolFrame, BinaryPipelineFilter>(builder =>
|
||||
{
|
||||
builder.ConfigureServerOptions((ctx, config) => config.GetSection("TcpServer"));
|
||||
})
|
||||
.AddWebSocketServer(builder =>
|
||||
{
|
||||
builder
|
||||
.UseWebSocketMessageHandler(async (session, package) =>
|
||||
{
|
||||
using var scope = session.Server.ServiceProvider.CreateScope();
|
||||
var handler = scope.ServiceProvider.GetRequiredService<WebSocketMessageHandler>();
|
||||
|
||||
await handler.HandleAsync(session, package);
|
||||
})
|
||||
.ConfigureServerOptions((ctx, config) => config.GetSection("WebSocketServer"));
|
||||
})
|
||||
.ConfigureLogging(logging => logging.ClearProviders())
|
||||
.UseNLog()
|
||||
.Build();
|
||||
|
||||
await host.RunAsync();
|
||||
```
|
||||
|
||||
  通过.AsMultipleServerHostBuilder()可以构造多服务器实例。
|
||||
|
||||
## 5.4 TCP服务器
|
||||
### 5.4.1 协议数据
|
||||
```cs
|
||||
/// <summary>
|
||||
/// 协议数据结构 示例：65 6D 00 05 00 01 68 65 6C 6C 6F
|
||||
/// </summary>
|
||||
/// <param name="Magic">帧头标识</param>
|
||||
/// <param name="Length">数据长度</param>
|
||||
/// <param name="Type">数据类型</param>
|
||||
/// <param name="Payload">数据内容</param>
|
||||
///
|
||||
record ProtocolFrame(
|
||||
ushort Magic,
|
||||
ushort Length,
|
||||
ushort Type,
|
||||
byte[] Payload
|
||||
);
|
||||
```
|
||||
|
||||
  可以根据实际情况自定义消息格式。
|
||||
|
||||
### 5.4.2 协议解析
|
||||
```cs
|
||||
class BinaryPipelineFilter : FixedHeaderPipelineFilter<ProtocolFrame>
|
||||
{
|
||||
/// <summary>
|
||||
/// 固定头长度
|
||||
/// </summary>
|
||||
public BinaryPipelineFilter() : base(6)
|
||||
{
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 从包头中解析出 Body 长度
|
||||
/// </summary>
|
||||
/// <param name="buffer">字节流</param>
|
||||
/// <returns>Body 长度</returns>
|
||||
protected override int GetBodyLengthFromHeader(ref ReadOnlySequence<byte> buffer)
|
||||
{
|
||||
var reader = new SequenceReader<byte>(buffer);
|
||||
|
||||
// 读取前2字节 → magic
|
||||
reader.TryReadBigEndian(out ushort magic);
|
||||
|
||||
// 再读2字节 → Length
|
||||
reader.TryReadBigEndian(out ushort length);
|
||||
|
||||
// 再读2字节 → Type
|
||||
reader.TryReadBigEndian(out ushort type);
|
||||
|
||||
// 校验帧头
|
||||
if (magic != 0x656D)
|
||||
{
|
||||
throw new Exception("非法帧头");
|
||||
}
|
||||
|
||||
// 限制长度
|
||||
if (length == 0 || length > 8192)
|
||||
{
|
||||
throw new Exception("非法长度");
|
||||
}
|
||||
|
||||
// 返回 Payload 长度
|
||||
return length;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 把完整字节包 → 转换成 ProtocolFrame
|
||||
/// </summary>
|
||||
/// <param name="buffer">字节流</param>
|
||||
/// <returns>ProtocolFrame</returns>
|
||||
protected override ProtocolFrame DecodePackage(ref ReadOnlySequence<byte> buffer)
|
||||
{
|
||||
var reader = new SequenceReader<byte>(buffer);
|
||||
|
||||
// 读取前2字节 → magic
|
||||
reader.TryReadBigEndian(out ushort magic);
|
||||
|
||||
// 再读2字节 → Length
|
||||
reader.TryReadBigEndian(out ushort length);
|
||||
|
||||
// 再读2字节 → Type
|
||||
reader.TryReadBigEndian(out ushort type);
|
||||
|
||||
var payload = buffer.Slice(6, length).ToArray();
|
||||
|
||||
// 构造ProtocolFrame
|
||||
return new ProtocolFrame(magic, length, type, payload);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
  根据协议数据ProtocolFrame来解包。
|
||||
|
||||
### 5.4.3 消息处理
|
||||
```cs
|
||||
class TcpPackageHandler(ILogger<TcpPackageHandler> logger) : IPackageHandler<ProtocolFrame>
|
||||
{
|
||||
private readonly ILogger<TcpPackageHandler> _logger = logger;
|
||||
|
||||
public async ValueTask Handle(IAppSession session, ProtocolFrame package, CancellationToken cancellationToken)
|
||||
{
|
||||
_logger.LogInformation($"Magic={package.Magic:X4}, Type={package.Type}, Len={package.Length}");
|
||||
await session.SendAsync(package.Payload, cancellationToken);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 5.4.4 TCPService
|
||||
```cs
|
||||
class TcpService(IServiceProvider serviceProvider, IOptions<ServerOptions> serverOptions) : SuperSocketService<ProtocolFrame>(serviceProvider, serverOptions)
|
||||
{
|
||||
}
|
||||
```
|
||||
|
||||
## 5.5 Websocket服务器
|
||||
### 5.5.1 消息处理
|
||||
```cs
|
||||
class WebSocketMessageHandler(ILogger<WebSocketMessageHandler> logger)
|
||||
{
|
||||
private readonly ILogger<WebSocketMessageHandler> _logger = logger;
|
||||
|
||||
public async ValueTask HandleAsync(WebSocketSession session, WebSocketPackage package)
|
||||
{
|
||||
_logger.LogInformation($"[WebSocket] {package.Message}");
|
||||
await session.SendAsync("ok");
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## 5.6 MQTT服务器
|
||||
### 5.6 MqttService
|
||||
```cs
|
||||
class MqttService(IMqttClient mqttClient, IOptions<MqttSettings> settings, ILogger<TcpPackageHandler> logger) : BackgroundService
|
||||
{
|
||||
private readonly IMqttClient _mqttClient = mqttClient;
|
||||
private readonly MqttSettings _settings = settings.Value;
|
||||
private readonly ILogger<TcpPackageHandler> _logger = logger;
|
||||
private MqttClientOptions? _mqttOptions;
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
// 构建连接配置
|
||||
_mqttOptions = new MqttClientOptionsBuilder()
|
||||
.WithClientId($"{_settings.ClientId}_{Guid.NewGuid():N}")
|
||||
.WithTcpServer(_settings.Host, _settings.Port)
|
||||
.WithCleanSession()
|
||||
.Build();
|
||||
|
||||
_mqttClient.ConnectedAsync += OnConnectedAsync;
|
||||
_mqttClient.DisconnectedAsync += OnDisconnectedAsync;
|
||||
_mqttClient.ApplicationMessageReceivedAsync += HandleMessage;
|
||||
|
||||
_logger.LogInformation("[MQTT] 正在连接到服务器 {Host}:{Port}...", _settings.Host, _settings.Port);
|
||||
|
||||
await _mqttClient.ConnectAsync(_mqttOptions, stoppingToken);
|
||||
|
||||
// 等待程序停止
|
||||
await Task.Delay(Timeout.Infinite, stoppingToken);
|
||||
}
|
||||
|
||||
public async Task OnConnectedAsync(MqttClientConnectedEventArgs arg)
|
||||
{
|
||||
_logger.LogInformation("[MQTT] 已成功连接到服务器");
|
||||
|
||||
// 连接成功后订阅主题
|
||||
foreach (var topic in _settings.Topics)
|
||||
{
|
||||
await _mqttClient.SubscribeAsync(topic, MqttQualityOfServiceLevel.AtLeastOnce);
|
||||
_logger.LogInformation($"[MQTT] 已订阅主题：{topic}");
|
||||
}
|
||||
|
||||
await Task.CompletedTask;
|
||||
}
|
||||
|
||||
public async Task OnDisconnectedAsync(MqttClientDisconnectedEventArgs arg)
|
||||
{
|
||||
_logger.LogError("[MQTT] 连接失败，原因：{Reason}", arg.Reason);
|
||||
|
||||
await Task.CompletedTask;
|
||||
}
|
||||
|
||||
public async Task HandleMessage(MqttApplicationMessageReceivedEventArgs arg)
|
||||
{
|
||||
var topic = arg.ApplicationMessage.Topic;
|
||||
var payload = Encoding.UTF8.GetString(arg.ApplicationMessage.Payload);
|
||||
|
||||
_logger.LogInformation($"\n[MQTT] 收到消息\n主题：{topic}\n内容：{payload}\n");
|
||||
|
||||
await Task.CompletedTask;
|
||||
}
|
||||
|
||||
public async Task<bool> PublishAsync(string topic, string payload, CancellationToken cancellationToken = default)
|
||||
{
|
||||
try
|
||||
{
|
||||
var message = new MqttApplicationMessageBuilder()
|
||||
.WithTopic(topic)
|
||||
.WithPayload(payload)
|
||||
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
|
||||
.Build();
|
||||
|
||||
await _mqttClient.PublishAsync(message, cancellationToken);
|
||||
return true;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "[MQTT] 发布消息到主题 {Topic} 失败", topic);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
public override async Task StopAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
await _mqttClient.DisconnectAsync();
|
||||
await base.StopAsync(stoppingToken);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
  需要实现后台服务接口一直运行。
|
||||
202
docs/Web/Vue/ElTable.md
Normal file
202
docs/Web/Vue/ElTable.md
Normal file
@@ -0,0 +1,202 @@
|
||||
---
|
||||
title: Element Plus 表格组件
|
||||
date: 2026-05-27
|
||||
---
|
||||
|
||||
# 一、加载中
|
||||
```vue
|
||||
<el-table
|
||||
v-loading="dispatchStore.isLoading"
|
||||
element-loading-text="加载中..."
|
||||
:data="dispatchStore.contractList"
|
||||
border
|
||||
stripeempty-text="暂无记录"
|
||||
class="common-table">
|
||||
|
||||
</el-table>
|
||||
```
|
||||
|
||||
::: tip
|
||||
要配合pinia存储使用，获取数据前将loading置为true，完成后置为false。
|
||||
:::
|
||||
|
||||
# 二、展开行
|
||||
```vue
|
||||
<template>
|
||||
<el-table :data="dispatchStore.contractList"
|
||||
:row-key="getRowKey"
|
||||
:expand-row-keys="expandedRows"
|
||||
@expand-change="handleExpandChange">
|
||||
<el-table-column type="expand">
|
||||
<template #default="props">
|
||||
<el-table :data="props.row.dispatchList"
|
||||
border stripeempty-text="暂无记录"
|
||||
class="dispatch-content"
|
||||
style="width: 100%">
|
||||
<el-table-column fixed type="index" align="center" width="60" />
|
||||
<el-table-column prop="customer" label="终端客户" align="center" width="180"/>
|
||||
</el-table>
|
||||
</template>
|
||||
</el-table-column>
|
||||
</el-table>
|
||||
</template>
|
||||
|
||||
<script lang="ts" setup>
|
||||
const expandedRows = ref<number[]>([])
|
||||
|
||||
// 获取行的唯一标识
|
||||
const getRowKey = (row: IDispatch) => {
|
||||
return row.id
|
||||
}
|
||||
|
||||
// 处理展开行变化
|
||||
const handleExpandChange = (row: IDispatch, expandedRowsList: IDispatch[]) => {
|
||||
if (expandedRowsList.length > 0) {
|
||||
// 只保留最新展开的行
|
||||
expandedRows.value = [getRowKey(row)]
|
||||
} else {
|
||||
expandedRows.value = []
|
||||
}
|
||||
}
|
||||
</script>
|
||||
```
|
||||
|
||||
  通过类型expand和插槽props实现，可以展示任意格式数据。
|
||||
  增加表格的`row-key`、`expand-row-key`属性可以实现只展开某一行功能。
|
||||
|
||||
# 三、排序和筛选
|
||||
```vue
|
||||
<template>
|
||||
<el-table :data="dispatchStore.contractList">
|
||||
<el-table-column prop="contractNum" label="合同编号" align="center" min-width="10%"
|
||||
:filters=contractNumFilters :filter-method="filterColumnMethod" sortable>
|
||||
</el-table-column>
|
||||
</el-table>
|
||||
</template>
|
||||
|
||||
<script lang="ts" setup>
|
||||
import { useDispatchStore } from '@/stores/dispatch.ts'
|
||||
import { computed } from 'vue'
|
||||
import type { IContract } from '@/types/dispatch.ts'
|
||||
|
||||
const dispatchStore = useDispatchStore()
|
||||
|
||||
type ContractField = keyof IContract
|
||||
|
||||
const filterColumnMethod = (value, row, column) => {
|
||||
return row[column.property] === value
|
||||
}
|
||||
|
||||
const getFiltersByKey = (key: ContractField) => {
|
||||
const result = [...new Set(dispatchStore.contractList.map((item) => item[key]).filter((value) => value != null && value !==''))]
|
||||
|
||||
return result.map((item) => ({
|
||||
text: item,
|
||||
value: item
|
||||
}))
|
||||
}
|
||||
|
||||
const contractNumFilters = computed(() => getFiltersByKey('contractNum'))
|
||||
</script>
|
||||
```
|
||||
|
||||
  在`el-table-column`中添加`sortable`即可实现排序功能。
|
||||
  在`el-table-column`中添加`filters`和`filter-method`实现筛选功能。
|
||||
  `filters`为原始数据中该字段下的所有值，并转为指定格式，也可以自定义筛选字段。
|
||||
|
||||
# 四、表头搜索
|
||||
```vue
|
||||
<template>
|
||||
<el-table :data="dispatchStore.contractList" >
|
||||
<el-table-column prop="contractNum" label="合同编号" align="center" min-width="10%">
|
||||
<template #header>
|
||||
<div class="custom-header">
|
||||
<span :class="{ 'active': contractNumSearch }">合同编号</span>
|
||||
<el-popover ref="contractNumPopoverRef" placement="bottom" width="230" trigger="click">
|
||||
<template #default>
|
||||
<el-select v-model="contractNumSearch" filterable clearable
|
||||
placeholder="请选择或输入合同编号" @change="applySearch('contractNum')">
|
||||
<el-option v-for="(item, index) in contractNumValue"
|
||||
:key="index" :label="item" :value="item"/>
|
||||
</el-select>
|
||||
<div style="margin-top: 8px;display: flex;justify-content: flex-end;">
|
||||
<el-button type="primary" size="small" @click="applySearch('contractNum')">搜索</el-button>
|
||||
<el-button size="small" @click="resetSearch('contractNum')">重置</el-button>
|
||||
</div>
|
||||
</template>
|
||||
<template #reference>
|
||||
<el-icon class="search-icon"><Search/></el-icon>
|
||||
</template>
|
||||
</el-popover>
|
||||
</div>
|
||||
</template>
|
||||
</el-table-column>
|
||||
</el-table>
|
||||
</template>
|
||||
|
||||
<script lang="ts" setup>
|
||||
import { useDispatchStore } from '@/stores/dispatch.ts'
|
||||
import { computed, ref } from 'vue'
|
||||
import { Search } from '@element-plus/icons-vue'
|
||||
import type { IContract } from '@/types/dispatch.ts'
|
||||
import type { PopoverInstance } from 'element-plus'
|
||||
import type { Ref } from 'vue'
|
||||
|
||||
const dispatchStore = useDispatchStore()
|
||||
|
||||
type ContractField = keyof IContract
|
||||
|
||||
const getFiltersByKey = (key: ContractField) => {
|
||||
const result = [...new Set(dispatchStore.contractList.map((item) => item[key]).filter((value) => value != null && value !== ''))]
|
||||
|
||||
return result.map((item) => ({
|
||||
text: item,
|
||||
value: item
|
||||
}))
|
||||
}
|
||||
|
||||
const contractNumFilters = computed(() => getFiltersByKey('contractNum'))
|
||||
|
||||
const contractNumValue = computed(() => contractNumFilters.value.map((item) => item.value))
|
||||
|
||||
const contractNumPopoverRef = ref<PopoverInstance>()
|
||||
const contractNumSearch = ref('')
|
||||
|
||||
const popoverMap = new Map<string, Ref<PopoverInstance | undefined>>([
|
||||
['contractNum', contractNumPopoverRef]
|
||||
])
|
||||
const searchMap = new Map<string, Ref<string>>([
|
||||
['contractNum', contractNumSearch]
|
||||
])
|
||||
|
||||
const applySearch = async (field: ContractField) => {
|
||||
const searchRef = searchMap.get(field)
|
||||
|
||||
if (!searchRef) return
|
||||
|
||||
if (!searchRef.value) {
|
||||
await dispatchStore.queryContractList()
|
||||
return
|
||||
}
|
||||
if (!searchRef.value.trim()) return
|
||||
|
||||
dispatchStore.contractList = dispatchStore.contractList.filter((item) => {
|
||||
const value = item[field]
|
||||
return typeof value === 'string' && value.includes(searchRef.value)
|
||||
})
|
||||
popoverMap.get(field)?.value?.hide()
|
||||
}
|
||||
|
||||
const resetSearch = async (field: string) => {
|
||||
const searchRef = searchMap.get(field)
|
||||
|
||||
if (searchRef) {
|
||||
searchRef.value = ''
|
||||
}
|
||||
popoverMap.get(field)?.value?.hide()
|
||||
await dispatchStore.queryContractList()
|
||||
}
|
||||
</script>
|
||||
```
|
||||
|
||||
  表头搜索主要利用`header`插槽和`el-popover`组件实现。
|
||||
@@ -1,5 +1,5 @@
|
||||
---
|
||||
title: 前端包管理器
|
||||
title: Vite核心原理
|
||||
date: 2025-12-13
|
||||
---
|
||||
|
||||
Reference in New Issue
Block a user