c# 全双工异步命名管道 .NET答案

【问题标题】：c# Full Duplex Asynchronous Named Pipes .NETc# 全双工异步命名管道 .NET
【发布时间】：2016-04-01 10:00:24
【问题描述】：

我正在尝试在 2 台不同的机器（仅）上实现全双工客户端-服务器通信方案，其中每个端点（客户端或服务器）可以随时异步发送内容（非阻塞管道），而另一端会拿起它并阅读它。

除了命名管道之外，我不希望任何其他技术的答案，我知道其他技术，但我想要这个特定问题的答案。（我已经在不同的论坛上多次看到这个问题，并且我不断看到建议使用其他技术的回复。我认为这有点粗鲁？）

我读过命名管道必须是单向的，否则它们会被锁定，但我猜这可能是错误的。我认为管道是基于套接字的，我无法想象底层套接字只是单向的。

这个问题的任何答案都需要解决这些问题才能真正有用：

答案需要解决异步管道，我不能使用同步解决方案。
答案需要证明或允许管道保持打开的事实。我厌倦了阅读打开管道、传输字符串、然后立即关闭管道的示例。我想要一个假设管道保持打开并在随机时间传输大量垃圾并不断重复的答案。没有挂起。
基于 C# 的解决方案

我很抱歉听起来很苛刻和流鼻涕，但是经过几天的互联网搜索，我仍然没有找到一个好的例子，我不想使用 WFC。如果你知道这个答案的细节并且回答得很好，我敢肯定，这个话题将成为未来时代的真正赢家。如果我弄明白了，我会自己发布答案。

如果你要写“你需要使用两个管道”，请解释为什么，以及你如何知道这是真的，因为我读过的任何内容都没有解释为什么就是这样。

谢谢！

【问题讨论】：

您为什么不想将 WCF 与命名管道绑定一起使用？你见过这样的解决方案吗stackoverflow.com/questions/16432813/…
嘿，为什么这么生气，我只是问为什么你不想要wcf。这是一条评论，而不是建议的解决方案。如果您知道 wcf 的一些缺点，请分享，它可能对社区有用。
不生气，只是具体。原因是我试图避开那些被认为“已弃用”的技术，即使它们仍在使用中，尤其是如果它们不会增加大量的附加值。 WCF 并没有在 Named Pipes + JSON 之上添加很多内容，因此我宁愿不使用它。我喜欢保持“最小”。我只是来回传输几个字节，为了保持分布式系统的运行，我只需要一个命名管道！
顺便说一句，我回答了我自己的问题，它很有用而且很清楚。请不要再对原作投反对票了吗？

标签： .net asynchronous named-pipes

【解决方案1】：

您不必使用两个管道。我在网上找到了很多答案，说明您需要使用两个管道。我四处寻找，熬夜，试了又试，然后想出了怎么做，超级简单，但你必须把所有事情都做好（尤其是把事情按照正确的调用顺序），否则它就行不通了.另一个技巧是始终确保您有一个未完成的读取调用，否则它也会被锁定。在你知道有人在读之前不要写。除非您先设置了事件，否则不要启动读取调用。那种东西。

这是我正在使用的管道类。它可能不够健壮，无法处理管道错误、关闭和溢出。

好的，我不知道这里出了什么问题，但是格式有点不对！呜呜呜

namespace Squall
{
    public interface PipeSender
    {
        Task SendCommandAsync(PipeCommandPlusString pCmd);
    }

    /******************************************************************************
     * 
     * 
     * 
     * 
     ******************************************************************************/
    public class ClientPipe : BasicPipe
    {
        NamedPipeClientStream m_pPipe;

        public ClientPipe(string szServerName, string szPipeName)
            : base("Client")
        {
            m_szPipeName = szPipeName; // debugging
            m_pPipe = new NamedPipeClientStream(szServerName, szPipeName, PipeDirection.InOut, PipeOptions.Asynchronous);
            base.SetPipeStream(m_pPipe); // inform base class what to read/write from
        }

