阿春拖拉机

阿春拖拉机

Ruby 3 is coming out at the end of 2020 🎉 and one of the most expected features is the new approach for parallel execution called Ractor (Ruby’s actor-like concurrent abstraction).

Ruby 3将于2020年底问世，而最令人期待的功能之一就是称为Ractor的并行执行新方法(Ruby的类似于actor的并发抽象)。

Here in this post, we will build a simple Sidekiq-like worker pool using Ractor.

在本文的此处，我们将使用Ractor构建一个类似于Sidekiq的简单工作池。

Remember that at this moment, Ruby 3 isn’t in a stable version yet, so the interface may be changed.

请记住，目前，Ruby 3尚未处于稳定版本，因此接口可能会更改。

为什么选择Ractor？ (Why Ractor?)

As koi explained, Ractor achieves the following goals:

正如koi解释的那样，Ractor实现了以下目标：

Parallel execution in a Ruby interpreter process;

Ruby解释程序中的并行执行； Avoid thread-safety issues (especially race issues) by limiting the object sharing;

通过限制对象共享避免线程安全问题(尤其是竞争问题)； Communication via copying and moving.

通过复制和移动进行通信。

The project described in this post is inspired by the worker pool example specified here. It depends on Redis. Here is the repository.

这篇文章中描述的项目是受此处指定的工作人员示例启发的。 这取决于Redis 。 这是存储库。

循环 (The Loops)

The program basically works running three loops with the following responsibilities:

该程序基本上可以运行三个循环，并具有以下职责：

Loop #1: Gets a job from the queue and send it to the pipe;

循环1：从队列中获取作业，并将其发送到pipe ；

Loop #2: (the pipe) Receives the job and provide it to the workers;

循环2 ：( pipe )接收工作并将其提供给workers ；

Loop #3: (a worker) Takes the job and process it.

循环3 ：(一个worker )接管并处理该工作。

This is the first loop:

这是第一个循环：

loop do job = RedisConn.conn.blpop(['queue:ractor-example'], 1) next if job.nil? FileLogger.log("#{JSON.parse(job[1]).to_s}") pipe << JSON.parse(job[1])end

The RedisConn is connecting to a Redis instance and poping a job with the blpop method. Then, it sends the message (which is a parsed JSON) to the pipe object.

RedisConn正在连接到Redis实例，并使用blpop方法blpop作业。 然后，它将消息(已解析的JSON)发送到pipe对象。

The pipe is a Ractor instance:

pipe是Ractor实例：

pipe = Ractor.new do loop do Ractor.yield Ractor.recv endend

Its responsibility is to receive the job (a copy of it) using Ractor.recv and provide it to the workers using Ractor.yield.

其职责是接收作业(复制它)使用Ractor.recv并将其提供给使用人员Ractor.yield 。

We can already see the complete push-type messaging communication with Ractor. The first loop is pushing a message to the Ractor by calling pipe <<, and then it receives the message with the Ractor.recv inside its implementation block.

我们已经可以看到与Ractor的完整的推送式消息传递通信。 第一个循环通过调用pipe <<将消息Ractor.recv ，然后它在其实现块内接收带有Ractor.recv的消息。

Looking into what happens in the workers, we can see the second type of communication (pull-type):

查看工作人员中发生的情况，我们可以看到第二种通信方式(拉式)：

workers = (1..WORKERS).map do |worker_id| Ractor.new(worker_id, pipe) do |worker_id, pipe| loop do job = pipe.take Performer.new.perform(worker_id, job) end endend

