我在“未来”的支持中遇到一些问题.
在Java / Scala Akka中,期待与Await电话同步等待.
很像.NET Task.Wait()
我的目标是支持真正的异步等待.
它现在可以工作,但是在我当前的解决方案中,延续是在错误的线程上执行的.
这是将消息传递给我的一个包含等待未来块的演员的结果.
正如你所看到的,演员总是在同一个线程上执行,而等待块在随机线程池线程上执行.
actor thread: 6 await thread 10 actor thread: 6 await thread 12 actor thread: 6 actor thread: 6 await thread 13 ...
演员使用DataFlow BufferBlock< Message>获取消息.
或者,我使用RX缓冲区来订阅消息.
它是这样配置的:
var messages = new BufferBlock<Message>() { BoundedCapacity = 100,TaskScheduler = TaskScheduler.Default,}; messages.AsObservable().Subscribe(this);
到现在为止还挺好.
但是,当我等待未来的结果.
像这样:
protected override void OnReceive(IMessage message) { .... var result = await Ask(logger,m); // This is not executed on the same thread as the above code result.match() .With<SomeMessage>(t => { Console.WriteLine("await thread {0}",System.Threading.Thread.CurrentThread.GetHashCode()); }) .Default(_ => Console.WriteLine("UnkNown message")); ...
我知道这是异步等待的正常行为,但我真的必须确保只有一个线程可以访问我的演员.
我不希望将来能够同步运行,我想象正常一样运行异步,但是我希望继续在与消息处理器/ actor相同的线程上运行.
public Task<IMessage> Ask(ActorRef actor,IMessage message) { taskcompletionsource<IMessage> result = new taskcompletionsource<IMessage>(); var future = Context.ActorOf<FutureActor>(name : Guid.NewGuid().ToString()); // once this object gets a response,// we set the result for the task completion source var futureActorRef = new FutureActorRef(result); future.Tell(new SetRespondTo(),futureActorRef); actor.Tell(message,future); return result.Task; }
解决方法
I’m making a port of the AKKA framework for .NET
甜.尽管从未触及过Java / Scala / Akka,但我还是在CodeMash ’13上进行了Akka演讲.我看到很多潜在的.NET库/框架.我希望最终能够提供微软is working on something similar(currently in a limited preview).
我怀疑尽可能多地停留在Dataflow / Rx世界中;当异步操作(每个操作具有单个启动和单个结果)时,异步最好,而Dataflow和Rx在流和订阅(具有单个起始和多个结果)时可以更好地工作.所以我的第一个反应是将缓冲区链接到具有特定调度器的ActionBlock,或者使用ObserveOn将Rx通知移动到特定的调度程序,而不是尝试在异步端执行.当然,我不太熟悉Akka API设计,所以用一粒盐就可以了.
无论如何,我的async
intro描述了调度等待延续的唯一两个可靠选项:SynchronizationContext.Current和TaskScheduler.Current.如果您的Akka端口更多是一个框架(您的代码执行托管,最终用户代码始终由代码执行),那么一个SynchronizationContext可能是有意义的.如果您的端口更多是一个库(最终用户代码根据需要托管并调用您的代码),那么TaskScheduler会更有意义.
自定义SynchronizationContext没有很多示例,因为这很罕见.我的AsyncEx library中有一个AsyncContextThread
type,它定义了该线程的SynchronizationContext和TaskScheduler.有几个自定义TaskSchedulers的例子,例如Parallel Extensions Extras和“current thread” scheduler.
原文地址:https://www.jb51.cc/csharp/93450.html
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。