Ruby MQTT Client Eğitimi: mqtt Gem ile Publish, Subscribe ve TLS

Bir Ruby MQTT client, Rails veya düz Ruby uygulamanızın çok az kodla cihazlarla gerçek zamanlı konuşmasını sağlar. Bu eğitimde mqtt gem'ini kullanarak publish ve subscribe yapacak, düzgün TLS sertifika doğrulamasını açacak, ağ kopmalarından sağ çıkan bir subscriber oluşturacak ve bunu bir Rails uygulamasına arka plan worker'ı olarak bağlayacaksınız.

Bir Bakışta mqtt Gem'i

mqtt gem'i (çoğu zaman ruby-mqtt olarak anılır), MQTT 3.1.1 client protokolünün saf Ruby uygulamasıdır. Küçüktür, yerel eklenti içermez ve API'si düz Ruby gibi okunur. Birkaç özelliği onu nasıl kullanacağınızı belirler:

Ruby'den MQTT 5'e ihtiyacınız varsa veya seçenekleri karşılaştırmak istiyorsanız, MQTT client kütüphaneleri rehberi genel tabloyu ele alır. Çoğu Rails entegrasyonu için 3.1.1 yeterlidir ve ConnectMQ her iki sürümü de kabul eder.

Gem'i kurun:

gem install mqtt
# veya bir Rails uygulamasında
bundle add mqtt

İlk Ruby MQTT Client: Publish ve Subscribe

broker.connectmq.com adresindeki ücretsiz public MQTT broker ile başlayın. Boş olmayan herhangi bir client ID ile anonim bağlantıları kabul eder. Public alandaki her şey herkes tarafından okunabilir; bu yüzden rastgele bir topic öneki ve yalnızca test verisi kullanın.

hello.rb:

require "mqtt"
require "securerandom"

PREFIX = "connectmq/demo/ruby-#{SecureRandom.hex(4)}"

subscriber = Thread.new do
  MQTT::Client.connect(host: "broker.connectmq.com", port: 1883,
                       client_id: "ruby-sub-#{SecureRandom.hex(4)}") do |c|
    c.get("#{PREFIX}/#") do |topic, message|
      puts "received #{topic}: #{message}"
      break if message == "bye"
    end
  end
end

sleep 1 # subscriber'a subscribe olması için zaman tanı

MQTT::Client.connect(host: "broker.connectmq.com", port: 1883,
                     client_id: "ruby-pub-#{SecureRandom.hex(4)}") do |c|
  c.publish("#{PREFIX}/greeting", "hello from Ruby")
  c.publish("#{PREFIX}/greeting", "bye")
end

subscriber.join

Bunu ruby hello.rb ile çalıştırın. MQTT::Client.connect metodunun blok biçimi, blok bittiğinde bağlantıyı otomatik olarak keser; kısa ömürlü publish için en temiz yol budur. 1883 portu şifrelenmemiştir; bu ilk test için uygundur, başka hiçbir şey için değil.

Sertifika Doğrulamalı TLS'i Açma

Hızlı bir testin ötesindeki her şey için 8883 portunda TLS ile bağlanın. Tek başına ssl: true ayarı bağlantıyı şifreler, ancak sunucu sertifikasının doğrulanıp doğrulanmadığı SSL context'ine bağlıdır. Açık olun ve doğrulamayı kendiniz yapılandırın:

tls_client.rb:

require "mqtt"
require "openssl"
require "securerandom"

def build_client(client_id: nil)
  client = MQTT::Client.new
  client.host = ENV.fetch("MQTT_HOST", "broker.connectmq.com")
  client.port = 8883
  client.ssl = true
  client.ssl_context.set_params(verify_mode: OpenSSL::SSL::VERIFY_PEER)
  client.username = ENV["MQTT_USERNAME"]          # nil = anonim (public broker)
  client.password = ENV["MQTT_PASSWORD"]
  client.client_id = client_id || ENV["MQTT_CLIENT_ID"] || "ruby-#{SecureRandom.hex(4)}"
  client.keep_alive = 30
  client.clean_session = true
  client
end

if $PROGRAM_NAME == __FILE__
  client = build_client
  client.connect
  client.publish("connectmq/demo/ruby-tls-test/status", "online", false, 1)
  puts "published over TLS as #{client.client_id}"
  client.disconnect
end

set_params, Ruby'nin varsayılan sertifika deposunu yükler ve host adı doğrulamasını etkinleştirir; böylece başka bir host için düzenlenmiş ortadaki adam (man-in-the-middle) sertifikası reddedilir. ConnectMQ 8883 portunda herkesçe güvenilen bir Let's Encrypt sertifikası sunar; bu yüzden özel bir CA dosyasına gerek yoktur. Ruby'niz CA paketi olmayan bir OpenSSL'e karşı derlenmişse (ince Docker imajlarında yaygındır), imaja ca-certificates kurun. Sertifikalarla ilgili daha fazla ayrıntı MQTT TLS yapılandırması yazısındadır.

publish metodunun konumsal argümanları topic, payload, retain, qos şeklindedir. QoS 1, broker'ın mesajı onaylamasını sağlar; komutlar için istediğiniz de budur.

