0

我已经构建了一个这样的 json 块：

data = {
    "url" : "http://www.example.com/post",
    "postdata" : [
        { "key1" : "value1" },
        { "key2" : "value2" },
        { "key3" : "value3" },
        { "key4" : "value4" },
        { "key5" : "value5" },
        { "key6" : "value6" }
    ]
}

我正在尝试将“postdata”中的每个块并行（例如 2 个池）发布到相同的“url”。我曾尝试使用 Faraday、Typhoeus 和 Parallel，但在这样做的过程中我还没有遇到一个正常运行的用例。

理想情况下，我想使用 Typhoeus::Hydra 或 Faraday，将 'data' 对象传递给它，并使用 data['url'] 作为端点汇集 data['postdata']，但我已经空了递。大多数情况下，我遇到了需要一个数据数组的情况，例如：

[
    { "url" : "http://...",
      "data" : { "key1" : "value1" },
    { "url" : "http://...",
      "data" : { "key2" : "value2" },
    { "url" : "http://...",
      "data" : { "key3" : "value3" }
]

但我显然想避免这种重复。

最终目标：将潜在的 100 个 json 主体并行发布到同一个 url，一次通过有限的（比如 10 个）汇集。任何人都可以帮助引导我走上正确的道路吗？

免责声明：这是一个内部端点，所以没有恶意。

基于 tadman 的解决方案：

class BatchWrapper
  def initialize(data, offset)
    @data = data
    @offset = offset
  end

  def as_json
    @data['postdata'][@offset]
  end
end

q = Queue.new

data['postdata'].each_index do |i|
  q << BatchWrapper.new(data, i)
end

t = Thread.new do
  n = q.size
  n.times do |i|
    value = q.pop
    res = http.post(data['url'], value.as_json)
    puts res
    puts "consumed #{value.as_json}"
  end
end

t.join
4

1 回答 1

1

通常的策略是您将源数据重新映射为更易于消化的数据，然后将其拆分为多个工作线程或通过某种异步事件驱动的方法，例如您在 EventMachine 中使用的方法。

线程比较容易理解，然后你可以使用Queue结构来批量处理每个线程消耗的工作。

将您的工作分解为一系列对象，将这些对象塞入队列，然后启动 N 个线程来处理这些队列。

由于您使用的是线程，因此可以使用共享数据。例如，您可以有一个瘦包装对象，给定该格式的结构，捕获您发送的偏移量：

class BatchWrapper
  def initialize(data, offset)
    @data = data
    @offset = offset
  end

  def as_json
    @data['postdata'][@offset]
  end
end

然后只需为您打算提出的每个请求填充这些对象之一：

q = Queue.new

data['postdata'].each_index do |i|
  q << BatchWrapper.new(data, i)
end

然后您可以在工作人员中旋转队列：

Thread.new do
  while (wrapper = q.pop)
    # Send it...
    make_request(wrapper.to_json)
  end
end

包装器方法允许您准确组合从该主对象共享的数据以及特定于请求的数据，并对数据本身进行任何必要的重组。该as_json方法返回该to_json方法最终编码的内容，因此您可以在那里完全控制。

于 2017-01-13T07:12:44.893 回答