如何在Ruby Celluloid线程池中获取线程返回值并汇总数据？

你的核心问题在于没有正确处理Celluloid异步任务的结果返回，同时直接跨线程修改实例变量存在线程安全风险。下面给你一套可行的解决方案，基于Celluloid的Future机制来安全收集所有线程的返回数据：

修正后的代码实现 #

1. TestRequest类：管理线程池并收集结果 #

class TestRequest
  attr_reader :returned_data

  def initialize
    @returned_data = []
  end

  def start
    urls = ['site1','site2','site3']
    existing_data = Data.pluck(:symbol, :page)
    data_pool = GetData.pool(size: 10)
    task_futures = []

    # 提交所有异步任务，保存每个任务的Future对象
    urls.each do |url|
      task_futures << data_pool.async.perform_requests(url, existing_data)
    end

    # 等待所有任务完成，并安全收集结果
    task_futures.each do |future|
      begin
        # 获取任务返回值，如果任务抛出异常这里会捕获
        result = future.value
        @returned_data << result
      rescue StandardError => e
        # 可选：处理单个任务的失败情况
        @returned_data << { url: url, error: "请求失败: #{e.message}" }
      end
    end

    # 所有任务完成后，执行后续逻辑
    process_results
  end

  private

  def process_results
    # 这里写你需要的后续操作，比如批量入库、数据清洗等
    puts "所有任务完成，共收集到#{@returned_data.size}条数据"
    p @returned_data
  end
end

2. GetData类：专注执行请求与解析 #

require 'celluloid/current'
class GetData
  include Celluloid

  def perform_requests(url, existing_data)
    # 1. 执行HTTP请求（示例用Net::HTTP，你可以替换成Faraday等线程安全的客户端）
    response = Net::HTTP.get_response(URI(url))
    
    # 2. 解析响应中的表格（示例用Nokogiri，根据实际HTML结构调整）
    doc = Nokogiri::HTML(response.body)
    table_data = doc.css('table tr').map do |row|
      row.css('td').map(&:text).reject(&:empty?)
    end

    # 3. 返回处理后的结果（可以是任意Ruby对象，比如Hash、Array）
    {
      url: url,
      parsed_table: table_data,
      fetched_at: Time.now
    }
  end
end

为什么之前的写法无效？ #

1. 线程安全问题：你之前让GetData直接修改TestRequest的@returned_data，多个异步线程同时操作同一个数组，会导致数据竞争、丢失或错乱。
2. 未等待任务完成：async调用是立即返回的，主线程会继续执行后续代码，此时异步任务可能还没完成，@returned_data自然是空的。
3. 错误的实例引用：你尝试在GetData中创建TestRequest实例，这完全是错误的逻辑——应该是主线程管理所有任务，而不是子线程反向创建主线程实例。

额外注意事项 #

• HTTP客户端线程安全：确保你使用的HTTP客户端是线程安全的（比如Faraday默认是线程安全的，Net::HTTP建议每个线程创建独立实例）。
• 异常处理：异步任务抛出的异常会被Future捕获，在调用future.value时会重新抛出，所以一定要加rescue处理单个任务的失败，避免整个流程崩溃。
• 线程池大小：size:10要根据你的服务器性能和目标网站的反爬策略调整，不要设置过大导致被封禁或资源耗尽。

内容的提问来源于stack exchange，提问作者LewlSauce