主页 / 问题 / 1585294
Accepted
Alexandr
Asked:2024-06-26 21:04:02 +0000 UTC2024-06-26 21:04:02 +0000 UTC

C# 通过 SerialPort 接收数据并写入变量，例如队列中

  • 772

有一个设备通过串行端口随机发送数据包。我想监听端口，接收数据包中接收到的字节并将其写入变量中，以便我可以从那里获取并进一步使用它。我选择队列类型是因为我认为使用它会很方便 - 按照我收到数据包的顺序，我按照该顺序处理它们。我发现了一个问题： Problem getting data from SerialPort in C# service。我尝试了一下，它将所有内容写入文件中，我重写了它以写入队列：

    class ComWatcher
    {
        public ComWatcher(SerialPort serialPort)
        {
            port = serialPort;
        }

        SerialPort port;
        CancellationTokenSource cts;

        private Queue<byte[]> _queueOfBytes = new();
        public Queue<byte[]> QueueOfBytes
        {
            get { return _queueOfBytes; }
            set { _queueOfBytes = value; }
        }

        public async void Start()
        {
            if (cts != null)
                throw new InvalidOperationException("Already started");
            cts = new CancellationTokenSource();
            CancellationToken ct = cts.Token;
            try
            {
                if(!port.IsOpen)
                    port.Open();
                byte[] buf = new byte[16];
                var stream = port.BaseStream;

                    while (!ct.IsCancellationRequested)
                    {
                        var actuallyRead = await stream.ReadAsync(buf, 0, buf.Length, ct);
                        if (actuallyRead == 0) // end of stream
                            return;
                        await EnqueueAsync(buf);
                    }
            }
            catch (IOException io)
            {
                Console.WriteLine(io.Message);
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
            }
        }

        public void Stop()
        {
            var cts = this.cts;
            this.cts = null;
            if (cts == null)
                throw new InvalidOperationException("Not started");
            port.Close();
            cts.Cancel();
        }

        public async Task EnqueueAsync(byte[] buf)
        {
            try
            {
                await Task.Run(() => QueueOfBytes.Enqueue(buf));
            }
            catch (Exception ex)
            {
                Console.WriteLine($"{ex.Message}");
            }

        }
    }

不幸的是，Queue没有WriteAsync，所以我尝试自己写。该代码无法正常工作，因为例如数据包是这样发送的：

[01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16]

结果，可能会出现以下选项：

[01 остальные байты 00], 
[02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 01], 
[01 01 02 03 04 05 06 07 08 09 10 11 12 13 14 15]

有时，单个字节会丢失或重复，就好像它们是从其他数据包中获取的一样。我怀疑异步有问题。如何通过串行端口读取数据包并写入队列或其他数据类型？

c#

1 个回答

  1. Best Answer

    Queue<T>不是一个线程安全的集合，有一个适合这种情况的集合ConcurrentQueue<T>，但即使那样对您来说也不是很方便。错误的本质是您将一个数组放入队列中，然后覆盖其中的数据。在这种情况下，必须为循环的每次迭代创建一个缓冲区。

    我建议使用普通的线程安全生产者/消费者 - Channel<T>。

    class ComWatcher
{
    private readonly SerialPort port;
    private CancellationTokenSource cts;
    private Channel<byte[]> channel;

    public ChannelReader<byte[]> Reader => channel?.Reader ?? throw new InvalidOperationException("Not initialized");

    public ComWatcher(SerialPort serialPort)
    {
        port = serialPort;
    }

    public async void Start()
    {
        if (cts != null)
            throw new InvalidOperationException("Already started");
        channel = Channel.CreateUnbounded<byte[]>();
        using (cts = new CancellationTokenSource())
        {
            CancellationToken ct = cts.Token;
            try
            {
                if (!port.IsOpen)
                    port.Open();
                var stream = port.BaseStream;
                byte[] buf = new byte[16];

                while (true)
                {
                    var actuallyRead = await stream.ReadAsync(buf, 0, buf.Length, ct);
                    if (actuallyRead == 0) // end of stream
                        return;
                    channel.Writer.TryWrite(buf[..actuallyRead]); // здесь Range создаёт копию массива, поэтому всё будет ОК
                }
            }
            catch (IOException io)
            {
                Console.WriteLine(io.Message);
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
            }
        }
        cts = null;
        channel.Writer.Complete();
        port.Close();
    }

    public void Stop()
    {
        if (cts == null)
            throw new InvalidOperationException("Not started");            
        cts.Cancel();
    }
}

    使用这样的东西

    ComWatcher watcher = new(new SerialPort("COM1"));
watcher.Start();
await foreach (byte[] buffer in watcher.Reader.ReadAllAsync())
{
    Console.WriteLine($"[{string.Join(" ", buffer.Select(x => x.ToString("x2")))}]");
}

    调用时，Stop()循环将自行结束。顺便说一下，每 1000 个可以使用一次更厚的缓冲区。

    • 2

相关问题