如何将异步操作映射到有序的数据流并获得相同或有序的结果？

Question

我目前正在使用Twisted框架设计一个应用程序，我的计划遇到了一些障碍。你知道吗

问题的正式描述

我的应用程序有以下限制：

1. 数据按顺序到达，但异步到达。我不知道我的下一份数据何时到达
2. 数据到达的顺序必须在应用程序进程的整个生命周期内保持不变。你知道吗
3. 其他异步操作必须映射到此数据“流”上。你知道吗

对我的问题的描述可能会让人们想起函数式反应式编程（FRP）范式，这是一个公平的比较。事实上，我认为我的问题在这些术语中得到了很好的描述，我的问题可以非常准确地概括为：“我如何能够以这样一种方式利用Twisted，从而从数据流的角度进行推理？”你知道吗

更具体地说，这就是我所理解的：

1. 一个数据到达并被解包到一个自定义类的实例中，以后称为“数据实例”
2. 新到达的数据实例被附加到一个collections.deque对象，由一个定制的Stream类封装。你知道吗

3. Stream类公开了Stream.map等方法，这些方法异步地将非阻塞计算应用于：

   1. Stream实例的deque中已经存在的所有元素。你知道吗
   2. 所有未来的元素，当它们到达时。你知道吗

4. 在第3项中执行的操作的结果被附加到新的Stream对象。这是因为保留原始数据很重要，因为通常需要将几个可调用的映射到给定的流。

冒着失败的风险，我想坚持这样一个事实：映射到Stream实例的计算应该返回Deferred实例。你知道吗

问题

顺便说一句，这正是我陷入困境的地方：我可以很轻松地实现第1、2和3项，但我正在努力处理如何填充结果Stream。困难源于我无法保证流的长度，所以在等待一些异步操作完成时，数据完全有可能到达。异步操作O_i也完全有可能在O_i+n之后完成，因此我不能仅仅添加deque.append作为回调。你知道吗

那么我应该如何处理这个问题呢？有没有什么漂亮的，隐藏的扭曲我还没有发现的特点？任何一个迂回曲折的开发人员有什么想法或模式我可以应用吗？你知道吗

Answer 1

我不知道有什么妙招能帮到你。我认为您可能只需要在Stream.map实现中实现重新排序（或维护顺序，具体取决于您如何看待它）逻辑。你知道吗

如果操作i+1在操作i之前完成，那么Stream.map可能只需要保留该结果，直到操作i完成。然后它可以将结果i和i+1添加到输出Stream。你知道吗

这意味着你可能还想支持你的输入背压。重新排序要求意味着您的应用程序中有一个额外的缓冲区。您不希望允许缓冲区无限制地增长，因此当它达到某个最大大小时，您可能希望告诉向您发送输入的人您无法跟上，他们应该后退。Twisted中的IProducer和IConsumer接口现在是实现这一点的标准方法（尽管称为“tubes”的东西已经开发了一段时间，用一些更易于使用的东西来替换这些接口——但我不建议您对此屏住呼吸）。你知道吗