สร้าง Ruby Background Processing System: เรียนรู้จากศูนย์

ในโพสต์วันนี้ เราจะ implement naive background processing system เพื่อสนุก! เราอาจเรียนรู้บางอย่างระหว่างทางเป็นการสำรวจ internals ของ popular background processing systems เช่น Sidekiq ผลลัพธ์ของความสนุกนี้ไม่ได้ intended สำหรับ production use

ลองจินตนาการว่าเรามี task ใน application ที่ load หนึ่งหรือหลายเว็บไซต์และ extract titles ของพวกมัน เนื่องจากเราไม่มีอิทธิพลต่อ performance ของเว็บไซต์เหล่านี้ เราจึงอยาก perform task นี้ outside main thread (หรือ current request — ถ้าเราสร้าง web application) แต่ใน background

Encapsulating a Task

ก่อนที่เราจะเข้า background processing มาสร้าง service object เพื่อ perform task ที่มีอยู่ เราจะใช้ OpenURI และ Nokogiri เพื่อ extract contents ของ title tag

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

การเรียก service จะ print title ของ URL ที่ให้มา

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

สิ่งนี้ทำงานได้ตามคาด แต่ลองดูว่าเราสามารถปรับปรุง syntax เล็กน้อยเพื่อให้ดูและรู้สึกคล้าย background processing systems อื่นๆ ได้ไหม โดยการสร้างโมดูล Magique::Worker เราสามารถเพิ่ม syntactic sugar ให้ 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

โมดูลเพิ่มเมธอด perform ให้ worker instance และเมธอด perform_now ให้ worker class เพื่อให้ invocation ดีขึ้น

มา include โมดูลเข้าไปใน service object ของเรา ขณะที่เราอยู่นี้ ลอง rename มันเป็น TitleExtractorWorker และเปลี่ยนเมธอด call เป็น 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

การ invoke ยังคงผลลัพธ์เดียวกัน แต่ชัดเจนขึ้นว่ากำลังเกิดอะไรขึ้น

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

Implementing Asynchronous Processing

ตอนนี้เราได้ title extraction ทำงานแล้ว เราสามารถ grab titles ทั้งหมดจาก Ruby Magic articles ในอดีตได้ มาลองสมมติว่าเรามี constant RUBYMAGIC ที่มี list ของ URLs ทั้งหมดของ articles ในอดีต

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
# ...

เราได้ titles ของ articles ในอดีต แต่ใช้เวลานานในการ extract ทั้งหมด นั่นเพราะเรารอจนแต่ละ request เสร็จก่อนจึงไปตัวถัดไป

มาปรับปรุงด้วยการเพิ่มเมธอด perform_async ให้ worker module ของเรา เพื่อ speed things up มันสร้าง thread ใหม่สำหรับแต่ละ URL

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

หลังจากเปลี่ยน invocation เป็น TitleExtractorWorker.perform_async(url) เราได้ titles ทั้งหมดเกือบพร้อมกัน แต่สิ่งนี้ก็หมายความว่าเราเปิด connections มากกว่า 20 ไปที่ Ruby Magic blog พร้อมกัน (ขออภัยที่รบกวนบล็อกของคุณครับ! 😅)

ถ้าคุณ follow along กับ implementation ของตัวเองและทดสอบนอก long-running process (เช่น web server) อย่าลืมเพิ่ม something like loop { sleep 1 } ท้าย script เพื่อให้แน่ใจว่า process ไม่ terminate ทันที

Queueing up Tasks

ด้วย approach ของการสร้าง thread ใหม่สำหรับทุก invocation เราจะสุดท้ายก็ hit resource limits (ทั้งฝั่งเราและเว็บไซต์ที่เรา access) เนื่องจากเราอยากเป็น good citizens มาเปลี่ยน implementation เป็น something ที่ asynchronous แต่ไม่รู้สึก像 denial-of-service attack

วิธีทั่วไปในการแก้ปัญหานี้คือใช้ producer/consumer pattern หนึ่งหรือหลาย producers push tasks เข้า queue ในขณะที่หนึ่งหรือหลาย consumers take tasks จาก queue และ process พวกมัน

Queue 基本上是一個元素列表 在理論上簡單的陣列就足夠了 但是 因為我們處理併發 我們需要確保一次只有一個 producer 或 consumer 可以訪問隊列 如果我們不小心處理這一點 情況將會陷入混亂 — 就像兩個人試圖同時擠過一扇門一樣

