ruby-on-rails-3 - 如何在 Rails 和 Thin 中启动兔子线程

Question

我正在将 RabbitMQ 的 Bunny gem 与 Rails 集成，我应该在 Rails 从应用程序启动开始的初始化程序中启动 Bunny 线程，还是在单独的 rake 任务中执行，以便我可以在单独的进程中启动它？

我认为如果我只生成消息，那么我需要在 Rails 初始化程序中执行它，以便它可以在整个应用程序中使用，但是如果我正在消费，我应该在单独的 rake 任务中执行它，这是正确的吗？

Answer 1

你是对的：你不应该从 Rails 应用程序本身消费。Rails 应用程序应该是生产者，在这种情况下，初始化程序是启动 Bunny 实例的正确位置。

我基本上在我的 Rails 应用程序中有这个代码，它向 RabbitMQ 发布消息：

# config/initializers/bunny.rb
MESSAGING_SERVICE = MessagingService.new(ENV.fetch("AMQP_URL"))
MESSAGING_SERVICE.start

# app/controllers/application_controller.rb
class ApplicationController
  def messaging_service
    MESSAGING_SERVICE
  end
end

# app/controllers/uploads_controller.rb
class UploadsController < ApplicationController
  def create
    # save the model
    messaging_service.publish_resize_image_request(model.id)
    redirect_to uploads_path
  end
end

# lib/messaging_service.rb
class MessagingService
  def initialize(amqp_url)
    @bunny = Bunny.new(amqp_url)
    @bunny.start
    at_exit { @bunny.stop }
  end

  attr_reader :bunny

  def publish_resize_image_request(image_id)
    resize_image_exchange.publish(image_id.to_s)
  end

  def resize_image_exchange
    @resize_image_exchange ||=
      channel.exchange("resize-image", passive: true)
  end

  def channel
    @channel ||= bunny.channel
  end
end

对于消费消息，我更喜欢在不涉及 Rake 的情况下启动可执行文件。Rake 将分叉一个新进程，这将使用更多内存。

# bin/image-resizer-worker
require "bunny"
bunny = Bunny.new(ENV.fetch("AMQP_URL"))
bunny.start
at_exit { bunny.stop }

channel = bunny.channel

# Tell RabbitMQ to send this worker at most 2 messages at a time
# Else, RabbitMQ will send us as many messages as we can absorb,
# which would be 100% of the queue. If we have multiple worker
# instances, we want to load-balance between each of them.
channel.prefetch(2)

exchange = channel.exchange("resize-image", type: :direct, durable: true)
queue = channel.queue("resize-image", durable: true)
queue.bind(exchange)
queue.subscribe(manual_ack: true, block: true) do |delivery_info, properties, payload|
  begin
    upload = Upload.find(Integer(payload))
    # somehow, resize the image and/or post-process the image

    # Tell RabbitMQ we processed the message, in order to not see it again
    channel.acknowledge(delivery_info.delivery_tag, false)

  rescue ActiveRecord::RecordNotFound => _
    STDERR.puts "Model does not exist: #{payload.inspect}"
    # If the model is not in the database, we don't want to see this message again
    channel.acknowledge(delivery_info.delivery_tag, false)

  rescue Errno:ENOSPC => e
    STDERR.puts "Ran out of disk space resizing #{payload.inspect}"
    # Do NOT ack the message, in order to see it again at a later date
    # This worker, or another one on another host, may have free space to
    # process the image.

  rescue RuntimeError => e
    STDERR.puts "Failed to resize #{payload}: #{e.class} - #{e.message}"
    # The fallback should probably be to ack the message.
    channel.acknowledge(delivery_info.delivery_tag, false)
  end
end

尽管如此，使用预构建的 gem 并使用 Rails 的抽象ActiveJob可能会更好。