The workers receive a worker_id (just used for logging) and the pipe instance as parameters. Inside of their block, at the beginning of the loop, they start taking a job from the pipe (using #take) which is provided by Ractor.yield. Then, the job is processed, and I’ll explain how it works later.

Worker接收worker_id (仅用于记录)和pipe实例作为参数。 在循环的开头，在块的内部，他们开始从pipe获取作业(使用#take )，该pipe由Ractor.yield提供。 然后，处理该作业，稍后我将解释其工作方式。

So this is how the job is passing to each Ractor:

所以这就是工作传递给每个Ractor的方式：

The job is sent to the pipe using << (which is an alias for #send);

作业使用<< (是#send的别名)发送到pipe 。

The pipe receives the job with Ractor.recv;

pipe通过Ractor.recv接收作业；

The pipe provides the job with Ractor.yield;

pipe为作业提供Ractor.yield ;

A worker Ractor takes the job with #take.

worker #take用#take这项工作。

工人 (The Worker)

In this section, I’ll briefly explain the implementation of the enqueue and the worker classes.

在本节中，我将简要解释入队和worker类的实现。

We have a module called Worker that extends the perform_async. This method pushes a job to the Redis queue, specifying its parameters and the class responsible to handle it.

我们有一个名为Worker的module ，该module扩展了perform_async 。 此方法将作业推送到Redis队列，并指定其参数和负责处理该任务的类。

def perform_async(data) RedisConn.conn.rpush( 'queue:ractor-example', { 'klass' => self.name, 'data' => data }.to_json )end

Then we have a Fibonacci solver that includes this module, and we’ll use it later for tests:

然后，我们有一个包含此模块的斐波那契解算器，稍后我们将使用它进行测试：

class FibWorker include Worker def perform(data) fib(data['n']) end def fib(n) if n <= 1 n else fib(n - 1) + fib(n - 2) end endend

And finally, the Performer instantiates the related class and performs the job:

最后， Performer实例化相关类并执行任务：

result = Object.const_get(klass).new.perform(data)

测试！ (Testing!)

As you can see in the code snippets before, I inserted some logs for a better understanding of what’s going on when it’s running.

如您在前面的代码片段中所看到的，我插入了一些日志，以更好地了解其运行时的状况。

That’s what we’ll see in the log file:

这就是我们在日志文件中看到的：

A print when it pops a job from Redis;

从Redis弹出作业时的打印；

A print when the pipe Ractor receives a job;

pipe收到作业时的打印；

A print when the FibWorker starts to solve the job (and the worker ID);

FibWorker开始解决作业时的打印内容(以及工作人员ID)；

A print when the FibWorker finishes the job (and the result);

FibWorker完成作业(和结果)时的打印；

In my test, I initialized the loop.rb (which contains the loop that watches to the queue and instantiates the pipe and the workers), and then I pushed some jobs:

在测试中，我初始化了loop.rb (包含loop.rb队列并实例化pipe和workers的循环)，然后推送了一些作业：

(34..44).each do |n| FibWorker.perform_async({ 'n' => n })end

So it’s basically including 10 inputs.

因此，它基本上包括10个输入。

This is the result log:

这是结果日志：

19:20:10-redis-blpop-{"klass":"FibWorker","data":{"n":34}}19:20:10-pipe-recv-{"klass"=>"FibWorker", "data"=>{"n"=>34}}19:20:10-redis-blpop-{"klass":"FibWorker","data":{"n":35}}19:20:10-performer-1:FibWorker:{"n"=>34}:start19:20:10-pipe-recv-{"klass"=>"FibWorker", "data"=>{"n"=>35}}19:20:10-redis-blpop-{"klass":"FibWorker","data":{"n":36}}19:20:10-performer-2:FibWorker:{"n"=>35}:start19:20:10-pipe-recv-{"klass"=>"FibWorker", "data"=>{"n"=>36}}19:20:10-redis-blpop-{"klass":"FibWorker","data":{"n":37}}19:20:10-performer-3:FibWorker:{"n"=>36}:start19:20:10-pipe-recv-{"klass"=>"FibWorker", "data"=>{"n"=>37}}19:20:10-redis-blpop-{"klass":"FibWorker","data":{"n":38}}19:20:10-redis-blpop-{"klass":"FibWorker","data":{"n":39}}19:20:10-redis-blpop-{"klass":"FibWorker","data":{"n":40}}19:20:10-redis-blpop-{"klass":"FibWorker","data":{"n":41}}19:20:10-redis-blpop-{"klass":"FibWorker","data":{"n":42}}19:20:10-redis-blpop-{"klass":"FibWorker","data":{"n":43}}19:20:10-redis-blpop-{"klass":"FibWorker","data":{"n":44}}19:20:11-performer-1:FibWorker:{"n"=>34}:570288719:20:11-performer-1:FibWorker:{"n"=>37}:start19:20:11-pipe-recv-{"klass"=>"FibWorker", "data"=>{"n"=>38}}19:20:11-performer-2:FibWorker:{"n"=>35}:922746519:20:11-performer-2:FibWorker:{"n"=>38}:start19:20:11-pipe-recv-{"klass"=>"FibWorker", "data"=>{"n"=>39}}19:20:11-performer-3:FibWorker:{"n"=>36}:1493035219:20:11-performer-3:FibWorker:{"n"=>39}:start19:20:11-pipe-recv-{"klass"=>"FibWorker", "data"=>{"n"=>40}}19:20:13-performer-1:FibWorker:{"n"=>37}:2415781719:20:13-performer-1:FibWorker:{"n"=>40}:start19:20:13-pipe-recv-{"klass"=>"FibWorker", "data"=>{"n"=>41}}19:20:14-performer-2:FibWorker:{"n"=>38}:3908816919:20:14-performer-2:FibWorker:{"n"=>41}:start19:20:14-pipe-recv-{"klass"=>"FibWorker", "data"=>{"n"=>42}}19:20:17-performer-3:FibWorker:{"n"=>39}:6324598619:20:17-performer-3:FibWorker:{"n"=>42}:start19:20:17-pipe-recv-{"klass"=>"FibWorker", "data"=>{"n"=>43}}19:20:22-performer-1:FibWorker:{"n"=>40}:10233415519:20:22-performer-1:FibWorker:{"n"=>43}:start19:20:22-pipe-recv-{"klass"=>"FibWorker", "data"=>{"n"=>44}}19:20:30-performer-2:FibWorker:{"n"=>41}:16558014119:20:30-performer-2:FibWorker:{"n"=>44}:start19:20:47-performer-3:FibWorker:{"n"=>42}:26791429619:21:11-performer-1:FibWorker:{"n"=>43}:43349443719:21:41-performer-2:FibWorker:{"n"=>44}:701408733

As you can see, in the first second, the loop that is poping the jobs from Redis gets all the 10 jobs. Then, as the workers complete the tasks, the pipe is able to deliver the next one.

如您所见，在第一秒中，从Redis弹出作业的循环将获取所有10个作业。 然后，当工人完成任务时，管道就可以交付下一个管道。

I highlighted the steps of input number 40, just for example.

例如，我突出显示了输入数字40的步骤。

Here is another test running more jobs and using 8 workers. The operating system is using all CPUs when needed:

这是另一个测试，可以运行更多工作并使用8个工人。 操作系统在需要时使用所有CPU：

结论 (Conclusions)

It was fun to implement parallelism in Ruby using Ractor. Still, there is a huge change in the mindset because of the limiting of sharing objects.

使用Ractor在Ruby中实现并行性很有趣。 但是，由于共享对象的限制，思维方式发生了巨大变化。

I’m curious to see how it will impact the major gems of the Ruby community.

我很好奇它会如何影响Ruby社区的主要宝石。

翻译自: https://medium.com/@andresakata/background-job-processing-using-ractor-ruby-3-41c7956d14a0

阿春拖拉机

相关资源：asp.net结合aspnetpager使用SQL2005的存储过程分页