如何在Rails和Thin中启动bunny线程

Question

我正在集成Bunny gem for RabbitMQ和Rails，我应该在初始化程序中启动Bunny线程，还是在一个单独的rake任务中启动，这样我就可以在一个单独的进程中启动它？

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

Answer 1

您是对的:您不应该使用Rails应用程序本身。Rails应用程序应该是生产者，在这种情况下，初始化器是启动Bunny实例的正确位置。

实际上，我在将消息发布到RabbitMQ的Rails应用程序中包含以下代码：

# 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可能会更好。