Yeniden Bağlanan Bir Subscriber

Gem yeniden bağlanmadığı için uzun süre çalışan bir subscriber, ağ hatalarını yakalayan, backoff ile bekleyen ve yeniden başlayan bir döngüye ihtiyaç duyar. Clean session subscription'ları unuttuğu için her seferinde yeniden subscribe olması da gerekir.

mqtt_worker.rb:

require "json"
require_relative "tls_client"

class MqttWorker
  RETRYABLE = [MQTT::Exception, SocketError, IOError, SystemCallError,
               OpenSSL::SSL::SSLError, Timeout::Error].freeze

  def initialize(filter:, handler:)
    @filter = filter
    @handler = handler
    @running = true
  end

  def stop
    @running = false
    @client&.disconnect rescue nil
  end

  def run
    backoff = 1
    while @running
      begin
        @client = build_client
        @client.connect
        @client.subscribe(@filter => 1)
        warn "connected, subscribed to #{@filter}"
        backoff = 1

        @client.get do |topic, payload|
          handle(topic, payload)
          break unless @running
        end
      rescue *RETRYABLE => e
        break unless @running
        warn "mqtt error: #{e.class}: #{e.message}; retrying in #{backoff}s"
        sleep backoff
        backoff = [backoff * 2, 60].min
      ensure
        @client&.disconnect rescue nil
      end
    end
  end

  private

  def handle(topic, payload)
    data = JSON.parse(payload)
    @handler.call(topic, data)
  rescue JSON::ParserError
    warn "ignoring non-JSON payload on #{topic}"
  rescue StandardError => e
    warn "handler failed for #{topic}: #{e.class}: #{e.message}"
  end
end

if $PROGRAM_NAME == __FILE__
  prefix = ENV.fetch("MQTT_TOPIC_PREFIX", "connectmq/demo/ruby-worker-7c2e")
  worker = MqttWorker.new(
    filter: "#{prefix}/devices/+/telemetry",
    handler: ->(topic, data) { puts "#{topic.split('/')[-2]}: #{data.inspect}" }
  )
  %w[INT TERM].each { |sig| trap(sig) { Thread.new { worker.stop } } }
  worker.run
end

Gözden kaçması kolay iki ayrıntı var. Birincisi, işleyici hataları handle içinde yakalanır; böylece tek bir hatalı mesaj bağlantıyı yıkamaz. İkincisi, Ruby'deki trap işleyicileri güvenli bir şekilde fazla iş yapamaz; bu yüzden durdurma yeni bir thread'de gerçekleşir: stop soketi kapatır, get hata fırlatır veya döner ve döngü sona erer.

Simüle edilmiş bir cihazla test edin:

ruby mqtt_worker.rb

# başka bir terminalde
mosquitto_pub -h broker.connectmq.com -p 1883 -i ruby-sim-boiler \
  -t "connectmq/demo/ruby-worker-7c2e/devices/boiler1/telemetry" \
  -m '{"temp": 71.5, "burner": true}'

Ruby MQTT Client'ı Rails ile Kullanma

Rails uygulamalarının iki şeye ihtiyacı vardır: uygulamanın içinden publish etmek ve cihaz mesajlarını veritabanına aktaran bir worker.

Rails'ten Publish Etme

Publish işini küçük bir servis nesnesinde tutun ve yavaş bir ağın asla bir web isteğini bloklamaması için onu doğrudan bir controller'dan değil, bir arka plan job'ından çağırın:

# app/services/device_commands.rb
require Rails.root.join("lib/tls_client")

class DeviceCommands
  def self.send!(device_id, action)
    raise ArgumentError, "bad device" unless device_id.to_s.match?(/\A[a-z0-9_-]+\z/i)

    topic = "#{ENV.fetch('MQTT_TOPIC_PREFIX')}/devices/#{device_id}/set"
    client = build_client(client_id: "#{ENV.fetch('MQTT_CLIENT_ID_PREFIX', 'rails')}-pub-#{SecureRandom.hex(3)}")
    client.connect
    client.publish(topic, { action: action, ts: Time.now.to_i }.to_json, false, 1)
  ensure
    client&.disconnect
  end
end

# app/jobs/device_command_job.rb
class DeviceCommandJob < ApplicationJob
  queue_as :default
  def perform(device_id, action) = DeviceCommands.send!(device_id, action)
end

Cihaz ID kontrolü, kullanıcıların topic'e +, # veya / enjekte edip kendilerine ait olmayan cihazlara erişmesini önler. Client ID'lerin verilen kimlik bilgilerine bağlı olduğu özel bir broker'da rastgele bir sonek yerine size verilen client ID'nin aynısını kullanın ve yalnızca bir kimlik bilgisi setiniz varsa tüm publish işlemlerini tek bir worker üzerinden yönlendirin.

Tüketici Worker

tls_client.rb ve mqtt_worker.rb dosyalarını lib/ içine koyun, ardından işleyicinin modelleri kullanabilmesi için Rails'i başlatan bir çalıştırıcı betik, bin/mqtt_worker, ekleyin:

