Xây dựng Hệ thống Background Processing Ruby: Học từ Con số Không

Trong bài hôm nay, chúng ta sẽ triển khai một hệ thống background processing đơn giản cho vui! Chúng ta có thể học được một vài điều dọc theo đường như một cái nhìn vào nội bộ của các hệ thống background processing phổ biến như Sidekiq. Sản phẩm của sự vui vẻ này绝 không dành cho production use.

Hãy tưởng tượng chúng ta có một task trong ứng dụng load một hoặc nhiều website và trích xuất title của chúng. Vì chúng ta không có ảnh hưởng đến performance của các website này, chúng ta muốn thực hiện task đó ngoài main thread (hoặc request hiện tại — nếu chúng ta đang xây dựng web application), mà ở background.

Encapsulating a Task (Đóng gói một Task)

Trước khi đi vào background processing, hãy xây dựng một service object để thực hiện task hiện tại. Chúng ta sẽ dùng OpenURI và Nokogiri để trích xuất nội dung thẻ title.

require 'open-uri' require 'nokogiri' class TitleExtractorService def call(url) document = Nokogiri::HTML(open(url)) title = document.css('html > head > title').first.content puts title.gsub(/[[:space:]]+/, ' ').strip rescue puts "Unable to find a title for #{url}" end end

Gọi service sẽ in title của URL cho trước.

TitleExtractorService.new.call('https://appsignal.com') # AppSignal: Application Performance Monitoring for Ruby on Rails and Elixir

Điều này hoạt động như mong đợi, nhưng hãy xem nếu chúng ta có thể cải thiện syntax một chút để nó trông và cảm thấy giống các background processing systems khác. Bằng cách tạo một module Magique::Worker, chúng ta có thể thêm một chút syntactic sugar cho service object.

module Magique module Worker def self.included(base) base.extend(ClassMethods) end module ClassMethods def perform_now(*args) new.perform(*args) end end def perform(*) raise NotImplementedError end end end

Module này thêm phương thức perform vào worker instance và phương thức perform_now vào worker class để làm cho việc invocation tốt hơn một chút.

Hãy include module vào service object của chúng ta. Trong khi đang ở đó, hãy cũng đổi tên nó thành TitleExtractorWorker và đổi phương thức call thành perform.

class TitleExtractorWorker include Magique::Worker def perform(url) document = Nokogiri::HTML(open(url)) title = document.css('html > head > title').first.content puts title.gsub(/[[:space:]]+/, ' ').strip rescue puts "Unable to find a title for #{url}" end end

Việc invocation vẫn cho cùng kết quả, nhưng nó rõ ràng hơn một chút về những gì đang xảy ra.

TitleExtractorWorker.perform_now('https://appsignal.com') # AppSignal: Application Performance Monitoring for Ruby on Rails and Elixir

Triển khai Asynchronous Processing (Xử lý Bất đồng bộ)

Bây giờ chúng ta đã có title extraction hoạt động, chúng ta có thể lấy tất cả title từ các bài Ruby Magic trước. Để làm điều này, hãy giả sử chúng ta có một constant RUBYMAGIC với danh sách tất cả URL của các bài viết trước.

RUBYMAGIC.each do |url| TitleExtractorWorker.perform_now(url) end # Unraveling Classes, Instances and Metaclasses in Ruby | AppSignal Blog # Bindings and Lexical Scope in Ruby | AppSignal Blog # Building a Ruby C Extension From Scratch | AppSignal Blog # Closures in Ruby: Blocks, Procs and Lambdas | AppSignal Blog # ...

Chúng ta lấy được title của các bài trước, nhưng nó mất một chút thời gian để trích xuất tất cả. Đó là vì chúng ta chờ cho đến khi mỗi request hoàn thành trước khi chuyển sang cái tiếp theo.

Hãy cải thiện đó bằng cách giới thiệu phương thức perform_async vào worker module của chúng ta. Để tăng tốc, nó tạo một thread mới cho mỗi URL.

module Magique module Worker module ClassMethods def perform_async(*args) Thread.new { new.perform(*args) } end end end end

Sau khi đổi invocation thành TitleExtractorWorker.perform_async(url), chúng ta lấy được gần như tất cả title cùng lúc. Tuy nhiên, điều này cũng có nghĩa là chúng ta đang mở hơn 20 kết nối đến Ruby Magic blog cùng một lúc. (Xin lỗi vì làm loạn blog của các bạn! 😅)

