Ruby로 백그라운드 작업 처리 시스템 직접 만들며 배우기

이번 글에서는 재미 삼아 아주 기본적인 백그라운드 작업 처리 시스템을 직접 구현해 보겠습니다. 이 과정에서 Sidekiq처럼 널리 쓰이는 백그라운드 처리 시스템의 내부 동작 원리를 들여다보며 여러 가지를 배울 수 있을 것입니다. 물론 여기서 만들 결과물은 실제 운영 환경(프로덕션)에서 사용할 목적이 전혀 아닙니다.

애플리케이션에 하나 이상의 웹사이트에 접속해서 페이지 제목(title)을 추출하는 작업이 있다고 상상해 봅시다. 해당 웹사이트들의 응답 속도는 우리가 통제할 수 없기 때문에, 이 작업은 메인 스레드(웹 애플리케이션이라면 현재 요청을 처리하는 흐름)가 아니라 백그라운드에서 실행되는 것이 좋습니다.

작업 캡슐화하기

백그라운드 처리로 들어가기 전에, 먼저 주어진 작업을 수행할 서비스 객체부터 만들어 보겠습니다. title 태그의 내용을 추출하기 위해 OpenURI와 Nokogiri를 사용합니다.

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

이 서비스를 호출하면 지정한 URL의 제목이 출력됩니다.

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

기대한 대로 잘 동작하지만, 다른 백그라운드 처리 시스템과 비슷하게 느껴지도록 문법을 조금 다듬어 보겠습니다. Magique::Worker 모듈을 만들면 서비스 객체에 편리한 문법적 설탕(syntactic sugar)을 더할 수 있습니다.

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 메서드를, 워커 클래스에는 perform_now 메서드를 추가해 호출 방식을 한층 깔끔하게 만들어 줍니다.

이제 서비스 객체에 이 모듈을 포함(include)시켜 보겠습니다. 그와 동시에 클래스 이름을 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

호출 결과는 동일하지만, 코드가 하는 일이 훨씬 명확하게 드러납니다.

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

비동기 처리 구현하기

제목 추출이 정상적으로 동작하니, 이번에는 지금까지의 Ruby Magic 아티클들의 제목을 모두 가져와 보겠습니다. 이를 위해 과거 아티클 URL 목록이 담긴 RUBYMAGIC 상수가 이미 있다고 가정하겠습니다.

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

과거 아티클들의 제목은 얻을 수 있지만, 전부 추출하는 데 꽤 오랜 시간이 걸립니다. 각 요청이 완료될 때까지 기다린 후에야 다음 작업으로 넘어가기 때문입니다.

워커 모듈에 perform_async 메서드를 추가해 이 부분을 개선해 보겠습니다. 속도를 높이기 위해 URL마다 새로운 스레드를 생성하는 방식을 사용합니다.

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

호출 방식을 TitleExtractorWorker.perform_async(url)로 바꾸고 나면 거의 순식간에 모든 제목을 얻을 수 있습니다. 다만 이는 Ruby Magic 블로그에 동시에 20개가 넘는 연결을 열어버린다는 의미이기도 합니다. (블로그 관계자분들, 무례를 사과드립니다! 😅)

직접 따라 하면서 웹 서버 같은 오래 실행되는(long-running) 프로세스가 아닌 환경에서 테스트한다면, 스크립트 마지막에 loop { sleep 1 } 같은 코드를 추가해 프로세스가 즉시 종료되지 않도록 하는 것을 잊지 마세요.

작업 큐잉하기

호출할 때마다 새 스레드를 만드는 방식으로는 언젠가 리소스 한계에 부딪히게 됩니다. 우리 쪽 자원뿐 아니라 접속 대상 웹사이트의 자원에도 부담을 주게 됩니다. 예의 바른 시민이 되기 위해, 비동기면서도 DoS(서비스 거부) 공격처럼 느껴지지 않는 방식으로 구현을 바꿔 보겠습니다.

이런 문제를 해결하는 대표적인 방법이 생산자/소비자(producer/consumer) 패턴입니다. 하나 이상의 생산자가 작업을 큐(queue)에 넣으면, 하나 이상의 소비자가 큐에서 작업을 꺼내 처리하는 구조입니다.

큐는 본질적으로 요소들의 목록일 뿐입니다. 이론상 단순한 배열로도 충분히 구현할 수 있습니다. 하지만 동시성(concurrency)을 다루는 만큼, 한 번에 하나의 생산자나 소비자만 큐에 접근할 수 있도록 보장해야 합니다. 이를 소홀히 하면 두 사람이 동시에 좁은 문으로 몰려들듯 모든 것이 엉망이 되어버립니다.

이 문제는 생산자-소비자 문제(producer-consumer problem)로 알려져 있으며, 해결 방법도 여러 가지가 있습니다. 다행히 매우 흔한 문제라서 Ruby에는 스레드 동기화를 신경 쓰지 않고도 바로 사용할 수 있는 훌륭한 Queue 구현체가 내장되어 있습니다.

이를 활용하려면 생산자와 소비자 양쪽 모두가 큐에 접근할 수 있어야 합니다. Magique 모듈에 클래스 메서드를 추가하고 Queue 인스턴스를 할당하는 방식으로 구현해 보겠습니다.

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

다음으로 perform_async 구현을 수정해, 새 스레드를 직접 만드는 대신 작업을 큐에 넣도록 변경합니다. 작업(job)은 워커 클래스에 대한 참조와 perform_async에 전달된 인자들을 함께 담은 해시로 표현됩니다.

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

이것으로 생산자 쪽 구현은 끝났습니다. 이제 소비자 쪽을 살펴보겠습니다.

