在不使用跳过的情况下使用 mongoid 进行线程批处理

Question

我有一个大型 mongo db，我想抓取一批记录，在一个线程中处理它们，抓取下一批，在一个线程中处理，等等。如本文所述，.skip 中存在重大衰减@ 987654321@。我能弄清楚如何做到这一点的唯一方法是获取当前批次的最后一个 id，如下所示（这是非线程的）：

batch_size = 1000

starting_id = Person.first.id

batch = Person.where(:id.gte => starting_id).limit(batch_size)

while(batch.present?)
    batch.each do |b|
    # process
    starting_id = batch.last.id
    batch = Person.where(:id.gte => starting_id).limit(batch_size)
end

问题是，发现是缓慢的部分（相对），我真正想做的是并行化这条线（我会负责管理太多线程，所以这不是问题）：

batch = Person.where(:id.gte => starting_id).limit(batch_size)

我无法弄清楚将其放入线程中的非跳过方法，因为我必须等到慢行（上面）完成才能启动下一个线程。谁能想到一种方法来解决这个问题？这是我尝试过的，但它的性能提升几乎为零：

batch_size = 1000
starting_id = Person.first.id
thread_count = 10

keep_going = true

while(keep_going)
    batch = Person.where(:id.gte => starting_id).limit(batch_size)
    if batch.present?
        while Thread.list.count > (thread_count - 1)
            sleep(1)
        end
        Thread.new do
            batch.each do |b|
            # process
            starting_id = batch.last.id
        end
    else
        keep_going = false
    end
end

这个不太行，不过结构不是问题，主要问题是如何在mongo/mongoid中快速获取第n批记录？如果我能得到第 n 批（这是 limit 和 skip 得到的），我可以轻松并行化。

感谢您的帮助， 凯文

Answer 1

类似：

while batch.any?
  batch = Person.where(:id.gte => starting_id).order(id: :ASC).limit(batch_size)
  Thread.new{ process batch }
  starting_id = batch.last.id
end

或者，添加一个processing 键和索引，并在获取文档后更新它们。它将在单个查询中完成，因此不应太慢。至少这将是恒定的时间。批量查询将是.where(:id.gte => starting_id, :processing => nil).limit(batch_size)

主要问题是如何在mongo/mongoid中快速获取第n批记录？

我认为您不需要第 n 批，也不需要并行化查询。我相信缓慢的部分将处理每批......