异步组件和WinForms

Question

我目前正在编写一个组件来与基于以太网的设备通信，并且必须使用异步套接字。有时，当我从设备接收到特定的“命令”时，我需要为正在使用我的组件的任何程序(通常是WinForm)引发一个事件。我正在为用户创建一个示例表单，但我在允许客户端表单接收事件和修改表单时遇到了困难；我得到了典型的“跨线程操作无效:控件'listStrings‘从创建它的线程以外的线程访问”。

我尝试过阅读Implementing the Event-based Asynchronous Pattern和Walkthrough: Implementing a Component That Supports the Event-based Asynchronous Pattern，尽管它似乎并不是我所需要的，特别是在阅读第一个链接中的“实现基于事件的异步模式的机会”时。

.Net / C#更多的是一种爱好而不是专业，在这个项目中-这是我在能够完成它之前需要弄清楚的最后一部分。使用“线程安全”(我知道，每个人都认为它只意味着一件事)现有的TCP/IP组件是不是比自己尝试实现它更好？

编辑:这是我的网络类代码，向您展示我现在如何实现它。我忘了我是在哪里发现这个代码片段的，但在我添加表单之前，它一直工作得很好。

internal class Network
{
    private Device dev;
    private TcpClient client;
    private NetworkStream ns;
    private byte[] buffer = new byte[2048];
    private Queue<byte[]> _msgQ = new Queue<byte[]>();

    public Network(Device d)
    {
        dev = d;
    }

    internal void Connect(string ipAddress, int port)
    {
        client = new TcpClient();
        client.BeginConnect(ipAddress, port, new AsyncCallback(OnConnect), null);
    }

    internal byte[] getLocalIp()
    {
        return ((IPEndPoint)client.Client.LocalEndPoint).Address.GetAddressBytes();
    }

    private void OnConnect(IAsyncResult ar)
    {
        try
        {
            client.EndConnect(ar);
            ns = new NetworkStream(client.Client);
            ns.BeginRead(buffer, 0, 2048, new AsyncCallback(OnRead), null);
            while (_msgQ.Count > 0)
            {
                byte[] message = _msgQ.Dequeue();
                ns.Write(message, 0, message.Length);
            }
            dev.dvDevice._connected = true;
        }

        catch (Exception ex)
        {
            Console.WriteLine(ex.Message);
        }
    }

    internal void Disconnect()
    {
        try
        {
            client.Close();
            dev.dvDevice._connected = false;
        }

        catch (Exception ex)
        {
            Console.WriteLine(ex.Message);
        }
    }

    internal void Write(byte[] message)
    {
        if ((!client.Connected) || ns == null)
        {
            _msgQ.Enqueue(message);
            return;
        }
        ns.Write(message, 0, message.Length);
    }

    private void OnWrite(IAsyncResult ar)
    {
        try
        {
            ns.EndWrite(ar);
        }

        catch (Exception ex)
        {
            Console.WriteLine(ex.Message);
        }
    }

    private void OnRead(IAsyncResult ar)
    {
        try
        {
            int recv = ns.EndRead(ar);
            byte[] message = new byte[recv];
            Buffer.BlockCopy(buffer, 0, message, 0, recv);
            dev.dvDevice._mh.Parse(message);
            ns.BeginRead(buffer, 0, 2048, new AsyncCallback(OnRead), null);
        }

        catch (Exception ex)
        {
            Console.WriteLine(ex.Message);
        }
    }
}

Device是向客户端公开的类。它包含一个执行所有解析的MessageHandler (_mh)类。Device包含由MessageHandler在特定响应时调用的公共事件。希望这对我到目前为止所拥有的东西有所帮助；我不希望重写太多，但为了让它正确(并正常工作)，如果必要的话，我会这样做的。

EDIT (2)：我对这个库的目标是，用户根本不需要管理任何线程-所以当一个事件被引发时，比如说"ReceiveString"，用户应该能够不假思索地对它采取行动。

编辑(3)：更多代码以确保完整性。

public delegate void OnStringEvent(byte[] str);

public class Device
{
    internal struct _device
    {
        // other stuff too, but here's what's important
        public bool _connected;
        public bool _online;
        public MessageHandler _mh;
        public Network _net;
    }

    public event  OnStringEvent OnString;

    internal void ReceiveString(byte[] str)
    {            
        OnString(str);
    }

    internal _device dvDevice;
    public Device(int device_number, int system_number)
    {
        dvDevice = new _device(device_number, system_number);
        dvDevice._mh = new MessageHandler(this);
        dvDevice._net = new Network(this);
    }
}

internal class MessageHandler
{
    private Device dev;

    public MessageHandler(Device d)
    {
        dev = d;
    }

    public void Parse(byte[] message)
    {
        // The code goes through the message and does what it needs to
        // and determines what to do next - sometimes write back or something else

        // Eventually if it receives a specific command, it will do this:
        dev.ReceiveString(ParseMessage(ref _reader));
     }
}

Answer 1

我喜欢@Polity的答案，作为一个Rx的粉丝，我会说使用Rx (反应式扩展)

 //we convert a typical begin/end (IAsyncPattern) into an observable sequence
 //it returns a Func -read() that takes a byte, two ints and returns one.
 var read = Observable.FromAsyncPattern<byte[], int, int, int>
                            (networkStream.BeginRead, networkStream.EndRead)
.ObserveOn(Scheduler.Dispatcher);

// Now, you can get an IObservable instead of an IAsyncResult when calling it.
byte[] someBytes = new byte[10];
IObservable<int> observable = read(someBytes, 0, 10);

observable.Subscribe(x=> 
//x will be the returned int. You can touch UI from here.
);

根据您的代码，我可以看到另一个线程调用OnString事件，然后我假设当您订阅它时，您只是将字符串添加到列表框中。

device.OnString += new OnStringEvent(device_onstring);

void device_onstring(byte[] str)
{ 
 listStrings.Items.Add(...);//this is wrong, will give cross thread op error.
 //you do this: 
 this.Invoke(new MethodInvoker(delegate()
   {
       listStrings.Items.Add(..);
       //or anything else that touches UI
   });
 // this should refer to a form or control.
}

Answer 2

帮你自己一个忙，依靠TPL为你做同步提升。示例：

NetworkStream stream = MySocket.NetworkStream;

// creat a Task<int> returning the number of bytes read based on the Async patterned Begin- and EndRead methods of the Stream
Task<int> task = Task<int>.Factory.FromAsync(
        fs.BeginRead, fs.EndRead, data, 0, data.Length, null);

// Add the continuation, which returns a Task<string>. 
return task.ContinueWith((task) =>
{
    if (task.IsFaulted)
    {
        ExceptionTextBox.Text = task.Exception.Message;
    }
    else 
    {
        ResultTextBox.Text = string.Format("Read {0} bytes into data", task.Result);
    }
}, TaskScheduler.FromCurrentSynchronizationContext());

Answer 3

根据您的设计，您可以在两个地方处理此问题。如果该事件是从其他线程引发的，则可以通过检查处理该事件的窗体(或其他控件)的.invokeReqeuired属性，在事件处理程序中处理该事件。如果返回true，则应使用.beginInvoke方法将调用封送到适当的线程。

根据您的设计，您可以从另一端处理它，方法是向组件传递要封送到的窗体的实例。在引发事件之前，请检查.invokeRequired并封送处理调用，以便在正确的线程中引发事件。这样，使用库的代码就不必担心线程，但需要库有一个对system.windows.forms的引用。