Nếu bạn đang theo dõi với triển khai riêng và test điều này bên ngoài một process chạy lâu (như web server), đừng quên thêm gì đó như loop { sleep 1 } vào cuối script để đảm bảo process không terminate ngay lập tức.

Queueing up Tasks (Đưa Task vào Hàng đợi)

Với cách tiếp tạo thread mới cho mỗi invocation, chúng ta cuối cùng sẽ chạm giới hạn resource (cả phía chúng ta và phía website chúng ta truy cập). Vì chúng ta muốn là công dân tốt, hãy đổi triển khai thành cái gì đó asynchronous nhưng không cảm giác như denial-of-service attack.

Một cách phổ biến để giải quyết vấn đề này là dùng pattern producer/consumer. Một hoặc nhiều producer đẩy task lên queue trong khi một hoặc nhiều consumer lấy task từ queue và xử lý chúng.

Một queue về cơ bản là một danh sách các element. Về lý thuyết, một mảng đơn giản sẽ làm được việc. Tuy nhiên, vì chúng ta đang deal với concurrency, chúng ta cần đảm bảo chỉ một producer hoặc consumer có thể truy cập queue tại một thời điểm. Nếu chúng ta không cẩn trọng về điều này, mọi thứ sẽ kết thúc trong hỗn loạn — giống như hai người cố nén qua một cửa cùng lúc.

Vấn đề này được biết đến là producer-consumer problem và có nhiều giải pháp cho nó. May mắn thay, đây là một vấn đề rất phổ biến và Ruby đi kèm với một triển khai Queue thích hợp mà chúng ta có thể dùng mà không cần lo về thread synchronization.

Để dùng nó, hãy đảm bảo cả producer và consumer đều có thể truy cập queue. Chúng ta làm điều này bằng cách thêm một phương thức lớp vào module Magique và gán một instance của Queue cho nó.

module Magique def self.backend @backend end def self.backend=(backend) @backend = backend end end Magique.backend = Queue.new

Tiếp theo, chúng ta thay đổi triển khai perform_async để đẩy một task lên queue thay vì tạo thread riêng. Một task được biểu diễn như một hash bao gồm tham chiếu đến worker class cũng như các đối số truyền cho phương thức perform_async.

module Magique module Worker module ClassMethods def perform_async(*args) Magique.backend.push(worker: self, args: args) end end end end

Với đó, chúng ta xong phía producer. Tiếp theo, hãy nhìn phía consumer.

Mỗi consumer là một thread riêng lấy task từ queue và thực hiện chúng. Thay vì dừng sau một task, như thread, consumer sau đó lấy task khác từ queue và thực hiện, và cứ thế. Đây là triển khai cơ bản của một consumer gọi là Magique::Processor. Mỗi processor tạo một thread mới lặp vô hạn. Mỗi iteration, nó cố gắng lấy task mới từ queue, tạo instance mới của worker class, và gọi phương thức perform của nó với các đối số cho trước.

module Magique class Processor def self.start(concurrency = 1) concurrency.times { |n| new("Processor #{n}") } end def initialize(name) thread = Thread.new do loop do payload = Magique.backend.pop worker_class = payload[:worker] worker_class.new.perform(*payload[:args]) end end thread.name = name end end end

Ngoài vòng lặp xử lý, chúng ta thêm một phương thức tiện ích gọi là Magique::Processor.start. Điều này cho phép chúng ta spin up nhiều processor cùng lúc. Việc đặt tên cho thread không thực sự cần thiết, nhưng nó sẽ cho phép chúng ta thấy nếu mọi thứ thực sự hoạt động như mong đợi.

Hãy điều chỉnh output của TitleExtractorWorker để bao gồm tên của thread hiện tại.

puts "[#{Thread.current.name}] #{title.gsub(/[[:space:]]+/, ' ').strip}"

Để test setup background processing của chúng ta, chúng ta trước hết cần spin up một tập processors trước khi enqueue task.

Magique.backend = Queue.new Magique::Processor.start(5) RUBYMAGIC.each do |url| TitleExtractorWorker.perform_async(url) end # [Processor 3] Bindings and Lexical Scope in Ruby | AppSignal Blog # [Processor 4] Building a Ruby C Extension From Scratch | AppSignal Blog # [Processor 1] Unraveling Classes, Instances and Metaclasses in Ruby | AppSignal Blog # [Processor 0] Ruby's Hidden Gems, StringScanner | AppSignal Blog # [Processor 2] Fibers and Enumerators in Ruby: Turning Blocks Inside Out | AppSignal Blog # [Processor 4] Closures in Ruby: Blocks, Procs and Lambdas | AppSignal Blog # ...