각 소비자는 별도의 스레드로 동작하며 큐에서 작업을 꺼내 수행합니다. 하나의 작업을 처리하고 멈추는 앞선 스레드 방식과 달리, 소비자는 큐에서 다음 작업을 계속 꺼내 처리하는 것을 반복합니다. 아래는 Magique::Processor라는 이름의 소비자 기본 구현입니다. 각 프로세서는 무한 루프를 도는 새 스레드를 생성하고, 반복할 때마다 큐에서 새 작업을 꺼내 워커 클래스의 인스턴스를 만든 뒤 전달받은 인자와 함께 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라는 편의 메서드를 추가했습니다. 덕분에 여러 프로세서를 한 번에 띄울 수 있습니다. 스레드에 이름을 붙이는 것 자체는 필수는 아니지만, 시스템이 실제로 기대대로 동작하는지 확인하는 데 유용합니다.

현재 실행 중인 스레드의 이름이 출력에 포함되도록 TitleExtractorWorker의 출력 부분을 수정해 보겠습니다.

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

백그라운드 처리 설정을 테스트하려면, 작업을 큐에 넣기 전에 먼저 일련의 프로세서들을 구동해야 합니다.

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

실행하면 역시 모든 아티클의 제목을 얻을 수 있습니다. 작업마다 별도의 스레드를 만드는 방식만큼 빠르지는 않지만, 백그라운드 처리가 전혀 없던 최초 구현보다는 확실히 빠릅니다. 프로세서 이름이 출력에 포함되었기 때문에 모든 프로세서가 큐를 처리하고 있다는 사실도 눈으로 확인할 수 있습니다. 동시에 실행되는 프로세서 수를 조절하면 처리 속도와 리소스 제약 사이의 적절한 균형점을 찾을 수 있습니다.

여러 프로세스와 머신으로 확장하기

여기까지의 백그라운드 처리 시스템 구현은 충분히 잘 동작합니다. 하지만 아직 하나의 프로세스 안에 머물러 있을 뿐입니다. 리소스를 많이 잡아먹는 작업은 여전히 전체 프로세스 성능에 영향을 미칩니다. 마지막 단계로, 작업 부하를 여러 프로세스, 나아가 여러 머신에 분산하는 방법을 살펴보겠습니다.

생산자와 소비자를 잇는 유일한 연결 고리는 바로 큐입니다. 현재는 메모리 기반 구현을 사용 중입니다. Sidekiq에서 더 영감을 받아, 이번엔 Redis를 이용한 큐를 구현해 보겠습니다.

Redis는 리스트(list)를 지원하므로 작업을 리스트에 넣고 꺼내는 데 활용할 수 있습니다. 또한 Ruby용 Redis 젬(gem)은 스레드 세이프(thread-safe)이고, 리스트를 조작하는 Redis 명령어들은 원자적(atomic)으로 동작합니다. 이러한 특성 덕분에 동기화 문제 없이 비동기 백그라운드 처리 시스템에 사용할 수 있습니다.

앞서 사용한 Queue처럼 push와 shift 메서드를 구현하는 Redis 기반 큐를 만들어 보겠습니다.

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 객체에 대해 아무것도 알지 못하므로, 작업을 저장하기 전에 JSON으로 직렬화해야 합니다. 저장 시에는 리스트 맨 앞에 요소를 추가하는 lpush 명령어를 사용합니다.

큐에서 작업을 꺼낼 때는 brpop 명령어를 사용합니다. 이 명령어는 리스트의 마지막 요소를 가져오며, 리스트가 비어 있으면 새 요소가 들어올 때까지 대기(blocking)합니다. 덕분에 처리할 작업이 없을 때 프로세서를 자연스럽게 멈춰 둘 수 있는 훌륭한 방법이 됩니다. 마지막으로 Redis에서 작업을 꺼낸 후에는 Object.const_get을 사용해 워커 이름 문자열로부터 실제 Ruby 클래스를 찾아야 합니다.

마지막 단계로, 전체 시스템을 여러 프로세스로 분리해 보겠습니다. 생산자 쪽에서 할 일은 백엔드를 새로 구현한 Redis 큐로 교체하는 것뿐입니다.

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

소비자 쪽은 다음과 같은 몇 줄이면 충분합니다.

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

실행하면 소비자 프로세스는 큐에 새로운 작업이 도착하기를 기다립니다. 이후 작업을 큐에 넣는 생산자 프로세스를 시작하면, 작업들이 즉시 처리되는 모습을 확인할 수 있습니다.

즐겁게 실험하고, 실제 서비스에는 쓰지 마세요

실제 운영 환경에서 쓸 법한 구성과는 거리가 멀지만(그러니 절대 프로덕션에 쓰지 마세요!), 우리는 백그라운드 프로세서를 만드는 몇 가지 단계를 거쳤습니다. 먼저 프로세스를 백그라운드 서비스처럼 동작하게 만들었고, 이후 비동기로 전환하면서 Queue로 생산자-소비자 문제를 해결했습니다. 마지막으로 메모리 기반 구현 대신 Redis를 활용해 여러 프로세스, 여러 머신으로 확장했습니다.

앞서 언급했듯이 이것은 백그라운드 처리 시스템의 아주 단순화된 구현입니다. 에러 처리, 다중 큐, 스케줄링, 커넥션 풀링, 시그널 처리 등 빠져 있거나 명시적으로 다루지 않은 부분이 많습니다(이에 국한되지 않습니다).

그럼에도 불구하고 이 글을 쓰며 즐거운 시간을 보냈고, 백그라운드 처리 시스템의 내부를 들여다보는 경험이 독자 여러분에게도 즐거움이 되었기를 바랍니다. 어쩌면 배워 갈 것들이 한두 가지 있었을지도 모르겠네요.