如何在等待用户响应时在事件处理程序中对传入事件进行排队

Question

我使用 C#、.NET Framework 4.5、VS2015。

我为由黑盒模块触发的事件 (TcpRequest) 实现了一个事件处理程序。 我的事件处理程序调用一个私有方法，该方法向客户端/GUI 触发事件并等待用户的响应。 同时，黑盒模块不断触发 TcpRequest 事件。不知何故，我需要在等待时对传入事件进行排队。 一旦用户对第一个事件做出响应，队列就可以“刷新”。

这种情况是否有任何设计模式或最佳实践（在等待用户响应时排队）？我需要使用 await 什么的吗？

以下是我的代码。请随时修改它们。提前谢谢你。

public void TcpRequestHandler(int id, CONN_INFO connInfo)
{
    if (SomeCondition)
    {
        var myArgs = new MyEventArgs()
        {
            Id = id,
            ConnInfo = connInfo
        }

        // this the way I tried and I know it is wrong,
        // because it always fires event without "waiting in queue"
        lock (_eventQueue)
        {
            _eventQueue.Add(myArgs);
        }
        FireEventToClient(myArgs);
    }
}

private void FireEventToClient(MyEventArgs myArgs)
{
    EventToClient(this, myArgs);
    if (myArgs.Continue)
    {
        // "flush" the event queue
        ...
        // do other things
        ...
    }
}

public class MyEventArgs : EventArgs
{
    public int Id {get; private set;}
    public CONN_INFO ConnInfo {get; private set;}
    public bool Continue {get; set;} 
}

Answer 1

用返回值声明您的事件EventToClient，并使用返回值进行验证并刷新事件队列

Answer 2

检查一下，可能会有帮助

公共类 TCPServer { 私人 int _port; 私人 TcpListener _tcpListener; 私人布尔_运行，_处置； 私有 BinaryFormatter _bFormatter； 私有线程_connectionThread； 私有BackgroundWorker _bgwListener； 私有BackgroundWorker _bgwSender； 私有对象同步对象 = 新对象（）； 私有对象 syncsendmessageobject = new object();

    /// <summary>
    /// Constructor - Initialises the TCPServer with the given port and a tcplistener. Starts a thread to monitor the message queue
    /// </summary>
    /// <param name="port"></param>
    public TCPServer(int port)
    {
        try
        {
            this._port = port;
            this._tcpListener = new TcpListener(IPAddress.Loopback, port);
            this._running = false;
            this._bFormatter = new BinaryFormatter();
            Thread thread = new Thread(ReadQueue);
            thread.Start();
        }
        catch (Exception ex)
        {
        }
    }

    /// <summary>
    /// Starts the tcplistener.
    /// </summary>
    public void Start()
    {
        try
        {
            if (!_running)
            {
                this.MessageReceived -= TCPServer_MessageReceived;
                this._tcpListener.Start();                   
                this._running = true;
                this._connectionThread = new Thread(new ThreadStart(ListenForClientConnections));
                this._connectionThread.Start();
                this._connectionThread.IsBackground = true;
                this.MessageReceived += TCPServer_MessageReceived;
            }
        }           
        catch (Exception ex)
        {
        }
    }        

    /// <summary>
    /// Stops the tcplistener
    /// </summary>
    public void Stop()
    {
        if (this._running)
        {
            this._tcpListener.Stop();
            this._tcpListener = null;
            this._running = false;
        }
    }

    public bool Running()
    {
        return this._running;
    }

    public void StopListening()
    {
        try
        {
            lock (this)
            {
                if (this._running)
                {
                    this._tcpListener.AcceptTcpClient().Close();
                    _running = false;
                }
            }
        }
        catch (Exception ex)
        {
        }
    }

    /// <summary>
    /// Thread body for listening for client connections
    /// </summary>
    private void ListenForClientConnections()
    {
        try
        {
            while (this._running)
            {
                lock (this)
                {
                    TcpClient connectedTcpClient = this._tcpListener.AcceptTcpClient();
                    this._bgwListener = new BackgroundWorker();
                    this._bgwListener.DoWork += _bgwListener_DoWork;
                    this._bgwListener.WorkerReportsProgress = true;
                    this._bgwListener.ProgressChanged += _bgwListener_ProgressChanged;
                    this._bgwListener.RunWorkerAsync(connectedTcpClient);
                }
            }
        }
        catch (Exception ex)
        {
        }
    }



    /// <summary>
    /// background worker listens to messages sent by clients. Categorises them based on messages/commands/applicationcommands/
    /// progress response etc..
    /// </summary>
    /// <param name="sender"></param>
    /// <param name="e"></param>
    private TcpClient clientGUI = null;
    private Queue<string> messageQueue = new Queue<string>();
    private void _bgwListener_DoWork(object sender, DoWorkEventArgs e)        
    {           
        TcpClient client = e.Argument as TcpClient;
        BackgroundWorker bgWorker = sender as BackgroundWorker;

        if (client != null)
        {
            try
            {
                while (this._running)
                {
                    // Block until an instance Message is received 

                   string message = this._bFormatter.Deserialize(client.GetStream()).ToString();

                    if (message.Equals("GUI", StringComparison.InvariantCultureIgnoreCase)
                    {
                        // Chek for first message from tcp client, if it is GUI then store it in a member variable, if you have multiple cleints then use a dictionary
                        clientGUI = client;
                    }

                    if(!clientGUI.Equals(client))
                    {
                        lock (syncobject)
                        {                   
                            messageQueue.Enqueue(message);                 
                        }
                    }                       

                }
            }
            catch (Exception ex)
            {
            }

        }
    }

    /// <summary>
    /// Thread that continuously monitors the message queue for messages and 
    /// sends them to clients
    /// </summary>
    private void ReadQueue()
    {
        try
        {
            while (true)
            {
                try
                {
                    TCPMessage message = null;

                    if (messageQueue.Count > 0)
                    {                          
                        lock (syncobject)
                        {
                            message = messageQueue.Dequeue();                               
                        }

                        if (!string.IsNullOrEmpty(message) && clientGUI != null)
                        {
                            SendMessage(clientGUI, message);
                        }

                    }
                    else { Thread.Sleep(10); }
                }
                catch (Exception ex)
                {
                }
            }
        }
        catch (Exception ex)
        {

        }           
    }


    private MessageReceivedEventHandler _messageReceived;
    public event MessageReceivedEventHandler MessageReceived
    {
        add { _messageReceived += value; }
        remove { _messageReceived -= value; }
    }

    private EventHandler<MessageEventArgs> _serverMessageSent;

    public event EventHandler<MessageEventArgs> ServerMessageSent
    {
        add { _serverMessageSent += value; }
        remove { _serverMessageSent -= value; }
    }


    private void SendMessage(TcpClient client, TCPMessage message)
    {
        try
        {
            lock (syncsendmessageobject)
                _bFormatter.Serialize(client.GetStream(), message);
        }
        catch (Exception ex)
        {
        }
    }       
}