python - 如何使用参数化过滤创建类似 Twitter 的流 API？

Question

我正在尝试开发一个与 Twitter 的流 API ( https://dev.twitter.com/streaming/reference/post/statuses/filter ) 功能相同的数据流 API，即具有过滤功能的数据流。我正在生成大量数据并希望将其提供给客户。

我了解如何制作一个为所有客户端提供相同数据的应用程序。这相对容易。我遇到的困难来自于允许客户指定数据过滤器并向每个客户提供唯一的数据。

我的想法:

首先我想打开一个流式http请求(比如Twitter)。我可以创建一个接受带有参数的 GET 请求的端点(例如 https://stream.example.com/v1/filter.json?track=twitter )。根据这个答案Streaming API vs Rest API? ，这不容易扩展并且需要大量资源。

然后我想到使用 websockets 并让客户端提供过滤消息(例如locations=-122.75,36.8,-121.75,37.8)。但是，我找不到 WS 服务器向每个客户端提供唯一数据的好示例。如果这个类继承tornado.websocket.WebSocketHandler或类似的实现，它会是什么样子？

我还考虑将数据推送到消息服务 (RabbitMQ) 或数据库 (Redis)，并在客户端连接到其唯一 channel 时订阅客户端。 (我认为像这个问题 Any ideas how to create parameterised streaming api? )。我不知道创建独特 channel 的有效方法。这似乎也过于复杂。

我更愿意在 Python 中执行此操作，但我也会考虑使用 Ruby 和 JS 实现。

Answer 1

对 Python 不太熟悉，但我认为使用 Websockets 应该可以实现这一点。这是我对Ruby的看法，希望对您有所帮助。这些是精简版本，删除了大部分 Websocket 功能，只是为了演示。

但是，对于使用流 API 的最佳实践，我恐怕无法提供太多帮助。

服务器

require 'em-websocket'
require 'json'

def filtered_stream(filter, ws)
  loop do 
    # do something with filter, transform or send different kinds of data
    ws.send "#{filter} - hello"
    sleep 2
  end
end

EM.run {
  EM::WebSocket.run(:host => "127.0.0.1", :port => 9999) do |ws|
    ws.onopen { |handshake|
      # We can access some information in the handshake object such as headers
      # Perhaps even add the client to a list / table
    }

    ws.onmessage { |msg|
      # use ws.close to close the connection if needed
      # for example if the client attempts to send an invalid message or multiple messages?

      filter = JSON.parse(msg)['locations']
      Thread.new { filtered_stream(filter, ws) }
    }
  end
}

客户端

require 'websocket-eventmachine-client'

EM.run do
  ws = WebSocket::EventMachine::Client.connect(
    :uri => 'ws://localhost:9999',
   )

  ws.onopen do
    # Unsure on how to supply the filter, headers is an option too

    ws.send "{\"locations\":\"-122.75,36.8,-121.75,37.8\"}"
  end

  ws.onmessage do |msg|
    p msg
  end
end