這個問題稱為 producer-consumer problem 有多種解決方案 幸運的是這是一個非常常見的問題 Ruby 自帶適當的 Queue implementation 我們可以使用而不必擔心 thread synchronization

為了使用它 讓我們確保 producers 和 consumers 都可以訪問隊列 我們通過向 Magique 模塊添加類方法並將 Queue 的實例分配給它來實現這一點

module Magique
  def self.backend
    @backend
  end

  def self.backend=(backend)
    @backend = backend
  end
end

Magique.backend = Queue.new

接下來 我們更改 perform_async 實現以將任務推送到隊列而不是創建自己的新線程 任務表示為包含對 worker 類的引用以及傳遞給 perform_async 方法的參數的哈希

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

有了這個 我們完成了 producer 端的事情 接下來 讓我們看看 consumer 端

每個 consumer 是一個獨立的線程 從隊列中取出任務並執行它們 不像線程在一個任務後停止 consumer 然後從隊列中取出另一個任務並執行它 依此類推 這裡是 Magique::Processor 的基本實現 每個 processor 創建一個無限循環的新線程 對於每次迭代 它嘗試從隊列中獲取新任務 創建 worker 類的新實例 並使用給定的參數調用其 perform 方法

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

除了處理循環 我們添加了一個方便的方法叫 Magique::Processor.start 這允許我們一次啟動多個 processors 雖然命名線程不是真的必要 但它將允許我們看到事情是否真正按預期工作

讓我們調整 TitleExtractorWorker 的輸出以包含當前線程的名稱

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

為了測試我們的 background processing setup 我們首先需要啟動一組 processors 然後再 enqueue 我們的任務

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
# ...

當運行時 我們仍然得到所有文章的標題 雖然不如對每個任務使用單獨線程快 但仍比最初沒有 background processing 的實現快 得益於添加的 processor 名稱 我們也可以確認所有 processors 都在處理隊列 通過調整 concurrent processors 的數量 可以在處理速度和現有資源限制之間找到平衡

擴展到多進程和多機器

到目前為止 我們的 background processing system 實現效果足夠好 但它仍然限制在同一個進程中 資源密集型任務仍會影響整個進程的性能 作為最後一步 讓我們看看如何將工作負載分佈到多個進程甚至多台機器

隊列是 producers 和 consumers 之間唯一的連接 現在它使用 in-memory implementation 讓我們從 Sidekiq 獲得更多靈感 並使用 Redis 實現隊列

Redis 支持列表 允許我們推送和獲取任務 此外 Redis Ruby gem 是 thread-safe 的並且修改列表的 Redis 命令是原子的 這些屬性使得我們可以將其用於異步 background processing system 而不會遇到同步問題

讓我們創建一個 Redis-backed queue 實現 push 和 shift 方法就像我們之前使用的 Queue 一樣

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

由於 Redis 不了解 Ruby objects 我們必須將任務序列化為 JSON 然後使用 lpush 命令存儲在數據庫中 該命令將元素添加到列表前端

為了從隊列獲取任務 我們使用 brpop 命令 它從列表獲取最後一個元素 如果列表為空 它將阻塞直到新元素可用 這是一個不錯的方式來暫停我們的 processors 當沒有任務可用時 最後 在從 Redis 獲取任務後 我們必須使用 Object.const_get 根據 worker 的名稱查找真正的 Ruby 類

作為最後一步 讓我們把東西分成多個進程 在 producer 端 我們唯一要做的就是將 backend 改為新實現的 Redis queue

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

在 consumer 端 我們可以用幾行代碼搞定:

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

當執行時 consumer process 將等待新工作到達隊列 一旦我們啟動 producer process 推送任務進隊列 我們可以看到它們立即被處理

Enjoy Responsibly and Don't Use This in Production

雖然我們保持遠離真實世界的 setup 你會在 production 中使用 (所以不要用!) 但我們邁出了幾步來構建 background processor 我們開始時讓一個 process 作為 background service 運行 然後我們使其 async 並使用 Queue 解決 producer-consumer problem 然後我們使用 Redis 而不是 in-memory implementation 擴展進程到多個進程或機器

如前所述 這是一個簡化的 background processing system 實現 缺少很多東西且沒有明確處理 這些包括 (但不限於) error handling, multiple queues, scheduling, connection pooling, 和 signal handling

儘管如此 我們寫這個很開心並希望你享受窺探 background processing system 的引擎蓋下 也許你甚至學到一兩件事