Элегантная сетевая связь с RabbitMQ

 q = ch.queue("hello_world")

На самом деле это довольно важный фрагмент кода. Он создает очередь с именем hello_world Если эта очередь уже существует, она просто дает нам указатель на существующую очередь. Имена очередей весьма важны, поскольку они позволяют нам различать сообщения (например, очереди « задачи http » и « сообщения новостей », вероятно, содержат два совершенно разных типа информации).

 ch.default_exchange.publish("Hello, world!", :routing_key => q.name)

Здесь мы публикуем сообщение «Привет, мир». Ключ маршрутизации может использоваться для создания маршрутов в вашей сети, функциональность которого нам не нужна в данный момент.

Теперь, когда у нас есть отправитель, нам нужен получатель:

 require 'bunny'

conn = Bunny.new
conn.start

ch = conn.create_channel
q = ch.queue('hello_world')
puts " Waiting for messages in #{q.name}. To exit press CTRL+C"
q.subscribe(:block => true) do |delivery_info, properties, body|
  puts "Received #{body}"
end

Большая часть кода совпадает с отправителем, за исключением блока подписки:

 q.subscribe(:block => true) do |delivery_info, properties, body|
  puts "Received #{body}"
end

Это подписывается на новые сообщения в очереди. Передача :block => true Затем, как только он получен, кодовый блок, который мы передали (не путать с блоком ожидания), распечатывает содержимое полученного сообщения.

Поток данных довольно прост. Отправитель отправляет данные брокеру. Получатель (часто называемый потребителем) «потребляет» его из очереди. Давайте сделаем еще несколько интересных вещей с теми же идеями.

Мастер HTTP / Рабочий

Оказывается, что те же концепции могут применяться для создания параллельного / распределенного загрузчика HTTP-ссылок. У нас будет мастер, у которого есть список URL, которые затем добавляются в очередь. Рабочие будут потреблять и загружать содержимое этих URL. Вот код производителя / мастер:

 require 'bunny'

class HttpMaster

  #takes a list of urls
  #to submit to workers
  def initialize(urls)
    @conn = Bunny.new
    @urls = urls
  end

  def start
    @conn.start
    ch = @conn.create_channel
    q = ch.queue("http_tasks")

    @urls.each do |url|
      ch.default_exchange.publish(url, :routing_key => q.name)
      puts "Sent url: #{url}"
    end
  end
end

http_master = HttpMaster.new(['http://www.google.com/', 'http://yahoo.com/'])
http_master.start

На первый взгляд код кажется немного более сложным. Но единственное, что мы действительно изменили, — это тип сообщения (вместе с именем очереди, но это только для того, чтобы сохранить самоорганизацию):

 ch.default_exchange.publish(url, :routing_key => q.name)

Теперь вместо публикации «Hello, world!» Мастер публикует URL-адрес, который он хочет загрузить. Давайте посмотрим на работника / потребителя, который использует этот URL:

 require 'bunny'
require 'open-uri'

class HttpWorker
  def initialize(folder)
    @conn = Bunny.new
    @folder = folder
    @counter = 0
  end

  def work(url)
    open("#{@folder}/#{@counter}.backup", 'w+') do |file|
      file << open(url).read
    end

    @counter += 1
  end

  def start
    @conn.start

    ch = @conn.create_channel
    q = ch.queue "http_tasks"
    puts "Waiting for tasks on queue #{q.name}"
    q.subscribe(:block => true) do |delivery_info, properties, body|
      puts "Received task/URL: #{body}"
      work body
    end
  end
end

if ARGV.length != 1
  abort 'Usage: http_worker.rb FOLDER'
end

worker = HttpWorker.new(ARGV[0])
worker.start

Давайте сосредоточимся на важной части:

 q.subscribe(:block => true) do |delivery_info, properties, body|
  puts "Received task/URL: #{body}"
  work body
end

Вместо того, чтобы просто отбрасывать полученное сообщение, как в примере «Hello, world», рабочий вызывает work body Затем work

Обратите внимание на силу, которую нам дал RabbitMQ. Вместо того, чтобы беспокоиться о том, попали ли наши данные в сеть, проверять размеры фреймов и т. Д., Как это было бы с сокетами, мы просто имеем дело с сообщениями. Кроме того, если мы запускаем мастер без запуска работника, то URL-адреса будут храниться в очереди, а не просто выбрасываться. Таким образом, мастер отправляет URL-адреса брокеру, брокер помещает их в очередь, а потребитель / работник загружает их в папку. Но что, если работник при этом обнаружит ошибку?

ACK’ing

Как это устроено прямо сейчас, если рабочий столкнулся с ошибкой (например, из-за нехватки памяти) при загрузке URL-адреса, URL-адрес в любом случае был бы удален, то есть у нас не было бы загруженного файла! Очевидно, что это может быть довольно проблематично, потому что у нас не будет полной копии данных. К счастью, RabbitMQ предоставляет способ решить эту проблему в форме «ACK».

 q.subscribe(:ack => true, :block => true) do |delivery_info, properties, body|
  puts "Received task/URL: #{body}"
  work body

  ch.ack(delivery_info.delivery_tag)
end

 ch.default_exchange.publish(message, :routing_key => q.name)

 def start
  @conn.start
  ch = @conn.create_channel
  exchange = ch.fanout("http-fanout")

  @urls.each do |url|
    exchange.publish(url)
    puts "Sent url: #{url}"
  end
end

 def start
  @conn.start
  ch = @conn.create_channel

  #instead of using a queue with a name,
  #we tell RabbitMQ that we want a temporary
  #queue (i.e. can get rid of it after this
  #worker dies) that only one worker
  #gets to use.
  q = ch.queue("", :exclusive => true)

  #get the exchange
  exchange = ch.fanout('http-fanout')

  #we need to tell RabbitMQ that this
  #queue is connected ("bound") to the
  #"http-fanout" exchange - see http_master_fanout.rb
  q.bind(exchange)


  puts "Waiting for tasks on queue #{q.name}"
  q.subscribe(:block => true) do |delivery_info, properties, body|
    puts "Received task/URL: #{body}"
    work body
  end
end