1279 lines
44 KiB
Markdown
1279 lines
44 KiB
Markdown
# 软件开发文档 - DMS.Infrastructure - 数据访问与服务
|
||
|
||
本文档详细阐述了 `DMS.Infrastructure` 层的设计，它是所有外部技术和服务的具体实现地。它实现了 `DMS.Core` 定义的接口，为 `DMS.Application` 层提供数据和功能支持，并负责数据持久化、事务管理以及与外部世界的通信。
|
||
|
||
## 1. 目录结构
|
||
|
||
```
|
||
DMS.Infrastructure/
|
||
├── Data/
|
||
│ ├── SqlSugarDbContext.cs
|
||
│ └── RepositoryManager.cs
|
||
├── Entities/
|
||
│ ├── DbDevice.cs
|
||
│ ├── DbMqttServer.cs
|
||
│ ├── DbVariable.cs
|
||
│ ├── DbVariableTable.cs
|
||
│ ├── DbVariableHistory.cs
|
||
│ ├── DbVariableMqttAlias.cs
|
||
│ ├── DbMenu.cs
|
||
│ ├── DbNlog.cs
|
||
│ └── DbUser.cs
|
||
├── Repositories/
|
||
│ ├── BaseRepository.cs
|
||
│ ├── DeviceRepository.cs
|
||
│ ├── MqttServerRepository.cs
|
||
│ ├── VariableRepository.cs
|
||
│ ├── VariableTableRepository.cs
|
||
│ ├── VariableHistoryRepository.cs
|
||
│ ├── VariableMqttAliasRepository.cs
|
||
│ ├── MenuRepository.cs
|
||
│ └── UserRepository.cs
|
||
├── Services/
|
||
│ ├── Communication/
|
||
│ │ ├── S7DeviceAgent.cs
|
||
│ │ └── MqttPublishService.cs
|
||
│ ├── Processing/
|
||
│ │ ├── ChangeDetectionProcessor.cs
|
||
│ │ ├── HistoryStorageProcessor.cs
|
||
│ │ └── MqttPublishProcessor.cs
|
||
│ ├── S7BackgroundService.cs
|
||
│ ├── DataProcessingService.cs
|
||
│ ├── DatabaseInitializerService.cs
|
||
│ └── MenuService.cs
|
||
├── Logging/
|
||
│ ├── ThrottlingDatabaseTarget.cs
|
||
│ └── NLogService.cs
|
||
├── Profiles/
|
||
│ └── MappingProfile.cs
|
||
└── DMS.Infrastructure.csproj
|
||
```
|
||
|
||
## 2. 数据库设计与实体 (`Entities/`)
|
||
|
||
本文档详细描述了DMS系统的数据库结构，包括表、字段和关系。数据库实体类与数据库表一一对应，使用了 `SqlSugar` 的特性（Attribute）来定义主键、外键等。
|
||
|
||
### 2.1. 数据库关系图 (ERD)
|
||
|
||
```mermaid
|
||
erDiagram
|
||
DEVICE {
|
||
int Id PK
|
||
varchar Name
|
||
int Protocol
|
||
varchar IpAddress
|
||
int Port
|
||
bool IsActive
|
||
}
|
||
|
||
VARIABLE_TABLE {
|
||
int Id PK
|
||
varchar Name
|
||
varchar Description
|
||
bool IsActive
|
||
int DeviceId FK
|
||
}
|
||
|
||
VARIABLE {
|
||
int Id PK
|
||
varchar Name
|
||
varchar Address
|
||
int DataType
|
||
bool IsActive
|
||
int VariableTableId FK
|
||
}
|
||
|
||
MQTT_SERVER {
|
||
int Id PK
|
||
varchar ServerName
|
||
varchar BrokerAddress
|
||
int Port
|
||
varchar Username
|
||
varchar Password
|
||
bool IsActive
|
||
}
|
||
|
||
VARIABLE_HISTORY {
|
||
bigint Id PK
|
||
int VariableId FK
|
||
varchar Value
|
||
datetime Timestamp
|
||
}
|
||
|
||
VARIABLE_MQTT_ALIAS {
|
||
int Id PK
|
||
int VariableId FK
|
||
int MqttServerId FK
|
||
varchar Alias
|
||
}
|
||
|
||
MENU {
|
||
int Id PK
|
||
int ParentId FK
|
||
varchar Header
|
||
varchar Icon
|
||
varchar TargetViewKey
|
||
varchar NavigationParameter
|
||
int DisplayOrder
|
||
}
|
||
|
||
NLOG {
|
||
bigint Id PK
|
||
datetime Logged
|
||
varchar Level
|
||
varchar Message
|
||
varchar Exception
|
||
varchar CallSite
|
||
varchar MethodName
|
||
int AggregatedCount
|
||
}
|
||
|
||
USER {
|
||
int Id PK
|
||
varchar Username
|
||
varchar PasswordHash
|
||
varchar Role
|
||
bool IsActive
|
||
}
|
||
|
||
DEVICE ||--o{ VARIABLE_TABLE : "包含"
|
||
VARIABLE_TABLE ||--o{ VARIABLE : "包含"
|
||
VARIABLE ||--o{ VARIABLE_HISTORY : "记录"
|
||
VARIABLE }o--o{ VARIABLE_MQTT_ALIAS : "关联"
|
||
MQTT_SERVER }o--o{ VARIABLE_MQTT_ALIAS : "关联"
|
||
MENU ||--o{ MENU : "父子"
|
||
|
||
```
|
||
|
||
### 2.2. 数据库实体类示例
|
||
|
||
#### `DbDevice.cs`
|
||
|
||
```csharp
|
||
// 文件: DMS.Infrastructure/Entities/DbDevice.cs
|
||
using SqlSugar;
|
||
|
||
namespace DMS.Infrastructure.Entities;
|
||
|
||
[SugarTable("Devices")]
|
||
public class DbDevice
|
||
{
|
||
[SugarColumn(IsPrimaryKey = true, IsIdentity = true)]
|
||
public int Id { get; set; }
|
||
public string Name { get; set; }
|
||
public int Protocol { get; set; } // 对应 ProtocolType 枚举
|
||
public string IpAddress { get; set; }
|
||
public int Port { get; set; }
|
||
public bool IsActive { get; set; }
|
||
public int Rack { get; set; }
|
||
public int Slot { get; set; }
|
||
}
|
||
```
|
||
|
||
#### `DbVariableTable.cs`
|
||
|
||
```csharp
|
||
// 文件: DMS.Infrastructure/Entities/DbVariableTable.cs
|
||
using SqlSugar;
|
||
|
||
namespace DMS.Infrastructure.Entities;
|
||
|
||
[SugarTable("VariableTables")]
|
||
public class DbVariableTable
|
||
{
|
||
[SugarColumn(IsPrimaryKey = true, IsIdentity = true)]
|
||
public int Id { get; set; }
|
||
public string Name { get; set; }
|
||
public string Description { get; set; }
|
||
public bool IsActive { get; set; }
|
||
public int DeviceId { get; set; }
|
||
public int Protocol { get; set; } // 对应 ProtocolType 枚举
|
||
}
|
||
```
|
||
|
||
#### `DbVariable.cs`
|
||
|
||
```csharp
|
||
// 文件: DMS.Infrastructure/Entities/DbVariable.cs
|
||
using SqlSugar;
|
||
|
||
namespace DMS.Infrastructure.Entities;
|
||
|
||
[SugarTable("Variables")]
|
||
public class DbVariable
|
||
{
|
||
[SugarColumn(IsPrimaryKey = true, IsIdentity = true)]
|
||
public int Id { get; set; }
|
||
public string Name { get; set; }
|
||
public string Address { get; set; }
|
||
public int DataType { get; set; } // 对应 SignalType 枚举
|
||
public int PollLevel { get; set; } // 对应 PollLevelType 枚举
|
||
public bool IsActive { get; set; }
|
||
public int VariableTableId { get; set; }
|
||
public int Protocol { get; set; } // 对应 ProtocolType 枚举
|
||
public int CSharpDataType { get; set; } // 对应 CSharpDataType 枚举
|
||
public string ConversionFormula { get; set; }
|
||
public DateTime CreatedAt { get; set; }
|
||
public DateTime UpdatedAt { get; set; }
|
||
public string UpdatedBy { get; set; }
|
||
public bool IsModified { get; set; }
|
||
}
|
||
```
|
||
|
||
#### `DbMqttServer.cs`
|
||
|
||
```csharp
|
||
// 文件: DMS.Infrastructure/Entities/DbMqttServer.cs
|
||
using SqlSugar;
|
||
|
||
namespace DMS.Infrastructure.Entities;
|
||
|
||
[SugarTable("MqttServers")]
|
||
public class DbMqttServer
|
||
{
|
||
[SugarColumn(IsPrimaryKey = true, IsIdentity = true)]
|
||
public int Id { get; set; }
|
||
public string ServerName { get; set; }
|
||
public string BrokerAddress { get; set; }
|
||
public int Port { get; set; }
|
||
public string Username { get; set; }
|
||
public string Password { get; set; }
|
||
public bool IsActive { get; set; }
|
||
public string SubscribeTopic { get; set; }
|
||
public string PublishTopic { get; set; }
|
||
public string ClientId { get; set; }
|
||
public DateTime CreatedAt { get; set; }
|
||
public DateTime? ConnectedAt { get; set; }
|
||
public long ConnectionDuration { get; set; }
|
||
public string MessageFormat { get; set; }
|
||
}
|
||
```
|
||
|
||
#### `DbVariableHistory.cs`
|
||
|
||
```csharp
|
||
// 文件: DMS.Infrastructure/Entities/DbVariableHistory.cs
|
||
using SqlSugar;
|
||
|
||
namespace DMS.Infrastructure.Entities;
|
||
|
||
[SugarTable("VariableHistories")]
|
||
public class DbVariableHistory
|
||
{
|
||
[SugarColumn(IsPrimaryKey = true, IsIdentity = true)]
|
||
public long Id { get; set; }
|
||
public int VariableId { get; set; }
|
||
public string Value { get; set; }
|
||
public DateTime Timestamp { get; set; }
|
||
}
|
||
```
|
||
|
||
#### `DbVariableMqttAlias.cs`
|
||
|
||
```csharp
|
||
// 文件: DMS.Infrastructure/Entities/DbVariableMqttAlias.cs
|
||
using SqlSugar;
|
||
|
||
namespace DMS.Infrastructure.Entities;
|
||
|
||
/// <summary>
|
||
/// 数据库实体：对应数据库中的 VariableMqttAliases 表。
|
||
/// </summary>
|
||
[SugarTable("VariableMqttAliases")]
|
||
public class DbVariableMqttAlias
|
||
{
|
||
[SugarColumn(IsPrimaryKey = true, IsIdentity = true)]
|
||
public int Id { get; set; }
|
||
|
||
/// <summary>
|
||
/// 外键，指向 Variables 表的 Id。
|
||
/// </summary>
|
||
public int VariableId { get; set; }
|
||
|
||
/// <summary>
|
||
/// 外键，指向 MqttServers 表的 Id。
|
||
/// </summary>
|
||
public int MqttServerId { get; set; }
|
||
|
||
/// <summary>
|
||
/// 针对此特定[变量-服务器]连接的发布别名。
|
||
/// </summary>
|
||
public string Alias { get; set; }
|
||
}
|
||
```
|
||
|
||
#### `DbMenu.cs`
|
||
|
||
```csharp
|
||
// 文件: DMS.Infrastructure/Entities/DbMenu.cs
|
||
using SqlSugar;
|
||
|
||
namespace DMS.Infrastructure.Entities;
|
||
|
||
[SugarTable("Menus")]
|
||
public class DbMenu
|
||
{
|
||
[SugarColumn(IsPrimaryKey = true, IsIdentity = true)]
|
||
public int Id { get; set; }
|
||
|
||
[SugarColumn(IsNullable = true)]
|
||
public int? ParentId { get; set; }
|
||
|
||
public string Header { get; set; }
|
||
|
||
public string Icon { get; set; }
|
||
|
||
public string TargetViewKey { get; set; }
|
||
|
||
[SugarColumn(IsNullable = true)]
|
||
public string NavigationParameter { get; set; }
|
||
|
||
public int DisplayOrder { get; set; }
|
||
}
|
||
```
|
||
|
||
#### `DbNlog.cs`
|
||
|
||
```csharp
|
||
// 文件: DMS.Infrastructure/Entities/DbNlog.cs
|
||
using SqlSugar;
|
||
using System;
|
||
|
||
namespace DMS.Infrastructure.Entities;
|
||
|
||
/// <summary>
|
||
/// 数据库实体：对应数据库中的 Logs 表，用于存储应用程序日志。
|
||
/// </summary>
|
||
[SugarTable("Logs")]
|
||
public class DbNlog
|
||
{
|
||
[SugarColumn(IsPrimaryKey = true, IsIdentity = true)]
|
||
public long Id { get; set; }
|
||
|
||
/// <summary>
|
||
/// 日志记录的时间戳。
|
||
/// </summary>
|
||
public DateTime Logged { get; set; }
|
||
|
||
/// <summary>
|
||
/// 日志级别 (e.g., "Info", "Warn", "Error", "Debug")。
|
||
/// </summary>
|
||
public string Level { get; set; }
|
||
|
||
/// <summary>
|
||
/// 日志消息主体。
|
||
/// </summary>
|
||
[SugarColumn(Length = -1)] // 映射为NVARCHAR(MAX)或类似类型
|
||
public string Message { get; set; }
|
||
|
||
/// <summary>
|
||
/// 异常信息，包括堆栈跟踪。如果无异常则为null。
|
||
/// </summary>
|
||
[SugarColumn(IsNullable = true, Length = -1)]
|
||
public string Exception { get; set; }
|
||
|
||
/// <summary>
|
||
/// 记录日志的调用点信息 (文件路径:行号)。
|
||
/// </summary>
|
||
public string CallSite { get; set; }
|
||
|
||
/// <summary>
|
||
/// 记录日志的方法名。
|
||
/// </summary>
|
||
public string MethodName { get; set; }
|
||
|
||
/// <summary>
|
||
/// (用于聚合) 此条日志在指定时间窗口内被触发的总次数。默认为1。
|
||
/// </summary>
|
||
public int AggregatedCount { get; set; } = 1;
|
||
}
|
||
```
|
||
|
||
#### `DbUser.cs`
|
||
|
||
```csharp
|
||
// 文件: DMS.Infrastructure/Entities/DbUser.cs
|
||
using SqlSugar;
|
||
|
||
namespace DMS.Infrastructure.Entities;
|
||
|
||
[SugarTable("Users")]
|
||
public class DbUser
|
||
{
|
||
[SugarColumn(IsPrimaryKey = true, IsIdentity = true)]
|
||
public int Id { get; set; }
|
||
public string Username { get; set; }
|
||
public string PasswordHash { get; set; }
|
||
public string Role { get; set; }
|
||
public bool IsActive { get; set; }
|
||
}
|
||
```
|
||
|
||
## 3. 事务管理 (`RepositoryManager`)
|
||
|
||
`RepositoryManager` 是工作单元（Unit of Work, UoW）模式的具体实现。它作为所有仓储的统一入口，并管理数据库事务，确保所有通过它获取的仓储实例都共享同一个 `ISqlSugarClient` 实例，从而确保它们在同一个数据库会话和事务中操作。
|
||
|
||
### 3.1. 设计思路与考量
|
||
|
||
* **模式**：`RepositoryManager` 是工作单元（Unit of Work, UoW）模式的具体实现。它作为所有仓储的统一入口，并管理数据库事务。
|
||
* **共享上下文**：所有通过 `RepositoryManager` 获取的仓储实例都共享同一个 `ISqlSugarClient` 实例，从而确保它们在同一个数据库会话和事务中操作。
|
||
* **生命周期**：`RepositoryManager` 通常被注册为 `Scoped` 或 `Transient` 生命周期，确保每个业务操作都有一个独立的工作单元。
|
||
|
||
### 3.2. 设计优势
|
||
|
||
* **原子性**：确保跨多个仓储的操作（如创建设备、变量表和菜单）要么全部成功，要么全部失败，维护数据一致性。
|
||
* **简化事务管理**：应用层无需直接与数据库事务API交互，只需调用 `BeginTransaction()`, `CommitAsync()`, `RollbackAsync()`。
|
||
* **解耦**：应用层不直接依赖具体的仓储实现，而是依赖于 `IRepositoryManager` 这一抽象。
|
||
* **资源优化**：在单个业务操作中，所有仓储共享同一个数据库连接，减少了连接开销。
|
||
|
||
### 3.3. 设计劣势/权衡
|
||
|
||
* **复杂性增加**：相比于直接使用仓储，引入UoW模式增加了额外的抽象层和概念。
|
||
* **仓储依赖**：`IRepositoryManager` 接口需要列出所有它管理的仓储，当新增仓储时，需要修改此接口。
|
||
* **懒加载**：为了避免在 `RepositoryManager` 构造时就创建所有仓储实例的开销，通常会使用懒加载（`Lazy<T>`），这增加了少量复杂性。
|
||
|
||
### 3.4. 示例：`RepositoryManager.cs`
|
||
|
||
```csharp
|
||
// 文件: DMS.Infrastructure/Data/RepositoryManager.cs
|
||
using DMS.Core.Interfaces;
|
||
using SqlSugar;
|
||
using System;
|
||
|
||
namespace DMS.Infrastructure.Data;
|
||
|
||
/// <summary>
|
||
/// IRepositoryManager 的 SqlSugar 实现，管理所有仓储实例和数据库事务。
|
||
/// </summary>
|
||
public class RepositoryManager : IRepositoryManager
|
||
{
|
||
private readonly ISqlSugarClient _db;
|
||
private readonly Lazy<IDeviceRepository> _lazyDevices;
|
||
private readonly Lazy<IVariableTableRepository> _lazyVariableTables;
|
||
private readonly Lazy<IVariableRepository> _lazyVariables;
|
||
private readonly Lazy<IMqttServerRepository> _lazyMqttServers;
|
||
private readonly Lazy<IVariableMqttAliasRepository> _lazyVariableMqttAliases;
|
||
private readonly Lazy<IMenuRepository> _lazyMenus;
|
||
private readonly Lazy<IVariableHistoryRepository> _lazyVariableHistories;
|
||
private readonly Lazy<IUserRepository> _lazyUsers;
|
||
|
||
/// <summary>
|
||
/// 构造函数，通过依赖注入获取 SqlSugar 客户端实例。
|
||
/// </summary>
|
||
public RepositoryManager(ISqlSugarClient db, IMapper mapper) // 添加 IMapper 依赖
|
||
{
|
||
_db = db;
|
||
// 使用 Lazy<T> 实现仓储的懒加载，确保它们在第一次被访问时才创建。
|
||
// 所有仓储都共享同一个 _db 实例，以保证事务的一致性。
|
||
_lazyDevices = new Lazy<IDeviceRepository>(() => new DeviceRepository(_db, mapper));
|
||
_lazyVariableTables = new Lazy<IVariableTableRepository>(() => new VariableTableRepository(_db, mapper));
|
||
_lazyVariables = new Lazy<IVariableRepository>(() => new VariableRepository(_db, mapper));
|
||
_lazyMqttServers = new Lazy<IMqttServerRepository>(() => new MqttServerRepository(_db, mapper));
|
||
_lazyVariableMqttAliases = new Lazy<IVariableMqttAliasRepository>(() => new VariableMqttAliasRepository(_db, mapper));
|
||
_lazyMenus = new Lazy<IMenuRepository>(() => new MenuRepository(_db, mapper));
|
||
_lazyVariableHistories = new Lazy<IVariableHistoryRepository>(() => new VariableHistoryRepository(_db, mapper));
|
||
_lazyUsers = new Lazy<IUserRepository>(() => new UserRepository(_db, mapper));
|
||
}
|
||
|
||
public IDeviceRepository Devices => _lazyDevices.Value;
|
||
public IVariableTableRepository VariableTables => _lazyVariableTables.Value;
|
||
public IVariableRepository Variables => _lazyVariables.Value;
|
||
public IMqttServerRepository MqttServers => _lazyMqttServers.Value;
|
||
public IVariableMqttAliasRepository VariableMqttAliases => _lazyVariableMqttAliases.Value;
|
||
public IMenuRepository Menus => _lazyMenus.Value;
|
||
public IVariableHistoryRepository VariableHistories => _lazyVariableHistories.Value;
|
||
public IUserRepository Users => _lazyUsers.Value;
|
||
|
||
/// <summary>
|
||
/// 开始一个新的数据库事务。
|
||
/// </summary>
|
||
public void BeginTransaction()
|
||
{
|
||
_db.BeginTran();
|
||
}
|
||
|
||
/// <summary>
|
||
/// 异步提交当前事务中的所有变更。
|
||
/// </summary>
|
||
public async Task CommitAsync()
|
||
{
|
||
await _db.CommitTranAsync();
|
||
}
|
||
|
||
/// <summary>
|
||
/// 异步回滚当前事务中的所有变更。
|
||
/// </summary>
|
||
public async Task RollbackAsync()
|
||
{
|
||
await _db.RollbackTranAsync();
|
||
}
|
||
|
||
/// <summary>
|
||
/// 释放 SqlSugar 客户端资源。通常由 DI 容器管理。
|
||
/// </summary>
|
||
public void Dispose()
|
||
{
|
||
_db.Dispose();
|
||
}
|
||
}
|
||
```
|
||
|
||
## 4. 仓储实现 (`Repositories/`)
|
||
|
||
仓储（Repository）是数据访问层的一部分，负责实现 `DMS.Core` 中定义的仓储接口。它们使用ORM框架（如SqlSugar）与数据库进行实际交互，将数据库实体转换为领域模型，反之亦然。
|
||
|
||
### 4.1. 设计思路与考量
|
||
|
||
* **职责**：仓储（Repository）是数据访问层的一部分，负责实现 `DMS.Core` 中定义的仓储接口。它们使用ORM框架（如SqlSugar）与数据库进行实际交互，将数据库实体转换为领域模型，反之亦然。
|
||
* **通用性**：可以有一个 `BaseRepository` 来实现通用的CRUD操作，减少重复代码。
|
||
* **特定查询**：为每个仓储接口实现其特有的查询方法（如 `GetActiveDevicesWithDetailsAsync`）。
|
||
|
||
### 4.2. 设计优势
|
||
|
||
* **实现细节封装**：将ORM框架和数据库查询的细节封装在仓储内部，上层无需关心。
|
||
* **可替换性**：如果需要更换ORM框架或数据库类型，只需修改仓储实现，而无需修改应用层或核心层。
|
||
* **性能优化点**：可以在仓储层进行查询优化（如使用 `Include` 或 `Join` 预加载关联数据）。
|
||
|
||
### 4.3. 设计劣势/权衡
|
||
|
||
* **代码量**：即使有 `BaseRepository`，每个仓储仍然需要一些样板代码。
|
||
* **复杂查询**：对于非常复杂的、跨多个聚合根的查询，有时仓储模式会显得笨拙，可能需要引入查询对象（Query Object）模式。
|
||
|
||
### 4.4. 示例：`BaseRepository.cs`
|
||
|
||
```csharp
|
||
// 文件: DMS.Infrastructure/Repositories/BaseRepository.cs
|
||
using AutoMapper;
|
||
using DMS.Core.Interfaces;
|
||
using SqlSugar;
|
||
using System.Collections.Generic;
|
||
using System.Threading.Tasks;
|
||
|
||
namespace DMS.Infrastructure.Repositories;
|
||
|
||
/// <summary>
|
||
/// 仓储基类，实现了 IBaseRepository 接口的通用 CRUD 操作。
|
||
/// </summary>
|
||
/// <typeparam name="TDomain">领域模型类型。</typeparam>
|
||
/// <typeparam name="TDbEntity">数据库实体类型。</typeparam>
|
||
public abstract class BaseRepository<TDomain, TDbEntity> : IBaseRepository<TDomain>
|
||
where TDomain : class
|
||
where TDbEntity : class, new()
|
||
{
|
||
protected readonly ISqlSugarClient _db;
|
||
protected readonly IMapper _mapper;
|
||
|
||
public BaseRepository(ISqlSugarClient db, IMapper mapper)
|
||
{
|
||
_db = db;
|
||
_mapper = mapper;
|
||
}
|
||
|
||
public virtual async Task AddAsync(TDomain entity)
|
||
{
|
||
var dbEntity = _mapper.Map<TDbEntity>(entity);
|
||
await _db.Insertable(dbEntity).ExecuteCommandAsync();
|
||
// 映射回ID，如果实体是自增ID
|
||
_mapper.Map(dbEntity, entity);
|
||
}
|
||
|
||
public virtual async Task DeleteAsync(int id)
|
||
{
|
||
await _db.Deleteable<TDbEntity>(id).ExecuteCommandAsync();
|
||
}
|
||
|
||
public virtual async Task<List<TDomain>> GetAllAsync()
|
||
{
|
||
var dbEntities = await _db.Queryable<TDbEntity>().ToListAsync();
|
||
return _mapper.Map<List<TDomain>>(dbEntities);
|
||
}
|
||
|
||
public virtual async Task<TDomain> GetByIdAsync(int id)
|
||
{
|
||
var dbEntity = await _db.Queryable<TDbEntity>().InSingleAsync(id);
|
||
return _mapper.Map<TDomain>(dbEntity);
|
||
}
|
||
|
||
public virtual async Task UpdateAsync(TDomain entity)
|
||
{
|
||
var dbEntity = _mapper.Map<TDbEntity>(entity);
|
||
await _db.Updateable(dbEntity).ExecuteCommandAsync();
|
||
}
|
||
}
|
||
```
|
||
|
||
### 4.5. 示例：`DeviceRepository.cs`
|
||
|
||
```csharp
|
||
// 文件: DMS.Infrastructure/Repositories/DeviceRepository.cs
|
||
using AutoMapper;
|
||
using DMS.Core.Interfaces;
|
||
using DMS.Core.Models;
|
||
using DMS.Infrastructure.Entities;
|
||
using SqlSugar;
|
||
using System.Collections.Generic;
|
||
using System.Linq;
|
||
using System.Threading.Tasks;
|
||
|
||
namespace DMS.Infrastructure.Repositories;
|
||
|
||
/// <summary>
|
||
/// 设备仓储的具体实现。
|
||
/// </summary>
|
||
public class DeviceRepository : BaseRepository<Device, DbDevice>, IDeviceRepository
|
||
{
|
||
public DeviceRepository(ISqlSugarClient db, IMapper mapper) : base(db, mapper) { }
|
||
|
||
/// <summary>
|
||
/// 异步获取所有激活的S7设备，并级联加载其下的变量表和变量。
|
||
/// </summary>
|
||
public async Task<List<Device>> GetActiveDevicesWithDetailsAsync(ProtocolType protocol)
|
||
{
|
||
var dbDevices = await _db.Queryable<DbDevice>()
|
||
.Where(d => d.IsActive && d.Protocol == (int)protocol)
|
||
.Mapper(d => d.VariableTables, d => d.Id, d => d.VariableTables.Select(vt => vt.DeviceId).ToList())
|
||
.Mapper(d => d.VariableTables.Select(vt => vt.Variables), vt => vt.Id, vt => vt.Variables.Select(v => v.VariableTableId).ToList())
|
||
.ToListAsync();
|
||
return _mapper.Map<List<Device>>(dbDevices);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 异步根据设备ID获取设备及其所有详细信息（变量表、变量、MQTT别名等）。
|
||
/// </summary>
|
||
public async Task<Device> GetDeviceWithDetailsAsync(int deviceId)
|
||
{
|
||
var dbDevice = await _db.Queryable<DbDevice>()
|
||
.Where(d => d.Id == deviceId)
|
||
.Mapper(d => d.VariableTables, d => d.Id, d => d.VariableTables.Select(vt => vt.DeviceId).ToList())
|
||
.Mapper(d => d.VariableTables.Select(vt => vt.Variables), vt => vt.Id, vt => vt.Variables.Select(v => v.VariableTableId).ToList())
|
||
.FirstAsync();
|
||
|
||
if (dbDevice == null) return null;
|
||
|
||
// 手动加载 VariableMqttAlias，因为 SqlSugar 的 Mapper 可能无法直接处理多层嵌套的关联实体
|
||
foreach (var variableTable in dbDevice.VariableTables)
|
||
{
|
||
foreach (var variable in variableTable.Variables)
|
||
{
|
||
var dbAliases = await _db.Queryable<DbVariableMqttAlias>()
|
||
.Where(a => a.VariableId == variable.Id)
|
||
.Mapper(a => a.MqttServer, a => a.MqttServerId)
|
||
.ToListAsync();
|
||
variable.MqttAliases = _mapper.Map<List<VariableMqttAlias>>(dbAliases);
|
||
}
|
||
}
|
||
|
||
return _mapper.Map<Device>(dbDevice);
|
||
}
|
||
}
|
||
```
|
||
|
||
## 5. 外部服务 (`Services/`)
|
||
|
||
### 5.1. S7通信架构 (“编排者-代理”模式)
|
||
|
||
采用“编排者-代理”（Orchestrator-Agent）模式。`S7BackgroundService` 作为编排者，负责管理所有S7设备的生命周期；每个 `S7DeviceAgent` 作为代理，专门负责与**一个**S7 PLC进行所有交互。
|
||
|
||
#### 5.1.1. 设计思路与考量
|
||
|
||
* **模式**：采用“编排者-代理”（Orchestrator-Agent）模式。`S7BackgroundService` 作为编排者，负责管理所有S7设备的生命周期；每个 `S7DeviceAgent` 作为代理，专门负责与**一个**S7 PLC进行所有交互。
|
||
* **职责分离**：将设备管理（启动、停止、配置更新）与具体设备通信（连接、轮询、读写）的职责分离。
|
||
* **并发性**：每个 `S7DeviceAgent` 独立运行，可以并行处理多个设备的通信，提高系统吞吐量。
|
||
* **热重载**：通过消息机制，允许在运行时动态更新设备的变量配置，而无需重启整个服务。
|
||
|
||
#### 5.1.2. 设计优势
|
||
|
||
* **高可靠性**：单个设备的通信故障不会影响其他设备的正常运行。
|
||
* **可扩展性**：易于添加新的设备类型或通信协议，只需实现新的 `DeviceAgent`。
|
||
* **动态配置**：支持运行时修改设备和变量配置，提高了系统的灵活性和可用性。
|
||
* **资源隔离**：每个Agent管理自己的连接和资源，避免资源争用。
|
||
|
||
#### 5.1.3. 设计劣势/权衡
|
||
|
||
* **复杂性增加**：引入了Agent的概念和Agent与Orchestrator之间的通信机制，增加了初期设计和实现复杂性。
|
||
* **资源消耗**：每个Agent可能需要维护独立的连接和线程，当设备数量非常庞大时，可能会增加资源消耗。
|
||
|
||
#### 5.1.4. `S7BackgroundService.cs` (编排者)
|
||
|
||
作为 `IHostedService` 运行，负责从数据库加载激活的S7设备，并为每个设备创建和管理 `S7DeviceAgent` 实例。它还监听配置变更消息，以触发Agent的热重载。
|
||
|
||
```csharp
|
||
// 文件: DMS.Infrastructure/Services/S7BackgroundService.cs
|
||
using Microsoft.Extensions.Hosting;
|
||
using DMS.Core.Interfaces;
|
||
using DMS.WPF.Services;
|
||
using CommunityToolkit.Mvvm.Messaging;
|
||
using System.Collections.Concurrent;
|
||
using System.Threading;
|
||
using System.Threading.Tasks;
|
||
using DMS.Core.Enums;
|
||
using DMS.WPF.Messages;
|
||
|
||
namespace DMS.Infrastructure.Services;
|
||
|
||
/// <summary>
|
||
/// S7后台服务编排者，作为IHostedService运行，管理所有S7设备的通信代理。
|
||
/// </summary>
|
||
public class S7BackgroundService : IHostedService
|
||
{
|
||
private readonly IRepositoryManager _repoManager;
|
||
private readonly IChannelBus _channelBus;
|
||
private readonly IMessenger _messenger;
|
||
private readonly ConcurrentDictionary<int, S7DeviceAgent> _activeAgents = new();
|
||
|
||
/// <summary>
|
||
/// 构造函数，通过依赖注入获取所需服务。
|
||
/// </summary>
|
||
public S7BackgroundService(IRepositoryManager repo, IChannelBus bus, IMessenger msg)
|
||
{
|
||
_repoManager = repo;
|
||
_channelBus = bus;
|
||
_messenger = msg;
|
||
// 注册配置变更消息，以便在设备或变量配置更新时通知Agent
|
||
_messenger.Register<ConfigChangedMessage>(this, async (r, m) => await HandleConfigChange(m));
|
||
}
|
||
|
||
/// <summary>
|
||
/// 服务启动时调用，加载所有激活的S7设备并启动其代理。
|
||
/// </summary>
|
||
public async Task StartAsync(CancellationToken cancellationToken)
|
||
{
|
||
// 获取所有激活的S7设备及其详细信息
|
||
var s7Devices = await _repoManager.Devices.GetActiveDevicesWithDetailsAsync(ProtocolType.S7);
|
||
foreach (var device in s7Devices)
|
||
{
|
||
// 为每个设备创建一个S7DeviceAgent实例
|
||
var agent = new S7DeviceAgent(device, _channelBus, _messenger);
|
||
if (_activeAgents.TryAdd(device.Id, agent))
|
||
{
|
||
await agent.StartAsync(cancellationToken);
|
||
}
|
||
}
|
||
|
||
// 启动数据处理消费者服务，它将从ChannelBus中读取数据
|
||
var dataProcessor = new DataProcessingService(_channelBus, _messenger, _repoManager);
|
||
_ = dataProcessor.StartProcessingAsync(cancellationToken); // 在后台运行，不阻塞启动
|
||
}
|
||
|
||
/// <summary>
|
||
/// 处理配置变更消息，通知相关Agent更新其变量列表。
|
||
/// </summary>
|
||
private async Task HandleConfigChange(ConfigChangedMessage message)
|
||
{
|
||
// 从数据库重新加载受影响的设备及其最新配置
|
||
var updatedDevice = await _repoManager.Devices.GetDeviceWithDetailsAsync(message.DeviceId);
|
||
if (updatedDevice != null && _activeAgents.TryGetValue(message.DeviceId, out var agent))
|
||
{
|
||
// 指示Agent使用新的变量列表进行热重载
|
||
agent.UpdateVariableLists(updatedDevice.VariableTables.SelectMany(vt => vt.Variables).ToList());
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 服务停止时调用，停止所有活动的Agent并释放资源。
|
||
/// </summary>
|
||
public async Task StopAsync(CancellationToken cancellationToken)
|
||
{
|
||
foreach (var agent in _activeAgents.Values)
|
||
{
|
||
await agent.DisposeAsync();
|
||
}
|
||
_activeAgents.Clear();
|
||
}
|
||
}
|
||
```
|
||
|
||
#### 5.1.5. `S7DeviceAgent.cs` (代理)
|
||
|
||
负责与单个PLC建立连接、维护连接、按不同频率并行轮询变量，并将读取到的数据通过 `IChannelBus` 写入数据处理队列。
|
||
|
||
```csharp
|
||
// 文件: DMS.Infrastructure/Services/Communication/S7DeviceAgent.cs
|
||
using S7.Net;
|
||
using DMS.Core.Models;
|
||
using DMS.Core.Enums;
|
||
using DMS.WPF.Services;
|
||
using CommunityToolkit.Mvvm.Messaging;
|
||
using System.Collections.Generic;
|
||
using System.Linq;
|
||
using System.Threading;
|
||
using System.Threading.Channels;
|
||
using System.Threading.Tasks;
|
||
using System;
|
||
|
||
namespace DMS.Infrastructure.Services.Communication;
|
||
|
||
/// <summary>
|
||
/// 单个S7 PLC的通信代理，负责连接、轮询和数据发送。
|
||
/// </summary>
|
||
public class S7DeviceAgent : IAsyncDisposable
|
||
{
|
||
private readonly Device _deviceConfig;
|
||
private readonly ChannelWriter<VariableContext> _processingQueueWriter;
|
||
private readonly IMessenger _messenger;
|
||
private readonly Plc _plcClient;
|
||
private CancellationTokenSource _cts; // 用于控制Agent内部的轮询任务
|
||
|
||
// 存储按轮询级别分组的变量列表
|
||
private List<Variable> _highFreqVars = new();
|
||
private List<Variable> _mediumFreqVars = new();
|
||
private List<Variable> _lowFreqVars = new();
|
||
|
||
/// <summary>
|
||
/// 构造函数。
|
||
/// </summary>
|
||
/// <param name="device">设备的配置信息。</param>
|
||
/// <param name="channelBus">中央通道总线服务。</param>
|
||
/// <param name="messenger">消息总线服务。</param>
|
||
public S7DeviceAgent(Device device, IChannelBus channelBus, IMessenger messenger)
|
||
{
|
||
_deviceConfig = device;
|
||
_messenger = messenger;
|
||
// 从中央总线获取指定名称的通道的写入端
|
||
_processingQueueWriter = channelBus.GetWriter<VariableContext>("DataProcessingQueue");
|
||
|
||
// 初始化S7.Net PLC客户端
|
||
_plcClient = new Plc(
|
||
(CpuType)Enum.Parse(typeof(CpuType), _deviceConfig.Protocol.ToString()), // 根据协议类型解析CPU类型
|
||
_deviceConfig.IpAddress,
|
||
(short)_deviceConfig.Rack,
|
||
(short)_deviceConfig.Slot
|
||
);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 启动Agent的通信循环。
|
||
/// </summary>
|
||
public async Task StartAsync(CancellationToken cancellationToken = default)
|
||
{
|
||
_cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||
await _plcClient.OpenAsync(); // 建立PLC连接
|
||
|
||
// 初始加载变量列表并分组
|
||
UpdateVariableLists(_deviceConfig.VariableTables.SelectMany(vt => vt.Variables).ToList());
|
||
|
||
// 启动并行的轮询任务，每个轮询级别一个任务
|
||
_ = Task.Run(() => PollingLoopAsync(_highFreqVars, 200, _cts.Token)); // 高频：200ms
|
||
_ = Task.Run(() => PollingLoopAsync(_mediumFreqVars, 1000, _cts.Token)); // 中频：1000ms
|
||
_ = Task.Run(() => PollingLoopAsync(_lowFreqVars, 5000, _cts.Token)); // 低频：5000ms
|
||
}
|
||
|
||
/// <summary>
|
||
/// 热重载方法，用于响应配置变更，更新Agent内部的变量列表。
|
||
/// </summary>
|
||
/// <param name="allActiveVariables">最新的所有激活变量列表。</param>
|
||
public void UpdateVariableLists(List<Variable> allActiveVariables)
|
||
{
|
||
// 重新分组变量
|
||
_highFreqVars = allActiveVariables.Where(v => v.PollLevel == PollLevelType.High).ToList();
|
||
_mediumFreqVars = allActiveVariables.Where(v => v.PollLevel == PollLevelType.Medium).ToList();
|
||
_lowFreqVars = allActiveVariables.Where(v => v.PollLevel == PollLevelType.Low).ToList();
|
||
// 可以考虑在这里重新启动轮询任务，或者让现有任务检测到列表变化
|
||
}
|
||
|
||
/// <summary>
|
||
/// 核心轮询循环，负责批量读取变量并将数据写入处理队列。
|
||
/// </summary>
|
||
private async Task PollingLoopAsync(List<Variable> varsToRead, int interval, CancellationToken token)
|
||
{
|
||
while (!token.IsCancellationRequested)
|
||
{
|
||
if (!varsToRead.Any()) // 如果没有变量需要轮询，则等待
|
||
{
|
||
await Task.Delay(interval, token);
|
||
continue;
|
||
}
|
||
|
||
try
|
||
{
|
||
// 使用 S7.Net Plus 的 ReadMultipleVarsAsync 批量读取变量
|
||
// 注意：S7.Net Plus 会将读取到的值直接更新到 Variable 对象的 DataValue 属性中
|
||
await _plcClient.ReadMultipleVarsAsync(varsToRead);
|
||
|
||
foreach (var variable in varsToRead)
|
||
{
|
||
// 将读取到的值（包含在VariableContext中）放入数据处理队列
|
||
var context = new VariableContext(variable, variable.DataValue);
|
||
await _processingQueueWriter.WriteAsync(context, token);
|
||
}
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
// 记录通信错误，但不中断整个Agent
|
||
_messenger.Send(new LogMessage(LogLevel.Error, ex, $"S7DeviceAgent: 设备 {_deviceConfig.Name} ({_deviceConfig.IpAddress}) 轮询错误。"));
|
||
}
|
||
await Task.Delay(interval, token); // 等待下一个轮询周期
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 异步释放Agent资源，包括停止轮询任务和关闭PLC连接。
|
||
/// </summary>
|
||
public async ValueTask DisposeAsync()
|
||
{
|
||
_cts?.Cancel(); // 取消所有内部轮询任务
|
||
if (_plcClient.IsConnected)
|
||
{
|
||
_plcClient.Close(); // 关闭PLC连接
|
||
}
|
||
_plcClient?.Dispose();
|
||
}
|
||
}
|
||
```
|
||
|
||
### 5.2. MQTT发布服务 (`MqttPublishService.cs`)
|
||
|
||
负责将数据发布到MQTT Broker。
|
||
|
||
```csharp
|
||
// 文件: DMS.Infrastructure/Services/Communication/MqttPublishService.cs
|
||
using MQTTnet;
|
||
using MQTTnet.Client;
|
||
using MQTTnet.Client.Options;
|
||
using System.Threading;
|
||
using System.Threading.Tasks;
|
||
|
||
namespace DMS.Infrastructure.Services.Communication;
|
||
|
||
public interface IMqttPublishService
|
||
{
|
||
Task PublishAsync(MqttServer server, string topic, string payload);
|
||
}
|
||
|
||
public class MqttPublishService : IMqttPublishService
|
||
{
|
||
private readonly IMqttClient _mqttClient;
|
||
|
||
public MqttPublishService()
|
||
{
|
||
var factory = new MqttFactory();
|
||
_mqttClient = factory.CreateMqttClient();
|
||
}
|
||
|
||
public async Task PublishAsync(MqttServer server, string topic, string payload)
|
||
{
|
||
if (!_mqttClient.IsConnected)
|
||
{
|
||
var options = new MqttClientOptionsBuilder()
|
||
.WithTcpServer(server.BrokerAddress, server.Port)
|
||
.WithCredentials(server.Username, server.Password)
|
||
.Build();
|
||
await _mqttClient.ConnectAsync(options, CancellationToken.None);
|
||
}
|
||
|
||
var message = new MqttApplicationMessageBuilder()
|
||
.WithTopic(topic)
|
||
.WithPayload(payload)
|
||
.WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce)
|
||
.Build();
|
||
|
||
await _mqttClient.PublishAsync(message, CancellationToken.None);
|
||
}
|
||
}
|
||
```
|
||
|
||
### 5.3. 数据处理消费者 (`DataProcessingService.cs`)
|
||
|
||
从 `IChannelBus` 读取数据，并启动数据处理链。它作为数据处理链的入口点。
|
||
|
||
```csharp
|
||
// 文件: DMS.Infrastructure/Services/DataProcessingService.cs
|
||
using DMS.Core.Models;
|
||
using DMS.WPF.Services;
|
||
using CommunityToolkit.Mvvm.Messaging;
|
||
using System.Threading.Channels;
|
||
using System.Threading.Tasks;
|
||
using DMS.Infrastructure.Services.Processing;
|
||
using DMS.Core.Interfaces;
|
||
using DMS.WPF.Messages;
|
||
using NLog;
|
||
using Microsoft.Extensions.Caching.Memory;
|
||
|
||
namespace DMS.Infrastructure.Services;
|
||
|
||
/// <summary>
|
||
/// 数据处理消费者服务，从ChannelBus中读取变量数据并启动处理链。
|
||
/// </summary>
|
||
public class DataProcessingService
|
||
{
|
||
private readonly ChannelReader<VariableContext> _queueReader;
|
||
private readonly IMessenger _messenger;
|
||
private readonly IRepositoryManager _repoManager;
|
||
private readonly IMemoryCache _memoryCache; // 用于ChangeDetectionProcessor
|
||
private readonly IMqttPublishService _mqttPublishService; // 用于MqttPublishProcessor
|
||
private static readonly ILogger _logger = LogManager.GetCurrentClassLogger();
|
||
|
||
/// <summary>
|
||
/// 构造函数，通过依赖注入获取所需服务。
|
||
/// </summary>
|
||
public DataProcessingService(IChannelBus channelBus, IMessenger messenger, IRepositoryManager repo, IMemoryCache memoryCache, IMqttPublishService mqttPublishService)
|
||
{
|
||
// 从中央总线获取数据处理队列的读取端
|
||
_queueReader = channelBus.GetReader<VariableContext>("DataProcessingQueue");
|
||
_messenger = messenger;
|
||
_repoManager = repo;
|
||
_memoryCache = memoryCache;
|
||
_mqttPublishService = mqttPublishService;
|
||
}
|
||
|
||
/// <summary>
|
||
/// 启动数据处理循环，持续从队列中读取数据并进行处理。
|
||
/// </summary>
|
||
public async Task StartProcessingAsync(CancellationToken token)
|
||
{
|
||
await foreach (var context in _queueReader.ReadAllAsync(token))
|
||
{
|
||
try
|
||
{
|
||
// 构建并执行数据处理链
|
||
var changeDetector = new ChangeDetectionProcessor(_memoryCache);
|
||
var historyStorage = new HistoryStorageProcessor(_repoManager.VariableHistories);
|
||
var mqttPublisher = new MqttPublishProcessor(_mqttPublishService, _repoManager);
|
||
|
||
// 链式连接处理器
|
||
changeDetector.SetNext(historyStorage).SetNext(mqttPublisher);
|
||
|
||
// 启动处理
|
||
await changeDetector.ProcessAsync(context);
|
||
|
||
// 处理完成后，发送消息通知UI更新变量值
|
||
_messenger.Send(new VariableValueUpdatedMessage(context.Variable.Id, context.CurrentValue));
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_logger.Error(ex, $"数据处理链执行错误，变量ID: {context.Variable.Id}");
|
||
}
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
### 5.4. 数据处理器 (`Processing/`)
|
||
|
||
采用责任链模式（Chain of Responsibility）来处理采集到的变量数据。每个处理器（Processor）负责一个单一的数据处理步骤（如变化检测、历史存储、MQTT发布）。
|
||
|
||
#### 5.4.1. `VariableProcessorBase.cs`
|
||
|
||
```csharp
|
||
// 文件: DMS.Application/Services/Processors/VariableProcessorBase.cs
|
||
using DMS.Core.Models;
|
||
using System.Threading.Tasks;
|
||
|
||
namespace DMS.Infrastructure.Services.Processing;
|
||
|
||
public abstract class VariableProcessorBase : IVariableProcessor
|
||
{
|
||
private IVariableProcessor _next;
|
||
|
||
public IVariableProcessor SetNext(IVariableProcessor next)
|
||
{
|
||
_next = next;
|
||
return next;
|
||
}
|
||
|
||
public virtual async Task ProcessAsync(VariableContext context)
|
||
{
|
||
if (context.IsProcessingTerminated) return;
|
||
|
||
await HandleAsync(context);
|
||
|
||
if (_next != null && !context.IsProcessingTerminated)
|
||
{
|
||
await _next.ProcessAsync(context);
|
||
}
|
||
}
|
||
|
||
// 模板方法，由子类实现具体的处理逻辑
|
||
protected abstract Task HandleAsync(VariableContext context);
|
||
}
|
||
```
|
||
|
||
#### 5.4.2. `ChangeDetectionProcessor.cs`
|
||
|
||
**职责**：检测本次读取的值与上一次的值是否相同。如果相同，则终止后续处理，以节省资源。
|
||
|
||
```csharp
|
||
// 文件: DMS.Infrastructure/Services/Processors/ChangeDetectionProcessor.cs
|
||
using DMS.Core.Models;
|
||
using Microsoft.Extensions.Caching.Memory;
|
||
using System;
|
||
using System.Threading.Tasks;
|
||
|
||
namespace DMS.Infrastructure.Services.Processing;
|
||
|
||
public class ChangeDetectionProcessor : VariableProcessorBase
|
||
{
|
||
private readonly IMemoryCache _cache; // 使用内存缓存来存储上一次的值
|
||
|
||
public ChangeDetectionProcessor(IMemoryCache cache)
|
||
{
|
||
_cache = cache;
|
||
}
|
||
|
||
protected override Task HandleAsync(VariableContext context)
|
||
{
|
||
var lastValue = _cache.Get(context.Variable.Id);
|
||
if (lastValue != null && lastValue.Equals(context.RawValue))
|
||
{
|
||
context.IsProcessingTerminated = true; // 值未变化，终止处理
|
||
}
|
||
else
|
||
{
|
||
context.IsValueChanged = true;
|
||
_cache.Set(context.Variable.Id, context.RawValue, TimeSpan.FromDays(1));
|
||
}
|
||
return Task.CompletedTask;
|
||
}
|
||
}
|
||
```
|
||
|
||
#### 5.4.3. `HistoryStorageProcessor.cs`
|
||
|
||
**职责**：将变化后的值存入数据库历史记录表。
|
||
|
||
```csharp
|
||
// 文件: DMS.Infrastructure/Services/Processors/HistoryStorageProcessor.cs
|
||
using DMS.Core.Interfaces;
|
||
using DMS.Core.Models;
|
||
using System.Threading.Tasks;
|
||
|
||
namespace DMS.Infrastructure.Services.Processing;
|
||
|
||
public class HistoryStorageProcessor : VariableProcessorBase
|
||
{
|
||
private readonly IVariableHistoryRepository _historyRepository;
|
||
|
||
public HistoryStorageProcessor(IVariableHistoryRepository historyRepository)
|
||
{
|
||
_historyRepository = historyRepository;
|
||
}
|
||
|
||
protected override async Task HandleAsync(VariableContext context)
|
||
{
|
||
if (!context.IsValueChanged) return;
|
||
|
||
var history = new VariableHistory
|
||
{
|
||
VariableId = context.Variable.Id,
|
||
Value = context.CurrentValue.ToString(),
|
||
Timestamp = context.Timestamp
|
||
};
|
||
await _historyRepository.AddAsync(history);
|
||
}
|
||
}
|
||
```
|
||
|
||
#### 5.4.4. `MqttPublishProcessor.cs`
|
||
|
||
**职责**：如果变量关联了MQTT服务器，则将值发布出去。
|
||
|
||
```csharp
|
||
// 文件: DMS.Infrastructure/Services/Processing/MqttPublishProcessor.cs
|
||
using DMS.Core.Interfaces;
|
||
using DMS.Core.Models;
|
||
using DMS.Infrastructure.Services.Communication;
|
||
using CommunityToolkit.Mvvm.Messaging;
|
||
using System.Linq;
|
||
using System.Text.Json;
|
||
using System.Threading.Tasks;
|
||
using NLog;
|
||
|
||
namespace DMS.Infrastructure.Services.Processing;
|
||
|
||
/// <summary>
|
||
/// MQTT发布处理器，负责将变量值发布到关联的MQTT服务器，并使用专属别名。
|
||
/// </summary>
|
||
public class MqttPublishProcessor : VariableProcessorBase
|
||
{
|
||
private readonly IMqttPublishService _mqttService;
|
||
private readonly IRepositoryManager _repoManager; // 使用 RepositoryManager 来获取仓储
|
||
private static readonly ILogger _logger = LogManager.GetCurrentClassLogger();
|
||
|
||
/// <summary>
|
||
/// 构造函数。
|
||
/// </summary>
|
||
public MqttPublishProcessor(IMqttPublishService mqttService, IRepositoryManager repoManager)
|
||
{
|
||
_mqttService = mqttService;
|
||
_repoManager = repoManager;
|
||
}
|
||
|
||
protected override async Task HandleAsync(VariableContext context)
|
||
{
|
||
if (!context.IsValueChanged) return; // 如果值未变化，则不发布
|
||
|
||
// 1. 从仓储获取变量及其完整的别名关联列表
|
||
// 这要求 IVariableRepository 有一个方法能加载 VariableMqttAlias 及其 MqttServer
|
||
var variableWithAliases = await _repoManager.Variables.GetVariableWithMqttAliasesAsync(context.Variable.Id);
|
||
|
||
if (variableWithAliases?.MqttAliases == null || !variableWithAliases.MqttAliases.Any())
|
||
{
|
||
return; // 没有关联的MQTT服务器，无需发布
|
||
}
|
||
|
||
foreach (var aliasInfo in variableWithAliases.MqttAliases)
|
||
{
|
||
try
|
||
{
|
||
// 确保 MqttServer 导航属性已加载且激活
|
||
var targetServer = aliasInfo.MqttServer;
|
||
if (targetServer == null || !targetServer.IsActive)
|
||
{
|
||
_logger.Warn($"MQTT发布失败：变量 {context.Variable.Name} 关联的MQTT服务器 {aliasInfo.MqttServerId} 不存在或未激活。");
|
||
continue;
|
||
}
|
||
|
||
// 使用别名构建Topic
|
||
// 示例Topic格式：DMS/DeviceName/VariableAlias
|
||
var topic = $"DMS/{context.Variable.VariableTable.Device.Name}/{aliasInfo.Alias}";
|
||
var payload = JsonSerializer.Serialize(new { value = context.CurrentValue, timestamp = context.Timestamp });
|
||
|
||
await _mqttService.PublishAsync(targetServer, topic, payload);
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_logger.Error(ex, $"MQTT发布失败：变量 {context.Variable.Name} 到服务器 {aliasInfo.MqttServer.ServerName}，别名 {aliasInfo.Alias}");
|
||
}
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
## 6. 数据映射 (`Profiles/`)
|
||
|
||
为了在领域模型 (`DMS.Core.Models`) 和数据库实体 (`DMS.Infrastructure.Entities`) 之间进行转换，我们将使用 `AutoMapper`。
|
||
|
||
```csharp
|
||
// 文件: DMS.Infrastructure/Profiles/MappingProfile.cs
|
||
using AutoMapper;
|
||
using DMS.Core.Models;
|
||
using DMS.Infrastructure.Entities;
|
||
using DMS.Core.Enums;
|
||
|
||
namespace DMS.Infrastructure.Profiles;
|
||
|
||
public class MappingProfile : Profile
|
||
{
|
||
public MappingProfile()
|
||
{
|
||
CreateMap<Device, DbDevice>().ReverseMap();
|
||
CreateMap<VariableTable, DbVariableTable>().ReverseMap();
|
||
CreateMap<Variable, DbVariable>().ReverseMap();
|
||
CreateMap<MqttServer, DbMqttServer>().ReverseMap();
|
||
CreateMap<VariableHistory, DbVariableHistory>().ReverseMap();
|
||
CreateMap<VariableMqttAlias, DbVariableMqttAlias>().ReverseMap();
|
||
CreateMap<MenuBean, DbMenu>().ReverseMap();
|
||
CreateMap<User, DbUser>().ReverseMap();
|
||
|
||
// 枚举到int的映射
|
||
CreateMap<ProtocolType, int>().ConvertUsing(src => (int)src);
|
||
CreateMap<int, ProtocolType>().ConvertUsing(src => (ProtocolType)src);
|
||
CreateMap<SignalType, int>().ConvertUsing(src => (int)src);
|
||
CreateMap<int, SignalType>().ConvertUsing(src => (SignalType)src);
|
||
CreateMap<PollLevelType, int>().ConvertUsing(src => (int)src);
|
||
CreateMap<int, PollLevelType>().ConvertUsing(src => (PollLevelType)src);
|
||
CreateMap<CSharpDataType, int>().ConvertUsing(src => (int)src);
|
||
CreateMap<int, CSharpDataType>().ConvertUsing(src => (CSharpDataType)src);
|
||
}
|
||
}
|
||
``` |