        public void Connect()
        {
            Debug.WriteLine("Pipe " + FullPipeNameDebug() + " connecting to server");
            m_pPipe.Connect(); // doesn't seem to be an async method for this routine. just a timeout.
            StartReadingAsync();
        }

        // the client's pipe index is always 0
        internal override int PipeId() { return 0; }
    }

    /******************************************************************************
     * 
     * 
     * 
     * 
     ******************************************************************************/
    public class ServerPipe : BasicPipe
    {
        public event EventHandler<EventArgs> GotConnectionEvent;

        NamedPipeServerStream m_pPipe;
        int m_nPipeId;

        public ServerPipe(string szPipeName, int nPipeId)
            : base("Server")
        {
            m_szPipeName = szPipeName;
            m_nPipeId = nPipeId;
            m_pPipe = new NamedPipeServerStream(
                szPipeName,
                PipeDirection.InOut,
                NamedPipeServerStream.MaxAllowedServerInstances,
                PipeTransmissionMode.Message,
                PipeOptions.Asynchronous);
            base.SetPipeStream(m_pPipe);
            m_pPipe.BeginWaitForConnection(new AsyncCallback(StaticGotPipeConnection), this);
        }

        static void StaticGotPipeConnection(IAsyncResult pAsyncResult)
        {
            ServerPipe pThis = pAsyncResult.AsyncState as ServerPipe;
            pThis.GotPipeConnection(pAsyncResult);
        }

        void GotPipeConnection(IAsyncResult pAsyncResult)
        {
            m_pPipe.EndWaitForConnection(pAsyncResult);

            Debug.WriteLine("Server Pipe " + m_szPipeName + " got a connection");

            if (GotConnectionEvent != null)
            {
                GotConnectionEvent(this, new EventArgs());
            }

            // lodge the first read request to get us going
            //
            StartReadingAsync();
        }

        internal override int PipeId() { return m_nPipeId; }
    }

    /******************************************************************************
     * 
     * 
     * 
     * 
     ******************************************************************************/

    public abstract class BasicPipe : PipeSender
    {
        public static int MaxLen = 1024 * 1024; // why not
        protected string m_szPipeName;
        protected string m_szDebugPipeName;

        public event EventHandler<PipeEventArgs> ReadDataEvent;
        public event EventHandler<EventArgs> PipeClosedEvent;

        protected byte[] m_pPipeBuffer = new byte[BasicPipe.MaxLen];

        PipeStream m_pPipeStream;

        public BasicPipe(string szDebugPipeName)
        {
            m_szDebugPipeName = szDebugPipeName;
        }

        protected void SetPipeStream(PipeStream p)
        {
            m_pPipeStream = p;
        }

        protected string FullPipeNameDebug()
        {
            return m_szDebugPipeName + "-" + m_szPipeName;
        }

        internal abstract int PipeId();

        public void Close()
        {
            m_pPipeStream.WaitForPipeDrain();
            m_pPipeStream.Close();
            m_pPipeStream.Dispose();
            m_pPipeStream = null;
        }

        // called when Server pipe gets a connection, or when Client pipe is created
        public void StartReadingAsync()
        {
            Debug.WriteLine("Pipe " + FullPipeNameDebug() + " calling ReadAsync");

            // okay we're connected, now immediately listen for incoming buffers
            //
            byte[] pBuffer = new byte[MaxLen];
            m_pPipeStream.ReadAsync(pBuffer, 0, MaxLen).ContinueWith(t =>
            {
                Debug.WriteLine("Pipe " + FullPipeNameDebug() + " finished a read request");

                int ReadLen = t.Result;
                if (ReadLen == 0)
                {
                    Debug.WriteLine("Got a null read length, remote pipe was closed");
                    if (PipeClosedEvent != null)
                    {
                        PipeClosedEvent(this, new EventArgs());
                    }
                    return;
                }

                if (ReadDataEvent != null)
                {
                    ReadDataEvent(this, new PipeEventArgs(pBuffer, ReadLen));
                }
                else
                {
                    Debug.Assert(false, "something happened");
                }

                // lodge ANOTHER read request
                //
                StartReadingAsync();

            });
        }

