Ruby并发与并行：实用教程

本文概述

• 我们的测试案例
• 多进程与多线程
• 多个过程
• Ruby多线程
• 线程不是免费的
• 线程池
• 赛璐珞
• 后台工作
• 总结

首先, 我们要消除Ruby开发人员中非常常见的混淆点；即：并发和并行性不是同一件事(即, 并发！=并行)。

特别是, Ruby并发是指两个任务可以在重叠的时间段内启动, 运行和完成。不过, 这不一定意味着它们都可以同时运行(例如, 单核计算机上有多个线程)。相反, 并行性是两个任务在字面上同时运行的时间(例如, 多核处理器上的多个线程)。

这里的关键点是并发线程和/或进程不一定要并行运行。

本教程对Ruby中的并发和并行性可用的各种技术和方法进行了实用(而非理论)处理。

有关更多现实世界中的Ruby示例, 请参阅我们有关Ruby解释器和运行时的文章。

我们的测试案例

对于一个简单的测试用例, 我将创建一个Mailer类并添加一个Fibonacci函数(而不是sleep()方法), 以使每个请求都占用更多的CPU资源, 如下所示：

class Mailer

  def self.deliver(&block)
    mail = MailBuilder.new(&block).mail
    mail.send_mail
  end

  Mail = Struct.new(:from, :to, :subject, :body) do 
    def send_mail
      fib(30)
      puts "Email from: #{from}"
      puts "Email to  : #{to}"
      puts "Subject   : #{subject}"
      puts "Body      : #{body}"
    end

    def fib(n)
      n < 2 ? n : fib(n-1) + fib(n-2)
    end  
  end

  class MailBuilder
    def initialize(&block)
      @mail = Mail.new
      instance_eval(&block)
    end
    
    attr_reader :mail

    %w(from to subject body).each do |m|
      define_method(m) do |val|
        @mail.send("#{m}=", val)
      end
    end
  end
end

然后, 我们可以如下调用Mailer类来发送邮件：

Mailer.deliver do 
  from    "[email protected]"
  to      "[email protected]"
  subject "Threading and Forking"
  body    "Some content"
end

(注意：此测试用例的源代码可在github上找到。)

为了建立比较基准, 让我们先做一个简单的基准, 调用邮件100次：

puts Benchmark.measure{
  100.times do |i|
    Mailer.deliver do 
      from    "eki_#{i}@eqbalq.com"
      to      "jill_#{i}@example.com"
      subject "Threading and Forking (#{i})"
      body    "Some content"
    end
  end
}

这在使用MRI Ruby 2.0.0p353的四核处理器上产生了以下结果：

15.250000   0.020000  15.270000 ( 15.304447)

多进程与多线程

在决定是使用多个进程还是对你的Ruby应用程序进行多线程处理时, 没有”一刀切”的答案。下表总结了要考虑的一些关键因素。

使用多个进程的Ruby解决方案示例：

• Resque：Redis支持的Ruby库, 用于创建后台作业, 将其放置在多个队列中, 并在以后进行处理。
• Unicorn：用于Rack应用程序的HTTP服务器, 旨在仅在低延迟, 高带宽连接上为快速客户端提供服务, 并利用Unix / Unix类内核中的功能。

使用多线程的Ruby解决方案示例：

• Sidekiq：Ruby的全功能后台处理框架。它旨在与任何现代Rails应用程序轻松集成, 并具有比其他现有解决方案更高的性能。
• Puma：专为并发而构建的Ruby Web服务器。
• 瘦：非常快速且简单的Ruby Web服务器。

多个过程

在研究Ruby多线程选项之前, 让我们探索产生多个进程的更简单路径。

在Ruby中, fork()系统调用用于创建当前进程的”副本”。这个新进程安排在操作系统级别, 因此它可以与原始进程同时运行, 就像其他任何独立进程一样。 (注意：fork()是POSIX系统调用, 因此, 如果在Windows平台上运行Ruby, 则将不可用。)

好的, 让我们运行测试用例, 但是这次使用fork()来采用多个过程：

puts Benchmark.measure{
  100.times do |i|
    fork do     
      Mailer.deliver do 
        from    "eki_#{i}@eqbalq.com"
        to      "jill_#{i}@example.com"
        subject "Threading and Forking (#{i})"
        body    "Some content"
      end
    end
  end
  Process.waitall
}

(Process.waitall等待所有子进程退出并返回一系列进程状态。)

现在, 此代码产生以下结果(同样, 在带有MRI Ruby 2.0.0p353的四核处理器上)：

0.000000   0.030000  27.000000 (  3.788106)

不是太寒酸！通过修改几行代码(即使用fork()), 我们使邮件程序的速度提高了约5倍。

不过不要太兴奋。尽管使用分叉可能很诱人, 因为它是Ruby并发的简单解决方案, 但它的主要缺点是它将消耗大量内存。派生比较昂贵, 尤其是如果你使用的Ruby解释器未使用写时复制(CoW)。例如, 如果你的应用使用20MB内存, 则对其进行100次分叉可能会消耗多达2GB的内存！

