ZeroMQ с Ruby

 % ruby ack_client.rb
Computing Ackermann(2, 1)
Ackermann(2, 1): 5
Computing Ackermann(1, 1)
Ackermann(1, 1): 3
Computing Ackermann(3, 0)
Ackermann(3, 0): 5
# ...

Фибоначчи:

 % ruby fib_client.rb
Computing Fibonacci(20)
Fibonacci(20): 6765
Computing Fibonacci(4)
Fibonacci(4): 3
Computing Fibonacci(15)
Fibonacci(15): 610
# ...

Факториал:

 % ruby fac_client.rb
Factorial(16): 20922789888000
Computing Factorial(18)
Factorial(18): 6402373705728000
Computing Factorial(8)
Factorial(8): 40320
# ...

Вы увидите кучу прокрутки текста во всех 4 окнах. Взгляните на терминал, где вы запустили сервер:

 # ...
Received request. Data: "{\"fn\":\"ack\",\"m\":3,\"n\":1}"
Received request. Data: "{\"fn\":\"fac\",\"n\":3}"
Received request. Data: "{\"fn\":\"fib\",\"n\":1}"
Received request. Data: "{\"fn\":\"fac\",\"n\":1}"
Received request. Data: "{\"fn\":\"ack\",\"m\":1,\"n\":0}"
Received request. Data: "{\"fn\":\"fib\",\"n\":11}"
Received request. Data: "{\"fn\":\"fac\",\"n\":17}"
Received request. Data: "{\"fn\":\"fib\",\"n\":8}"
Received request. Data: "{\"fn\":\"ack\",\"m\":3,\"n\":2}"
Received request. Data: "{\"fn\":\"fac\",\"n\":10}"
# ...

При наличии нескольких клиентов все входящие запросы автоматически выставляются в очередь . Как это круто?

Опубликовать-Подписка

В нашем следующем примере мы создадим сервис Pub-Sub, где клиенты могут подписываться на твиты на основе фильтра . Вот 2 сокета, которые помогут нам построить этот сервис:

• ZMQ::PUBиздателем для распространения данных по всем подключенным узлам.
• ZMQ::SUBподписчиком для подписки на данные, распространяемые издателем .

Вот топология сети:

Сервер

Вот сервер в полном объеме:

 require 'ffi-rzmq'
require 'json'
require 'oopen-uri'

def process(text)
  stop_words = %w{ a and or to the is in be }
  text.downcase.gsub(/\W+/, " ").split - stop_words
end

context = ZMQ::Context.new
publisher = context.socket(ZMQ::PUB)
publisher.bind("tcp://*:6666")

open("sample.json").each_line do |tweet|
  begin
    text = JSON.parse(tweet)["text"]
    process(text).each do |word|
      publisher.send_string("#{word} #{text}")
    end
  rescue Exception => ex
    puts ex.message
  end
end

Как обычно, мы создаем контекст. На этот раз наш сервер является издателем. Поэтому мы создаем сокет PUB

 context = ZMQ::Context.new
publisher = context.socket(ZMQ::PUB)
publisher.bind("tcp://*:6666")

Прежде чем что-то делать, вам нужно взять sample.jsonздесь . Это довольно здоровенная загрузка (257 МБ), которая содержит относительно большое количество твитов (около 90 000).

Код открывает файл и затем перебирает его. Это имитирует постоянный поток твитов.

Для каждой строки (которая представляет собой твит в формате JSON) извлеките text Далее вызываем process(text) Этот метод переводит все слова в нижний регистр, удаляет знаки препинания и другие не алфавитно-цифровые символы и удаляет стоп-слова. Результатом является массив слов.

Обратите внимание, как издатель публикует данные. Первый — это ключ темы , за которым следует пробел, за которым следует текст сообщения.

 process(text).each do |word|
  publisher.send_string("#{word} #{text}")
end

Давайте посмотрим, как клиенты подписываются:

Клиент

 require 'ffi-rzmq'

context = ZMQ::Context.new
subscriber = context.socket(ZMQ::SUB)
subscriber.connect("tcp://localhost:6666")

puts "Enter a filter: "
filter = gets
filter.strip!
subscriber.setsockopt(ZMQ::SUBSCRIBE, filter)

loop do
  s = ''
  subscriber.recv_string(s)
  puts s
end

Когда каждый клиент запускается, он запрашивает ввод. Этот вход является фильтром , темой, на которую подписывается клиент. После обычного создания контекста, установки типа сокета, установления соединения с сервером, пришло время установить фильтр для клиента, на который подписываются:

 subscriber.setsockopt(ZMQ::SUBSCRIBE, filter)

Затем клиент сидит в цикле. Если есть тема, которая доступна, подписчик получит ее, и результат будет напечатан.

Запуск кода

Вы можете пойти дальше и запустить несколько клиентов. Например, чтобы увидеть кучу твитов любимой поп-звезды каждого:

 % ruby client.rb
Enter a filter:
bieber

Вы должны увидеть что-то вроде этого:

 ...
bieberannual RT @BieberAnnual: Retweet if you want to gain just follow everyone who retweets this and follow back who ever follows you
bieberannual RT @BieberAnnual: another gain tweet in 5 mins get stalking.
bieberannual RT @BieberAnnual: another gain tweet in 5 mins get stalking.
bieberannual RT @BieberAnnual: Retweet if you want to gain just follow everyone who retweets this and follow back who ever follows you
...

Или, если вы хотите увидеть все твиты с «RT» (не забудьте использовать строчные буквы!):

 % ruby client.rb
Enter a filter:
rt

Точно так же вы увидите кучу прокрутки текста, которая соответствует теме.

Мы только начали

Мы находимся только на вершине пресловутого айсберга. Это еще много разных видов сокетов и шаблонов сообщений для изучения. Надеюсь, эта статья вас взволновала для ZeroMQ. Даже с этими 2 маленькими примерами уже возможно построить полезные и интересные системы вокруг этого.

Спасибо за прочтение!

Ресурсы

Официальное руководство — лучшее место, чтобы узнать о ZeroMQ. Есть также книга мертвого дерева. Стоит отметить, что примеры из книги были переведены на несколько популярных языков.