如何限制并发线程数

Question

在VIDEO_URL，有成千上万的视频可供下载。我想使用线程来完成这项工作，但一次最多只能使用十个线程。怎样重写下面的代码才能得到呢？

VIDEO_URL.each do | video |
  @workers << Thread.new{dl_video(video)}
end
@workers.each { |t| t.join }

更新

gem的线程池在工作线程超过10个后好像没有阻塞，是不是让线程池的I/O阻塞没有效果？

如果我下载没有线程池的视频，效果很好。

但是如果我用线程池下载视频，视频将不会被下载，当有 10 个工作人员时主线程应该被阻塞，但它没有。 （每个视频至少有 1 分钟的下载时间）

MAX_WORKERS = 10
@pool = Thread.pool(MAX_WORKERS)

def dl_video(video)
  File.open(video["title"], "wb") do |saved_file|
    @pool.process{
      saved_file.write open(video["link"], :allow_redirections => :safe).read
      # saved_file.write(HTTParty.get(video["link"]).parsed_response)
    }
  end
end

Answer 1

您尝试实现的是一种常用的模式，它被称为线程池。

我没有尝试过，但也许threadpool gem 或类似的东西值得研究：

require "threadpool"

pool = ThreadPool.new(10)
VIDEO_URL.each{|video| pool.process{dl_video(video)}}

Answer 2

你想要的叫做线程池。 Ruby 的线程有一个extension，其中包含此功能。

直接改编自库示例的未经测试的 sn-p：

require 'thread/pool'

# Create thread pool with up to 10 simultaneous running threads 
pool = Thread.pool(10)

VIDEO_URL.each do | video |
  # Add each download task the the thread pool
  pool.process do 
    dl_video(video)
  end
end

# Block and wait for the thread pool to run out of tasks
pool.shutdown

Answer 3

一个简单的解决方案（不涉及任何新的 gem）是启动 10 个 pop 线程并处理数组中的第一个 URL。

[].tap do |threads|
  urls = VIDEO_URLS.clone
  semaphore = Mutex.new
  number_of_threads = 10

  number_of_threads.times do
    threads << Thread.new do
      until urls.empty?        
        url = semaphore.synchronize { urls.pop }
        download_video(url)
      end
    end
  end
end.each(&:join)

另一种解决方案是将您的数组拆分为不同的切片（10 个或更少）；有不同的方法可以做到这一点。之后，每个线程可以处理每个切片。代码总体上可能更长，但如果你愿意，你可以去掉Mutex。

[].tap do |threads|
  slices # split VIDEO_URLS into required slices. leave this up to you.
  slices.each do |urls|
    threads << Thread.new do
      urls.each { |url| download_video(url) }
    end
  end
end.each(&:join)

Answer 4

您可以使用each_slice。

VIDEO_URL.each_slice(10) do | batch |
    batch.each do |video|
        @workers << Thread.new{dl_video(video)}
    end
    @workers.each { |t| t.join }
    @workers = []
end