同样, 尽管多线程也有其自身的复杂性, 但是在使用fork()时仍需要考虑许多复杂性, 例如共享文件描述符和信号灯(父子进程和子分支进程之间), 通过管道进行通信的需求, 等等。

Ruby多线程

好的, 现在让我们尝试使用Ruby多线程技术来使同一程序更快。

单个进程中的多个线程比相应数量的进程具有更少的开销, 因为它们共享地址空间和内存。

考虑到这一点, 让我们重新看一下测试用例, 但是这次使用Ruby的Thread类：

threads = []

puts Benchmark.measure{
  100.times do |i|
    threads << Thread.new do     
      Mailer.deliver do 
        from    "eki_#{i}@eqbalq.com"
        to      "jill_#{i}@example.com"
        subject "Threading and Forking (#{i})"
        body    "Some content"
      end
    end
  end
  threads.map(&:join)
}

现在, 此代码产生以下结果(同样, 在带有MRI Ruby 2.0.0p353的四核处理器上)：

13.710000   0.040000  13.750000 ( 13.740204)

mm那肯定不是很令人印象深刻！发生什么了？为什么这产生的结果几乎与同步运行代码时得到的结果相同？

答案是许多Ruby程序员存在的祸根, 那就是全局解释器锁(GIL)。多亏了GIL, CRuby(MRI实现)才真正不支持线程化。

全局解释器锁是一种用于计算机语言解释器的机制, 用于同步线程的执行, 以便一次只能执行一个线程。即使在多核处理器上运行, 使用GIL的解释器始终始终只允许一个线程和一个线程同时执行。 Ruby MRI和CPython是具有GIL的流行解释器的两个最常见示例。

回到问题所在, 根据GIL, 我们如何利用Ruby中的多线程来提高性能？

好吧, 在MRI(CRuby)中, 不幸的答案是你基本上陷入了困境, 而多线程对你的作用很小。

但是, 对于没有大量IO的任务(例如, 需要经常在网络上等待的任务), 没有并行性的Ruby并发仍然非常有用。因此线程对于IO繁重的任务在MRI中仍然有用。毕竟, 有理由甚至在多核服务器普及之前就发明和使用了线程。

但这就是说, 如果你可以选择使用CRuby以外的版本, 则可以使用替代的Ruby实现, 例如JRuby或Rubinius, 因为它们没有GIL, 并且它们确实支持真正的并行Ruby线程。

为了证明这一点, 这是当我们运行与以前完全相同的线程版本的代码时获得的结果, 但是这次在JRuby(而不是CRuby)上运行它：

43.240000   0.140000  43.380000 (  5.655000)

现在我们在说话！

但…

线程不是免费的

多线程性能的提高可能使人们相信, 我们基本上可以无限地增加更多的线程, 以使我们的代码越来越快地运行。如果确实如此, 那确实会很好, 但是现实是线程不是免费的, 因此迟早你将耗尽资源。

举例来说, 假设我们不希望样本邮件运行100次, 而是运行10, 000次。让我们看看发生了什么：

threads = []

puts Benchmark.measure{
  10_000.times do |i|
    threads << Thread.new do     
      Mailer.deliver do 
        from    "eki_#{i}@eqbalq.com"
        to      "jill_#{i}@example.com"
        subject "Threading and Forking (#{i})"
        body    "Some content"
      end
    end
  end
  threads.map(&:join)
}

繁荣！在产生大约2, 000个线程后, OS X 10.8出现错误：

can't create Thread: Resource temporarily unavailable (ThreadError)

不出所料, 我们迟早会开始崩溃或完全耗尽资源。因此, 这种方法的可扩展性显然受到限制。

线程池

幸运的是, 有更好的方法。即线程池。

线程池是一组预先实例化的, 可重用的线程, 可根据需要执行工作。当要执行大量的短任务而不是少数较长的任务时, 线程池特别有用。这样可以避免产生大量线程的开销。

线程池的关键配置参数通常是池中的线程数。这些线程既可以一次实例化(即在创建池时), 也可以懒惰地(即根据需要, 直到创建池中的最大线程数)实例化。

当池中有一个任务要执行时, 它将任务分配给当前空闲的线程之一。如果没有线程空闲(并且已经创建了最大数量的线程), 它将等待线程完成工作并变为空闲状态, 然后将任务分配给该线程。

因此, 回到我们的示例, 我们将从使用Queue(因为它是线程安全的数据类型)开始, 并采用线程池的简单实现：

要求” ./lib/mailer”要求”基准”要求”线程”

POOL_SIZE = 10

jobs = Queue.new

10_0000.times{|i| jobs.push i}

