aiopype 0.1.4

流量

流是aiopype的主要组件。流程是 运行管道管理器的可靠性。

Flow负责：

• 启动所有注册经理
• 处理管理器故障
• 报告错误
• 重新启动失败的管理器

经理

管理器负责注册从顶部到 底部。这意味着它必须注册一个源并连接到 消费者，直到管道最终输出。

处理器

处理器是消息的使用者/生产者。

消息处理程序

消息处理程序是允许aiopype 规模。

流将以一个或多个源作为每个源的起点 注册经理。一旦源生成事件，消息将 触发后，处理程序将识别并触发相应的 处理程序。

有两个可用的消息处理程序：

• 同步协议
• 异步协议

同步协议

同步事件处理程序，顾名思义，是同步的， 这意味着一旦源发出消息，就必须在 管道的末端和源头可以继续正常工作 行为。这有利于开发，但不能满足 允许组件所需的异步事件驱动模式 孤立。

异步协议

SyncProtocol和AsyncProtocol的主要区别在于 后者使用分离的事件循环来评估是否有新消息 在等待处理的队列中，而第一个简单地开始处理 即时收到消息。这允许完全隔离 处理器。

来源

我们的数据源是Yahoo Finance，用于从^{tt6}收集数据$ 股票价格。我们将使用aiopypeRestSource作为基类。

fromaiopype.sourcesimportRestSourceclassYahooRestSource(RestSource):"""
  Yahoo REST API source.
  """def__init__(self,name,handler,symbol):super().__init__(name,handler,'http://finance.yahoo.com/webservice/v1/symbols/{}/quote?format=json&view=detail'.format(symbol),{'exception_threshold':10,'request_interval':30})

处理器

我们的样本处理程序将从返回的 杰森。

fromaiopypeimportProcessorclassHandleRawData(Processor):defhandle(self,data,time):self.emit('price',time,data['list']['resources'][0]['resource']['fields']['price'])

输出

我们的输出处理器将价格数据写入csv文件。

importcsvclassCSVOutput(Processor):def__init__(self,name,handler,filename):super().__init__(name,handler)self.filename=filenamewithopen(self.filename,'w',newline='')ascsvfile:writer=csv.writer(csvfile,delimiter=';')writer.writerow(['time','price'])defwrite(self,time,price):withopen(self.filename,'w',newline='')ascsvfile:writer=csv.writer(csvfile,delimiter=';')writer.writerow([time,price])

经理

管理器将实例化Source、Processor和Output。 它将把Source的data事件连接到Processor.handle 处理程序和Processor'sprice事件到Output.write处理程序。 这将是我们的数据管道。

fromaiopypeimportManagerclassYahooManager(Manager):name='yahoo_apple'def__init__(self,handler):super().__init__(handler)self.processor=HandleRawData(self.build_processor_name('processor'),self.handler)self.source=YahooRestSource(self.build_processor_name('source'),self.handler,'AAPL')self.writer=CSVOutput(self.build_processor_name('writer'),self.handler,'yahoo_appl.csv')self.source.on('data',self.processor.handle)self.processor.on('price',self.writer.write)

流量

我们的流配置将只有yahoo_apple管理器。

fromaiopypeimportAsyncFlowclassFlowConfig(object):FLOWS=['yahoo_apple']dataflow=AsyncFlow(FlowConfig())

主要方法：

只需启动数据流。

if__name__=="__main__":dataflow.start()

运行示例

在名为example.py的文件中编译上述所有代码并运行：

python example.py

在制品：

这种分散的机制使得分布式管道成为可能， 如果我们有协调人在节点之间。

变更日志