        protected Task WriteByteArray(byte[] pBytes)
        {
            // this will start writing, but does it copy the memory before returning?
            return m_pPipeStream.WriteAsync(pBytes, 0, pBytes.Length);
        }

        public Task SendCommandAsync(PipeCommandPlusString pCmd)
        {
            Debug.WriteLine("Pipe " + FullPipeNameDebug() + ", writing " + pCmd.GetCommand() + "-" + pCmd.GetTransmittedString());
            string szSerializedCmd = JsonConvert.SerializeObject(pCmd);
            byte[] pSerializedCmd = Misc.StringToBytes(szSerializedCmd);
            Task t = WriteByteArray(pSerializedCmd);
            return t;
        }
    }

    /******************************************************************************
     * 
     * 
     * 
     * 
     ******************************************************************************/

    public class PipeEventArgs
    {
        public byte[] m_pData;
        public int m_nDataLen;

        public PipeEventArgs(byte[] pData, int nDataLen)
        {
            // is this a copy, or an alias copy? I can't remember right now.
            m_pData = pData;
            m_nDataLen = nDataLen;
        }
    }

    /******************************************************************************
     * if we're just going to send a string back and forth, then we can use this
     * class. It it allows us to get the bytes as a string. sort of silly.
     ******************************************************************************/

    [Serializable]
    public class PipeCommandPlusString
    {
        public string m_szCommand;  // must be public to be serialized
        public string m_szString;   // ditto

        public PipeCommandPlusString(string sz, string szString)
        {
            m_szCommand = sz;
            m_szString = szString;
        }

        public string GetCommand()
        {
            return m_szCommand;
        }

        public string GetTransmittedString()
        {
            return m_szString;
        }
    }
}

这是我的管道测试，在一个进程上运行。它也运行在两个进程上，我检查过

namespace NamedPipeTest
{
    public partial class Form1 : Form
    {
        SynchronizationContext _context;
        Thread m_pThread = null;
        volatile bool m_bDieThreadDie;
        ServerPipe m_pServerPipe;
        ClientPipe m_pClientPipe;

        public Form1()
        {
            InitializeComponent();
        }

        private void Form1_Load(object sender, EventArgs e)
        {
            _context = SynchronizationContext.Current;

            m_pServerPipe = new ServerPipe("SQUALL_PIPE", 0);
            m_pServerPipe.ReadDataEvent += M_pServerPipe_ReadDataEvent;
            m_pServerPipe.PipeClosedEvent += M_pServerPipe_PipeClosedEvent;

            // m_pThread = new Thread(StaticThreadProc);
            // m_pThread.Start( this );
        }

        private void M_pServerPipe_PipeClosedEvent(object sender, EventArgs e)
        {
            Debug.WriteLine("Server: Pipe was closed, shutting down");

            // have to post this on the main thread
            _context.Post(delegate
            {
                Close();
            }, null);
        }

        private void M_pServerPipe_ReadDataEvent(object sender, PipeEventArgs e)
        {
            // this gets called on an anonymous thread

            byte[] pBytes = e.m_pData;
            string szBytes = Misc.BytesToString(pBytes, e.m_pData.Length);
            PipeCommandPlusString pCmd = JsonConvert.DeserializeObject<PipeCommandPlusString>(szBytes);
            string szValue = pCmd.GetTransmittedString();

            if (szValue == "CONNECT")
            {
                Debug.WriteLine("Got command from client: " + pCmd.GetCommand() + "-" + pCmd.GetTransmittedString() + ", writing command back to client");
                PipeCommandPlusString pCmdToSend = new PipeCommandPlusString("SERVER", "CONNECTED");
                // fire off an async write
                Task t = m_pServerPipe.SendCommandAsync(pCmdToSend);
            }
        }

        static void StaticThreadProc(Object o)
        {
            Form1 pThis = o as Form1;
            pThis.ThreadProc();
        }

