同步异步事件

Question

我有一个从外部类接收标准.Net事件的类。这些事件有一个address属性(当然还有很多其他属性)，我可以使用它来同步事件，因此我应该能够创建一个方法来获取某些信息，等待正确的事件，然后返回Get方法中的事件中的数据。

但是，我对C#中的同步相当陌生，希望大家都能帮我解决这个问题。下面是我想要完成的任务的伪代码：

1. 有人叫DoAsynchronousToSynchronousCall
2. 该方法等待到具有相同地址的事件被接收(或直到超时)。
3. 该事件将检查所有当前请求。如果它找到具有相同地址的请求，请让DoAsynchronousToSynchronousCall知道答复已经到达
4. DoAsynchronousCall获取(或检索)应答并将其返回给调用方。

public class MyMessage
{
    public string Address { get; set; }
    public string Data { get; set; }
}

public Main
{
    externalClass.MessageReceived += MessageReceived;
}

public void MessageReceived(MyMessage message)
{
    MyMessage request = _requestQueue.FirstOrDefault(m => m.Address = message.Address);
    if (request != null)
    {
        // Do something to let DoAsynchronousToSynchronousCall() know the reply has arrived
    }
}

private List<MyMessage> _requestQueue = new List<MyMessage>();
public MyMessage DoAsynchronousToSynchronousCall(MyMessage message)
{
    _requestQueue.Add(message);
    externalClass.Send(message);

    // Do something to wait for a reply (as checked for above)
    MyMessage reply = WaitForCorrectReply(timeout: 10000);
    return reply;
}

我觉得我错过了一个使用async和await的机会(但我不知道如何使用)，我希望您能够理解我根据上面的信息想要完成的任务。

Answer 1

您可能正在寻找ManualResetEvent，它在线程阻塞和非阻塞行为之间切换为某种“切换”。DoAsynchronousToSynchronousCall将Reset，然后WaitOne(int timeoutMilliseconds)事件来阻塞线程，检查正确答复的事物到达时将执行Set调用，如果正确的东西到达，则让线程继续执行。

Answer 2

您不可能有多个动态调用和同步响应。如果您希望对多个调用进行同步响应，那么您也需要同步执行这些调用。

我会考虑使用微软的反应性扩展(NuGet“Rx”)来使您所做的工作尽可能简单。Rx允许您将事件转换为可以查询的值流。

我会这么做的。

我首先将接收到的消息流定义为IObservable<MyMessage> receivedMessages，如下所示：

receivedMessages =
    Observable
        .FromEvent<MessageReceivedHandler, MyMessage>(
            h => externalClass.MessageReceived += h,
            h => externalClass.MessageReceived -= h);

(您没有提供类def，所以我调用了事件委托MessageReceivedHandler。)

现在您可以将DoAsynchronousToSynchronousCall重新定义为：

public IObservable<MyMessage> DoAsynchronousCall(MyMessage message)
{
    return Observable.Create<MyMessage>(o =>
    {
        IObservable<MyMessage> result =
            receivedMessages
                .Where(m => m.Address == message.Address)
                .Take(1);

        IObservable<MyMessage> timeout =
            Observable
                .Timer(TimeSpan.FromSeconds(10.0))
                .Select(x => (MyMessage)null);

        IDisposable subscription =
            Observable
                .Amb(result, timeout)
                .Subscribe(o);

        externalClass.Send(message);

        return subscription;
    });
}

可观察到的result是为当前message.Address过滤的receivedMessages。

如果调用所用的时间比TimeSpan.FromSeconds(10.0)长，则可以观察到的timeout是返回的默认值。

最后，subscription使用Observable.Amb(...)来确定result或timeout中的哪个首先生成一个值并订阅该结果。

所以现在你可以这么做了

DoAsynchronousCall(new MyMessage() { Address = "Foo", Data = "Bar" })
    .Subscribe(response => Console.WriteLine(response.Data));

所以，如果我像这样对ExternalClass做一个简单的定义：

public class ExternalClass
{
    public event MessageReceivedHandler MessageReceived;
    public void Send(MyMessage message)
    {
        this.MessageReceived(new MyMessage()
        {
            Address = message.Address,
            Data = message.Data + "!"
        });
    }
}

...I获取控制台上打印的结果Bar!。

如果您有大量要处理的消息，您可以这样做：

var messagesToSend = new List<MyMessage>();

/* populate `messagesToSend` */

var query =
    from message in messagesToSend.ToObservable()
    from response in DoAsynchronousCall(message)
    select new
    {
        message,
        response
    };

query
    .Subscribe(x =>
    {
        /* Do something with each correctly paired
            `x.message` & `x.response`
        */
    });