mirror of
https://gitee.com/langsisi_admin/serein-flow
synced 2026-03-02 15:50:47 +08:00
436 lines
15 KiB
C#
436 lines
15 KiB
C#
using System.Buffers.Binary;
|
||
using System.Collections.Concurrent;
|
||
using System.Net.Sockets;
|
||
using System.Threading.Channels;
|
||
|
||
|
||
namespace Serein.Proto.Modbus
|
||
{
|
||
/// <summary>
|
||
/// Modbus TCP 客户端
|
||
/// </summary>
|
||
public class ModbusTcpClient : IModbusClient
|
||
{
|
||
/// <summary>
|
||
/// 消息发送时触发的事件
|
||
/// </summary>
|
||
public Action<byte[]> OnTx { get; set; }
|
||
|
||
/// <summary>
|
||
/// 接收到消息时触发的事件
|
||
/// </summary>
|
||
public Action<byte[]> OnRx { get; set; }
|
||
|
||
/// <summary>
|
||
/// 消息通道
|
||
/// </summary>
|
||
private readonly Channel<ModbusTcpRequest> _channel = Channel.CreateUnbounded<ModbusTcpRequest>();
|
||
/// <summary>
|
||
/// TCP客户端
|
||
/// </summary>
|
||
private readonly TcpClient _tcpClient;
|
||
/// <summary>
|
||
/// TCP客户端的网络流，用于发送和接收数据
|
||
/// </summary>
|
||
private readonly NetworkStream _stream;
|
||
/// <summary>
|
||
/// 存储未完成请求的字典，键为事务ID，值为任务完成源
|
||
/// </summary>
|
||
private readonly ConcurrentDictionary<ushort, TaskCompletionSource<byte[]>> _pendingRequests = new();
|
||
/// <summary>
|
||
/// 事务ID计数器，用于生成唯一的事务ID
|
||
/// </summary>
|
||
private int _transactionId = 0;
|
||
|
||
#pragma warning disable CS8618 // 在退出构造函数时，不可为 null 的字段必须包含非 null 值。请考虑添加 "required" 修饰符或声明为可为 null。
|
||
public ModbusTcpClient(string host, int port = 502)
|
||
#pragma warning restore CS8618 // 在退出构造函数时，不可为 null 的字段必须包含非 null 值。请考虑添加 "required" 修饰符或声明为可为 null。
|
||
{
|
||
_tcpClient = new TcpClient();
|
||
_tcpClient.Connect(host, port);
|
||
_stream = _tcpClient.GetStream();
|
||
|
||
_ = ProcessQueueAsync(); // 启动后台消费者
|
||
_ = ReceiveLoopAsync(); // 启动接收响应线程
|
||
}
|
||
|
||
#region 功能码封装
|
||
|
||
/// <summary>
|
||
/// 读取线圈状态
|
||
/// </summary>
|
||
/// <param name="startAddress"></param>
|
||
/// <param name="quantity"></param>
|
||
/// <returns></returns>
|
||
public async Task<bool[]> ReadCoils(ushort startAddress, ushort quantity)
|
||
{
|
||
var pdu = BuildReadPdu(startAddress, quantity);
|
||
var responsePdu = await SendAsync(ModbusFunctionCode.ReadCoils, pdu);
|
||
return ParseDiscreteBits(responsePdu, quantity);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 读取离散输入状态
|
||
/// </summary>
|
||
/// <param name="startAddress"></param>
|
||
/// <param name="quantity"></param>
|
||
/// <returns></returns>
|
||
public async Task<bool[]> ReadDiscreteInputs(ushort startAddress, ushort quantity)
|
||
{
|
||
var pdu = BuildReadPdu(startAddress, quantity);
|
||
var responsePdu = await SendAsync(ModbusFunctionCode.ReadDiscreteInputs, pdu);
|
||
return ParseDiscreteBits(responsePdu, quantity);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 读取保持寄存器
|
||
/// </summary>
|
||
/// <param name="startAddress"></param>
|
||
/// <param name="quantity"></param>
|
||
/// <returns></returns>
|
||
public async Task<ushort[]> ReadHoldingRegisters(ushort startAddress, ushort quantity)
|
||
{
|
||
var pdu = BuildReadPdu(startAddress, quantity);
|
||
var responsePdu = await SendAsync(ModbusFunctionCode.ReadHoldingRegisters, pdu);
|
||
return ParseRegisters(responsePdu, quantity);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 读取输入寄存器
|
||
/// </summary>
|
||
/// <param name="startAddress"></param>
|
||
/// <param name="quantity"></param>
|
||
/// <returns></returns>
|
||
public async Task<ushort[]> ReadInputRegisters(ushort startAddress, ushort quantity)
|
||
{
|
||
var pdu = BuildReadPdu(startAddress, quantity);
|
||
var responsePdu = await SendAsync(ModbusFunctionCode.ReadInputRegisters, pdu);
|
||
return ParseRegisters(responsePdu, quantity);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 写单个线圈
|
||
/// </summary>
|
||
/// <param name="address"></param>
|
||
/// <param name="value"></param>
|
||
/// <returns></returns>
|
||
public async Task WriteSingleCoil(ushort address, bool value)
|
||
{
|
||
var pdu = new byte[]
|
||
{
|
||
(byte)(address >> 8), // 地址高字节
|
||
(byte)(address & 0xFF), // 地址低字节
|
||
value ? (byte)0xFF : (byte)0x00, // 线圈值高字节，高电平为0xFF，低电平为00
|
||
0x00 // 线圈值低字节
|
||
};
|
||
await SendAsync(ModbusFunctionCode.WriteSingleCoil, pdu);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 写单个寄存器
|
||
/// </summary>
|
||
/// <param name="address"></param>
|
||
/// <param name="value"></param>
|
||
/// <returns></returns>
|
||
public async Task WriteSingleRegister(ushort address, ushort value)
|
||
{
|
||
var pdu = new byte[]
|
||
{
|
||
(byte)(address >> 8), // 地址高字节
|
||
(byte)(address & 0xFF), // 地址低字节
|
||
(byte)(value >> 8), // 寄存器值高字节
|
||
(byte)(value & 0xFF) // 寄存器值低字节
|
||
};
|
||
await SendAsync(ModbusFunctionCode.WriteSingleRegister, pdu);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 写多个线圈
|
||
/// </summary>
|
||
/// <param name="startAddress"></param>
|
||
/// <param name="values"></param>
|
||
/// <returns></returns>
|
||
public async Task WriteMultipleCoils(ushort startAddress, bool[] values)
|
||
{
|
||
int byteCount = (values.Length + 7) / 8; // 计算需要的字节数
|
||
byte[] data = new byte[byteCount];
|
||
|
||
for (int i = 0; i < values.Length; i++)
|
||
{
|
||
if (values[i])
|
||
data[i / 8] |= (byte)(1 << (i % 8)); // 设置对应位
|
||
}
|
||
|
||
var pdu = new List<byte>
|
||
{
|
||
(byte)(startAddress >> 8), // 地址高字节
|
||
(byte)(startAddress & 0xFF), // 地址低字节
|
||
(byte)(values.Length >> 8), // 数量高字节
|
||
(byte)(values.Length & 0xFF), // 数量低字节
|
||
(byte)data.Length // 字节数
|
||
};
|
||
|
||
pdu.AddRange(data);
|
||
|
||
await SendAsync(ModbusFunctionCode.WriteMultipleCoils, pdu.ToArray());
|
||
}
|
||
|
||
/// <summary>
|
||
/// 写多个寄存器
|
||
/// </summary>
|
||
/// <param name="startAddress"></param>
|
||
/// <param name="values"></param>
|
||
/// <returns></returns>
|
||
public async Task WriteMultipleRegisters(ushort startAddress, ushort[] values)
|
||
{
|
||
var pdu = new List<byte>
|
||
{
|
||
(byte)(startAddress >> 8), // 地址高字节
|
||
(byte)(startAddress & 0xFF), // 地址低字节
|
||
(byte)(values.Length >> 8), // 数量高字节
|
||
(byte)(values.Length & 0xFF), // 数量低字节
|
||
(byte)(values.Length * 2) // 字节数
|
||
};
|
||
|
||
foreach (var val in values)
|
||
{
|
||
pdu.Add((byte)(val >> 8)); // 寄存器值高字节
|
||
pdu.Add((byte)(val & 0xFF)); // 寄存器值低字节
|
||
}
|
||
|
||
await SendAsync(ModbusFunctionCode.WriteMultipleRegister, pdu.ToArray());
|
||
}
|
||
|
||
/// <summary>
|
||
/// 构建读取PDU数据
|
||
/// </summary>
|
||
/// <param name="startAddress"></param>
|
||
/// <param name="quantity"></param>
|
||
/// <returns></returns>
|
||
private byte[] BuildReadPdu(ushort startAddress, ushort quantity)
|
||
{
|
||
byte[] buffer = new byte[4];
|
||
BinaryPrimitives.WriteUInt16BigEndian(buffer.AsSpan(0, 2), startAddress); // 起始地址高低字节
|
||
BinaryPrimitives.WriteUInt16BigEndian(buffer.AsSpan(2, 2), quantity); // 读取数量高低字节
|
||
return buffer;
|
||
}
|
||
|
||
|
||
/// <summary>
|
||
/// 解析离散位数据
|
||
/// </summary>
|
||
/// <param name="pdu"></param>
|
||
/// <param name="count"></param>
|
||
/// <returns></returns>
|
||
private bool[] ParseDiscreteBits(byte[] pdu, ushort count)
|
||
{
|
||
int byteCount = pdu[2]; // 第2字节是后续的字节数量
|
||
int dataIndex = 3; // 数据从第3字节开始（0-based）
|
||
|
||
var result = new bool[count];
|
||
|
||
for (int i = 0, bytePos = 0, bitPos = 0; i < count; i++, bitPos++)
|
||
{
|
||
if (bitPos == 8)
|
||
{
|
||
bitPos = 0;
|
||
bytePos++;
|
||
}
|
||
result[i] = ((pdu[dataIndex + bytePos] >> bitPos) & 0x01) != 0;
|
||
}
|
||
return result;
|
||
}
|
||
|
||
/// <summary>
|
||
/// 解析寄存器数据
|
||
/// </summary>
|
||
/// <param name="pdu"></param>
|
||
/// <param name="count"></param>
|
||
/// <returns></returns>
|
||
private ushort[] ParseRegisters(byte[] pdu, ushort count)
|
||
{
|
||
var result = new ushort[count];
|
||
int dataStart = 3; // 数据从第3字节开始
|
||
|
||
for (int i = 0; i < count; i++)
|
||
{
|
||
int offset = dataStart + i * 2;
|
||
result[i] = (ushort)((pdu[offset] << 8) | pdu[offset + 1]);
|
||
}
|
||
|
||
return result;
|
||
}
|
||
|
||
|
||
#endregion
|
||
|
||
/// <summary>
|
||
/// 处理消息队列，发送请求到服务器
|
||
/// </summary>
|
||
/// <returns></returns>
|
||
private async Task ProcessQueueAsync()
|
||
{
|
||
while (_tcpClient.Connected)
|
||
{
|
||
var request = await _channel.Reader.ReadAsync();
|
||
if (request.PDU is null)
|
||
{
|
||
request.Completion?.TrySetCanceled();
|
||
continue;
|
||
}
|
||
byte[] packet = BuildPacket(request.TransactionId, 0x01, (byte)request.FunctionCode, request.PDU);
|
||
OnTx?.Invoke(packet); // 触发发送日志
|
||
await _stream.WriteAsync(packet, 0, packet.Length);
|
||
await _stream.FlushAsync();
|
||
}
|
||
|
||
}
|
||
|
||
/// <summary>
|
||
/// 发送请求并等待响应
|
||
/// </summary>
|
||
/// <param name="functionCode">功能码</param>
|
||
/// <param name="pdu">内容</param>
|
||
/// <param name="timeout">超时时间</param>
|
||
/// <param name="maxRetries">最大重发次数</param>
|
||
/// <returns></returns>
|
||
/// <exception cref="TimeoutException"></exception>
|
||
public Task<byte[]> SendAsync(ModbusFunctionCode functionCode, byte[] pdu)
|
||
{
|
||
int id = Interlocked.Increment(ref _transactionId);
|
||
var transactionId = (ushort)(id % ushort.MaxValue); // 0~65535 循环
|
||
var tcs = new TaskCompletionSource<byte[]>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||
|
||
var request = new ModbusTcpRequest
|
||
{
|
||
TransactionId = transactionId,
|
||
FunctionCode = functionCode,
|
||
PDU = pdu,
|
||
Completion = tcs
|
||
};
|
||
_pendingRequests[transactionId] = tcs;
|
||
_channel.Writer.TryWrite(request);
|
||
|
||
return tcs.Task;
|
||
}
|
||
|
||
/// <summary>
|
||
/// 接收数据循环
|
||
/// </summary>
|
||
/// <returns></returns>
|
||
private async Task ReceiveLoopAsync()
|
||
{
|
||
var buffer = new byte[1024];
|
||
while (true)
|
||
{
|
||
int len = await _stream.ReadAsync(buffer, 0, buffer.Length).ConfigureAwait(false);
|
||
//int len = await _stream.ReadAsync(buffer.AsMemory(0, buffer.Length)).ConfigureAwait(false);
|
||
if (len == 0) return; // 连接关闭
|
||
|
||
if (len < 6)
|
||
{
|
||
Console.WriteLine("接收到的数据长度不足");
|
||
return;
|
||
}
|
||
|
||
|
||
|
||
ushort protocolId = BinaryPrimitives.ReadUInt16BigEndian(buffer.AsSpan(2, 2));
|
||
if (protocolId != 0x0000)
|
||
{
|
||
Console.WriteLine($"协议不匹配: {protocolId:X4}");
|
||
return;
|
||
}
|
||
|
||
ushort dataLength = BinaryPrimitives.ReadUInt16BigEndian(buffer.AsSpan(4, 2));
|
||
// 检查数据长度是否合法
|
||
if (dataLength > 253 || len < 6 + dataLength)
|
||
{
|
||
Console.WriteLine($"数据长度异常: dataLength={dataLength}, 实际接收={len}");
|
||
return;
|
||
}
|
||
|
||
ushort transactionId = BinaryPrimitives.ReadUInt16BigEndian(buffer.AsSpan(0, 2));
|
||
if (_pendingRequests.TryRemove(transactionId, out var tcs))
|
||
{
|
||
var responsePdu = new ReadOnlySpan<byte>(buffer, 6, dataLength).ToArray();
|
||
if (OnRx is not null)
|
||
{
|
||
var packet = new ReadOnlySpan<byte>(buffer, 0, 6 + dataLength).ToArray();
|
||
OnRx?.Invoke(packet); // 触发接收日志
|
||
}
|
||
|
||
// 检查是否异常响应
|
||
if ((responsePdu[1] & 0x80) != 0)
|
||
{
|
||
byte exceptionCode = responsePdu[2];
|
||
tcs.SetException(new ModbusException(responsePdu[1], exceptionCode));
|
||
}
|
||
else
|
||
{
|
||
tcs.SetResult(responsePdu);
|
||
}
|
||
}
|
||
else
|
||
{
|
||
Console.WriteLine($"未匹配到 TransactionId={transactionId} 的请求");
|
||
}
|
||
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 构造 Modbus Tcp 报文
|
||
/// </summary>
|
||
/// <param name="transactionId"></param>
|
||
/// <param name="unitId"></param>
|
||
/// <param name="functionCode"></param>
|
||
/// <param name="pduData"></param>
|
||
/// <returns></returns>
|
||
private byte[] BuildPacket(ushort transactionId, byte unitId, byte functionCode, byte[] pduData)
|
||
{
|
||
int pduLength = 1 + pduData.Length; // PDU 长度 = 功能码1字节 + 数据长度
|
||
int totalLength = 7 + pduLength; // MBAP头长度 = 7字节 + PDU长度
|
||
|
||
Span<byte> packet = totalLength <= 256 ? stackalloc byte[totalLength] : new byte[totalLength];
|
||
|
||
// 写入事务ID（大端序）
|
||
packet[0] = (byte)(transactionId >> 8);
|
||
packet[1] = (byte)(transactionId);
|
||
|
||
// 协议ID（固定0x0000）
|
||
packet[2] = 0;
|
||
packet[3] = 0;
|
||
|
||
// 长度（PDU长度 + 1字节UnitID）
|
||
ushort length = (ushort)(pduLength + 1);
|
||
packet[4] = (byte)(length >> 8);
|
||
packet[5] = (byte)(length);
|
||
|
||
// UnitID & 功能码
|
||
packet[6] = unitId;
|
||
packet[7] = functionCode;
|
||
|
||
// 复制PDU数据
|
||
pduData.AsSpan().CopyTo(packet.Slice(8));
|
||
|
||
return packet.ToArray();
|
||
}
|
||
|
||
|
||
|
||
public void Dispose()
|
||
{
|
||
var tcs = _pendingRequests.Values.ToArray();
|
||
foreach (var pending in tcs)
|
||
{
|
||
pending.TrySetCanceled();
|
||
}
|
||
_stream?.Close();
|
||
_tcpClient?.Close();
|
||
}
|
||
}
|
||
|
||
|
||
}
|
