通过 TextWriter -> TextReader 进行通信

Question

进行进程间通信的一种方法是通过进程之间的（命名）管道。

我想在两个线程之间实现相同的“队列”式通信。生产者应该编写一个文本基础命令（使用 TextWriter 或输出流）。 消费者应该从 TextReader 中读取。因为，你想一想，OutputStream/-Writer 是 InputStream/-Reader 的另一面。因此，理论上，使用 Writer 向 Reader 填充数据应该很容易。

(这里的标准方法是在线程之间有一个队列。但是我希望使用 TextReader 和 TextWriter，因为我已经有前端和后端的代码。这样通过连接控制台可以很容易地调试。 In/Console.Out 到生产者/消费者。）

我认为将作家与读者联系起来真的很容易，但我不知道该怎么做。

我可以自己写一个这样的连接，但感觉它“应该”已经存在了。

有什么想法吗？

干杯 雷夫

Answer 1

我不鼓励使用流和 TextWriter/TextReader 作为线程之间通信的有效手段。您需要为每个“队列”拥有一个流，并且为确保有效数据被完全写入或读取，您需要为每个写入或读取操作锁定该流。更好的解决方案可能是这样的：

设置一个字符串类型的队列，以及几个 ManualResetEvents。一般的想法是使用线程信号来允许两个线程在不需要锁的情况下进行通信。

public static class ThreadTest
{
    public void Main()
    {
        long exit = 0;

        Queue<string> messages = new Queue<string>();
        ManualResetEvent signal1 = new ManualResetEvent();
        ManualResetEvent signal2 = new ManualResetEvent();

        signal2.Set();

        Thread writer = new Thread(() =>
        {
            while (exit == 0)
            {
                string value = Console.ReadLine();
                if (value == "exit")
                {
                    Interlocked.Exchange(ref exit, 1);
                }
                else
                {
                    messages.Enqueue(value);
                    Console.WriteLine("Written: " + value);
                    signal1.Set();
                }

                signal2.WaitOne();
            }
        });

        Thread reader = new Thread(() =>
        {
            while (exit == 0)
            {
                signal1.WaitOne();
                signal2.Reset();

                value = messages.Dequeue();
                Console.WriteLine("Read: " + value);

                signal2.Set();
                signal1.Reset();
            }
        });

        reader.Start();
        writer.Start();
    }
}

Answer 2

我放弃了寻找“现成”的解决方案。我自己写的。 一个新的吞吐量流类，它在它的写入端接收数据，通过线程安全队列将它们发送到读取端，读取端使用接收到的数据块进行读取。

namespace My.IO
{
    public class ThrouputStream 
    {
        private InputStreamClass inputStream;
        private OutputStreamClass outputStream;

        private Queue<byte[]> queue = new Queue<byte[]>();
        private System.Threading.EventWaitHandle queueEvent = new System.Threading.EventWaitHandle(false, System.Threading.EventResetMode.AutoReset);

        public ThrouputStream()
        {
            inputStream = new InputStreamClass(this);
            outputStream = new OutputStreamClass(this);
        }

        public Stream InputStream
        {
            get { return inputStream; }
        }

        public Stream OutputStream
        {
            get { return outputStream; }
        }

        private class InputStreamClass : Stream
        {
            private readonly Queue<byte[]> queue;
            private readonly ThrouputStream parent;
            private byte[] currentBlock = null;
            private int currentBlockPos = 0;
            private Boolean closed = false;
            private int readTimeoutMs = System.Threading.Timeout.Infinite;

            public InputStreamClass(ThrouputStream parent)
            {

                this.parent = parent;
                this.queue = parent.queue;
            }

            public override bool CanRead
            {
                get { return true; }
            }

            public override bool CanSeek
            {
                get { return false; }
            }

            public override bool CanWrite
            {
                get { return false; }
            }

            public override void Flush()
            {
                // Do nothing, always flushes.
            }

            public override long Length
            {
                get { throw new NotSupportedException(); }
            }

            public override long Position
            {
                get
                {
                    throw new NotSupportedException();
                }
                set
                {
                    throw new NotSupportedException();
                }
            }

            public override bool CanTimeout
            {
                get
                {
                    return true;
                }
            }

            public override int ReadTimeout
            {
                get
                {
                    return readTimeoutMs;
                }
                set
                {
                    readTimeoutMs = value;
                }
            }

            public override int Read(byte[] buffer, int offset, int count)
            {
                if (currentBlock == null)
                {
                    int queueCount;
                    lock (queue)
                    {
                        queueCount = queue.Count;
                        if ( queueCount > 0 )
                            currentBlock = queue.Dequeue();
                    }

                    if (currentBlock == null && !parent.outputStream.IsClosed )
                    {
                        parent.queueEvent.WaitOne(readTimeoutMs);

                        lock (queue)
                        {
                            if (queue.Count == 0)
                                return 0;

                            currentBlock = queue.Dequeue();
                        }
                    }

                    currentBlockPos = 0;
                }

                if (currentBlock == null)
                    return 0;

                int read = Math.Min(count, currentBlock.Length - currentBlockPos);
                Array.Copy(currentBlock, currentBlockPos, buffer, offset, read);
                currentBlockPos += read;
                if (currentBlockPos == currentBlock.Length)
                {
                    // did read whole block
                    currentBlockPos = 0;
                    currentBlock = null;
                }

                return read;
            }

            public override long Seek(long offset, SeekOrigin origin)
            {
                throw new NotImplementedException();
            }

            public override void SetLength(long value)
            {
                throw new NotImplementedException();
            }

            public override void Write(byte[] buffer, int offset, int count)
            {
                throw new NotImplementedException();
            }

            public override void Close()
            {
                this.closed = true;
                base.Close();
            }
        }

        private class OutputStreamClass : Stream
        {
            private bool isClosed = false;

            private readonly Queue<byte[]> queue;
            private readonly ThrouputStream parent;

            public OutputStreamClass(ThrouputStream parent)
            {
                this.parent = parent;
                this.queue = parent.queue;
            }

            public override bool CanRead
            {
                get { return false; }
            }

            public override bool CanSeek
            {
                get { return false; }
            }

            public override bool CanWrite
            {
                get { return true; }
            }

            public override void Flush()
            {
                // always flush
            }

            public override long Length
            {
                get { throw new NotSupportedException(); }
            }

            public override long Position
            {
                get
                {
                    throw new NotSupportedException();
                }
                set
                {
                    throw new NotSupportedException();
                }
            }

            public override int Read(byte[] buffer, int offset, int count)
            {
                throw new NotSupportedException();
            }

            public override long Seek(long offset, SeekOrigin origin)
            {
                throw new NotSupportedException();
            }

            public override void SetLength(long value)
            {
                throw new NotSupportedException();
            }

            public override void Write(byte[] buffer, int offset, int count)
            {
                byte[] copy = new byte[count];
                Array.Copy(buffer, offset, copy, 0, count);
                lock (queue)
                {
                    queue.Enqueue(copy);
                    try
                    {
                        parent.queueEvent.Set();
                    }
                    catch (Exception)
                    { }
                }
            }

            public override void Close()
            {
                this.isClosed = true;
                base.Close();

                // Signal event, to stop waiting consumer
                try
                {
                    parent.queueEvent.Set();
                }
                catch (Exception)
                { }
            }

            public bool IsClosed
            {
                get { return isClosed; }
            }
        }

    }
}