workers = (POOL_SIZE).times.map do
  Thread.new do
    begin      
      while x = jobs.pop(true)
        Mailer.deliver do 
          from    "eki_#{x}@eqbalq.com"
          to      "jill_#{x}@example.com"
          subject "Threading and Forking (#{x})"
          body    "Some content"
        end        
      end
    rescue ThreadError
    end
  end
end

workers.map(&:join)

在上面的代码中, 我们从为需要执行的作业创建作业队列开始。我们将Queue用于此目的是因为它是线程安全的(因此, 如果多个线程同时访问它, 它将保持一致性), 从而避免了需要使用互斥锁的更复杂的实现。

然后, 我们将邮件程序的ID推送到作业队列, 并创建了10个工作线程池。

在每个工作线程中, 我们从作业队列中弹出项目。

因此, 工作线程的生命周期是不断等待任务放入作业队列并执行。

因此, 好消息是, 它可以正常工作并且可以扩展, 没有任何问题。不幸的是, 即使对于我们简单的教程, 这也相当复杂。

赛璐珞

多亏了Ruby Gem生态系统, 多线程的大部分复杂性都被整齐地封装在许多易于使用的Ruby Gems中。

赛璐ul是一个很好的例子, 赛璐rub是我最喜欢的红宝石宝石之一。赛璐oid框架是在Ruby中实现基于参与者的并发系统的一种简单干净的方法。赛璐oid使人们能够从并发对象中构建并发程序, 就像他们从顺序对象中构建顺序程序一样容易。

在本文的讨论中, 我特别关注”池”功能, 但请帮个忙, 并进行更详细的检查。使用Celluloid, 你可以构建多线程Ruby程序, 而不必担心诸如死锁之类的麻烦问题, 并且发现使用诸如Futures和Promises之类的其他更复杂的功能并不容易。

这是我们的邮件程序的多线程版本使用Celluloid的简单程度：

require "./lib/mailer"
require "benchmark"
require "celluloid"

class MailWorker
  include Celluloid

  def send_email(id)
    Mailer.deliver do 
      from    "eki_#{id}@eqbalq.com"
      to      "jill_#{id}@example.com"
      subject "Threading and Forking (#{id})"
      body    "Some content"
    end       
  end
end

mailer_pool = MailWorker.pool(size: 10)

10_000.times do |i|
  mailer_pool.async.send_email(i)
end

干净, 简单, 可扩展且强大。你还能要求什么呢？

后台工作

当然, 根据你的操作要求和约束条件, 另一个可能可行的选择是采用后台作业。存在许多Ruby Gems以支持后台处理(即, 将作业保存在队列中, 并在以后处理它们而不会阻塞当前线程)。值得注意的示例包括Sidekiq, Resque, Delayed Job和Beanstalkd。

在本文中, 我将使用Sidekiq和Redis(开源键值缓存和存储)。

首先, 让我们安装Redis并在本地运行它：

brew install redis
redis-server /usr/local/etc/redis.conf

在运行本地Redis实例的情况下, 让我们使用Sidekiq查看示例邮件程序的版本(mail_worker.rb)：

require_relative "../lib/mailer"
require "sidekiq"

class MailWorker
  include Sidekiq::Worker
  
  def perform(id)
    Mailer.deliver do 
      from    "eki_#{id}@eqbalq.com"
      to      "jill_#{id}@example.com"
      subject "Threading and Forking (#{id})"
      body    "Some content"
    end  
  end
end

我们可以使用mail_worker.rb文件触发Sidekiq：

sidekiq  -r ./mail_worker.rb

然后从IRB：

⇒  irb
>> require_relative "mail_worker"
=> true
>> 100.times{|i| MailWorker.perform_async(i)}
2014-12-20T02:42:30Z 46549 TID-ouh10w8gw INFO: Sidekiq client with redis options {}
=> 100

非常简单。只需更改工人数量, 即可轻松扩展规模。

另一个选择是使用Sucker Punch, 这是我最喜欢的异步RoR处理库之一。使用Sucker Punch的实现将非常相似。我们只需要包含SuckerPunch :: Job而不是Sidekiq :: Worker, 以及MailWorker.new.async.perform()而不是MailWorker.perform_async()。

总结

高并发不仅可以在Ruby中实现, 而且比你想象的要简单。

一种可行的方法是简单地派生一个正在运行的进程以增加其处理能力。另一种技术是利用多线程。尽管线程比进程轻, 需要更少的开销, 但是如果同时启动太多线程, 仍然会耗尽资源。在某些时候, 你可能发现有必要使用线程池。幸运的是, 通过利用许多可用的宝石, 例如赛璐oid及其Actor模型, 可以简化多线程的许多复杂性。

处理耗时过程的另一种方法是使用后台处理。有许多库和服务可让你在应用程序中实施后台作业。一些流行的工具包括数据库支持的作业框架和消息队列。

分叉, 线程化和后台处理都是可行的选择。关于使用哪一个的决定取决于你的应用程序的性质, 你的操作环境和要求。希望本教程对可用选项提供了有用的介绍。