        void ThreadProc()
        {
            m_pClientPipe = new ClientPipe(".", "SQUALL_PIPE");
            m_pClientPipe.ReadDataEvent += PClientPipe_ReadDataEvent;
            m_pClientPipe.PipeClosedEvent += M_pClientPipe_PipeClosedEvent;
            m_pClientPipe.Connect();

            PipeCommandPlusString pCmd = new PipeCommandPlusString("CLIENT", "CONNECT");
            int Counter = 1;
            while (Counter++ < 10)
            {
                Debug.WriteLine("Counter = " + Counter);
                m_pClientPipe.SendCommandAsync(pCmd);
                Thread.Sleep(3000);
            }

            while (!m_bDieThreadDie)
            {
                Thread.Sleep(1000);
            }

            m_pClientPipe.ReadDataEvent -= PClientPipe_ReadDataEvent;
            m_pClientPipe.PipeClosedEvent -= M_pClientPipe_PipeClosedEvent;
            m_pClientPipe.Close();
            m_pClientPipe = null;
        }

        private void M_pClientPipe_PipeClosedEvent(object sender, EventArgs e)
        {
            // wait around for server to shut us down
        }

        private void PClientPipe_ReadDataEvent(object sender, PipeEventArgs e)
        {
            byte[] pBytes = e.m_pData;
            string szBytes = Misc.BytesToString(pBytes, e.m_nDataLen);
            PipeCommandPlusString pCmd = JsonConvert.DeserializeObject<PipeCommandPlusString>(szBytes);
            string szValue = pCmd.GetTransmittedString();

            Debug.WriteLine("Got command from server: " + pCmd.GetCommand() + "-" + pCmd.GetTransmittedString());

            if (szValue == "CONNECTED")
            {
                PipeCommandPlusString pCmdToSend = new PipeCommandPlusString("CLIENT", "DATA");
                m_pClientPipe.SendCommandAsync(pCmdToSend);
            }
        }

        private void Form1_FormClosing(object sender, FormClosingEventArgs e)
        {
            if (m_pThread != null)
            {
                m_bDieThreadDie = true;
                m_pThread.Join();
                m_bDieThreadDie = false;
            }

            m_pServerPipe.ReadDataEvent -= M_pServerPipe_ReadDataEvent;
            m_pServerPipe.PipeClosedEvent -= M_pServerPipe_PipeClosedEvent;
            m_pServerPipe.Close();
            m_pServerPipe = null;

        }
    }
}

【讨论】：

【解决方案2】：

只需将管道创建为重叠，您的代码就可以在一个线程中读取时阻塞，同时从另一个线程写入管道。

    void StartServer()
    {
        Task.Factory.StartNew(() =>
        {
            var server = new NamedPipeServerStream("PipesOfPiece", PipeDirection.InOut, 1, PipeTransmissionMode.Byte, PipeOptions.Asynchronous);
            server.WaitForConnection();
            reader = new StreamReader(server);
            writer = new StreamWriter(server);
        });
    }

    private async void timer1_Tick(object sender, EventArgs e)
    {
        timer1.Stop();
        if (null != reader)
        {
            char[] buf = new char[50];

            int count = await reader.ReadAsync(buf, 0, 50);

            if (0 < count)
            {
                m_textBox_from.Text = new string(buf, 0, count);
            }
        }
        timer1.Start();
    }

【讨论】：

一方面，第一句包含了重要的一点信息（与公认的答案不同），NamedPipeServerStream 的分配证明了这一点。另一方面，我发现代码示例的其余部分......有问题...... crooked +1

【解决方案3】：

我认为当你进行异步通信时，你必须使用两个管道。

一个是接收管道，另一个是发送管道

因为，您不知道何时接收数据。

当你用一个管道发送一些数据时，recv数据不能写入管道。

相比之下，你不能在管道上写发送数据。

因此您必须使用两个管道进行异步通信。

【讨论】：

管道是套接字支持的。任何普通的unix套接字并在两端发送和接收，如果以这种方式打开的话。我认为很久以前，Unix 管道只是单向的，但那已经过去了。在 Windows 上使用全双工管道的技巧是确保事情的完成顺序，然后管道不会混淆。如果你使用两个管道，我相信它会更简单，但你只是在浪费资源和端口分配。
哦..我明白了。我不知道。谢谢你的知识..我必须学习更多。