Khi chạy, chúng ta vẫn lấy được title của tất cả bài viết. Mặc dù nó không nhanh bằng cách dùng thread riêng cho mỗi task, nó vẫn nhanh hơn triển khai ban đầu không có background processing. Nhờ tên processor được thêm, chúng ta cũng có thể confirm rằng tất cả processor đang làm việc qua queue. Bằng cách điều chỉnh số concurrent processor, có thể tìm ra cân bằng giữa tốc độ xử lý và giới hạn resource hiện có.

Mở rộng sang Multiple Processes và Machines

Cho đến nay, triển khai hiện tại của hệ thống background processing của chúng ta hoạt động đủ tốt. Nó vẫn bị giới hạn trong cùng process, tuy nhiên. Task tốn nhiều resource vẫn sẽ ảnh hưởng performance của toàn bộ process. Làm bước cuối cùng, hãy nhìn vào việc phân phối workload qua multiple processes và thậm chí multiple machines.

Queue là kết nối duy nhất giữa producer và consumer. Ngay bây giờ, nó dùng in-memory implementation. Hãy lấy thêm cảm hứng từ Sidekiq và triển khai queue dùng Redis.

Redis có hỗ trợ list cho phép chúng ta push và fetch task từ đó. Ngoài ra, Redis Ruby gem là thread-safe và các Redis command để modify list là atomic. Những thuộc tính này làm cho nó có thể dùng cho asynchronous background processing system của chúng ta mà không gặp vấn đề synchronization.

Hãy tạo một Redis-backed queue triển khai phương thức push và shift giống như Queue chúng ta dùng trước.

require 'json' require 'redis' module Magique module Backend class Redis def initialize(connection = ::Redis.new) @connection = connection end def push(job) @connection.lpush('magique:queue', JSON.dump(job)) end def shift _queue, job = @connection.brpop('magique:queue') payload = JSON.parse(job, symbolize_names: true) payload[:worker] = Object.const_get(payload[:worker]) payload end end end end

Vì Redis không biết gì về Ruby object, chúng ta phải serialize task thành JSON trước khi lưu vào database dùng command lpush thêm element vào đầu list.

Để fetch task từ queue, chúng ta dùng command brpop, lấy element cuối từ list. Nếu list rỗng, nó sẽ block cho đến khi element mới có sẵn. Đây là cách hay để pause processor khi không có task. Cuối cùng, sau khi lấy task ra khỏi Redis, chúng ta phải lookup class Ruby thật dựa trên tên worker dùng Object.const_get.

Làm bước cuối, hãy tách thành multiple processes. Phía producer, việc duy nhất cần làm là đổi backend thành Redis queue mới triển khai.

# ... Magique.backend = Magique::Backend::Redis.new RUBYMAGIC.each do |url| TitleExtractorWorker.perform_async(url) end

Phía consumer, chúng ta có thể thoát với vài dòng như sau:

# ... Magique.backend = Magique::Backend::Redis.new Magique::Processor.start(5) loop { sleep 1 }

Khi thực thi, consumer process sẽ chờ work mới đến trong queue. Khi chúng ta start producer process đẩy task vào queue, chúng ta thấy chúng được xử lý ngay lập tức.

Enjoy Responsibly và Đừng Dùng Đây trong Production

Mặc dù chúng ta giữ nó xa khỏi setup thực tế bạn sẽ dùng trong production (vậy đừng dùng!), chúng ta đã đi vài bước trong việc xây dựng background processor. Chúng ta bắt đầu bằng cách làm cho process chạy như background service. Sau đó làm nó async và dùng Queue giải quyết producer-consumer problem. Sau đó mở rộng process sang multiple processes hoặc machines dùng Redis thay vì in-memory implementation.

Như đã nói trước, đây là triển khai đơn giản của background processing system. Có nhiều thứ thiếu và không được xử lý tường minh. Những thứ này bao gồm (nhưng không giới hạn ở) error handling, multiple queues, scheduling, connection pooling, và signal handling.

Tuy nhiên, chúng ta đã vui viết cái này và hy vọng bạn thích cái nhìn dưới mui xe của background processing system. Có lẽ bạn thậm chí học được một hai điều.