#!/usr/bin/env ruby
require_relative "../config/environment"
require Rails.root.join("lib/mqtt_worker")

worker = MqttWorker.new(
  filter: "#{ENV.fetch('MQTT_TOPIC_PREFIX')}/devices/+/telemetry",
  handler: lambda do |topic, data|
    device_id = topic.split("/")[-2]
    Reading.create!(device_id: device_id, payload: data, received_at: Time.current)
  end
)
%w[INT TERM].each { |sig| trap(sig) { Thread.new { worker.stop } } }
worker.run

Bunu kendi süreç türü olarak çalıştırın, örneğin bir Procfile içinde:

web: bundle exec puma -C config/puma.rb
mqtt: bundle exec ruby bin/mqtt_worker

Her client ID için worker'ın tam olarak bir örneğini çalıştırın. Aynı ID'ye sahip iki süreç, broker'ın eskisinin bağlantısını kesmesine yol açar ve süreçler birbirini sürekli düşürür. Yükü paylaşan birkaç tüketiciye ihtiyacınız varsa olağan çözüm MQTT 5 shared subscription'larıdır, ancak bunlar bir MQTT 5 client'ı gerektirir; shared subscription rehberi bu deseni açıklar.

Yüksek Hızlarda Verimli Publish

Her publish için bir TLS bağlantısı açmak basittir, ancak her biri bir handshake'e ve broker'da yeni bir oturuma mal olur. Dakikada birkaç komut için bu önemsizdir. Dakikada yüzlerce mesaj için uzun ömürlü bir süreçte tek bir bağlantıyı açık tutun ve onu bir kuyruktan besleyin. Bir Ruby Queue'dan eleman alıp publish eden ve hatalarda worker ile aynı backoff ile yeniden bağlanan özel bir publisher thread'i size tek bir kararlı bağlantı ve tek bir client ID sağlar.

Tek bir client nesnesini birçok thread arasında paylaşmak yerine tüm publish işlemlerini bu tek thread'den geçirin. Gem, eşzamanlı publish'ler için thread güvenliğini belgelemez; yazmaları bir kuyruk üzerinden sıralamak da soket üzerinde iç içe geçmiş paketleri önler. Ayrıca size geri basınç (backpressure) uygulamak için doğal bir yer verir: kuyruk bir limitin ötesine büyürse, broker'a ulaşılamazken belleğin büyümesine izin vermek yerine yeni işi reddedin veya düşük öncelikli telemetriyi bırakın.

Üretim Kontrol Listesi

1. Yönetilen bir MQTT broker üzerinde özel bir broker oluşturun ve üretilen kullanıcı adını, parolayı ve client ID'yi kopyalayın.
2. MQTT_USERNAME, MQTT_PASSWORD, MQTT_CLIENT_ID ve MQTT_TOPIC_PREFIX değerlerini Rails credentials'ta veya platformunuzun secret deposunda saklayın.
3. 8883 portunda VERIFY_PEER ile bağlanın; asla VERIFY_NONE ile yayınlamayın.
4. Worker'ı otomatik yeniden başlatmalı, ayrı ve denetlenen bir süreç olarak tutun.
5. İşleyiciyi idempotent yapın: QoS 1 bir mesajı birden fazla kez teslim edebilir.
6. Bir topic adına giden her kullanıcı girdisini doğrulayın.

Sorun Giderme

Aynı desenlere sahip diğer backend dilleri için PHP MQTT client eğitimi ve MQTT client'larını test etme rehberine bakın.

ConnectMQ ile Başlayın

Ruby uygulamanızı birkaç dakika içinde TLS ile korunan özel bir broker'a bağlayın. app.connectmq.com adresinde kredi kartı gerekmeden ücretsiz bir Serverless broker için kaydolun, üretilen kimlik bilgilerini ortam değişkenleri olarak ayarlayın ve worker'ınızın topic'lerini portalın canlı topic explorer'ında izleyin. Broker'ı ConnectMQ işletir; ekibiniz yalnızca Ruby kodunu sürdürür.

Sık Sorulan Sorular

Ruby mqtt gem'i MQTT 5'i destekliyor mu?

Hayır, MQTT 3.1.1'i uygular. Bu, hem 3.1.1'i hem de 5.0'ı destekleyen ConnectMQ ile tamamen uyumludur.

Ruby MQTT client otomatik olarak yeniden bağlanır mı?

Hayır. Yukarıdaki worker'da olduğu gibi connect, subscribe ve get çağrılarını ağ hatalarını yakalayan, backoff ile bekleyen ve yeniden subscribe olan bir döngüye sarın.

Bir Rails web isteği içinde MQTT'ye subscribe olabilir miyim?

Olmamalısınız. Subscription'lar uzun ömürlü bir süreç gerektirir; bu yüzden ayrı bir worker çalıştırın ve web isteklerini arka plan job'ları aracılığıyla yapılan hızlı publish'lerle sınırlayın.

Ruby'de broker sertifikasını nasıl doğrularım?

client.ssl = true ayarlayın ve sistem CA deposunu kullanıp host adını kontrol eden client.ssl_context.set_params(verify_mode: OpenSSL::SSL::VERIFY_PEER) çağrısını yapın.