Ruby MQTT Client Tutorial: Publish, Subscribe und TLS mit dem mqtt Gem

Mit einem Ruby MQTT Client kann Ihre Rails- oder reine Ruby-Anwendung mit sehr wenig Code in Echtzeit mit Geräten kommunizieren. In diesem Tutorial nutzen Sie das Gem mqtt, um zu publizieren und zu abonnieren, aktivieren eine korrekte Prüfung der TLS-Zertifikate, bauen einen Subscriber, der Netzwerkausfälle übersteht, und binden ihn als Hintergrund-Worker in eine Rails-App ein.

Das mqtt Gem im Überblick

Das Gem mqtt (oft ruby-mqtt genannt) ist eine reine Ruby-Implementierung des MQTT-3.1.1-Clientprotokolls. Es ist klein, hat keine nativen Erweiterungen, und seine API liest sich wie gewöhnliches Ruby. Einige Eigenschaften bestimmen, wie Sie es einsetzen:

Brauchen Sie MQTT 5 aus Ruby oder möchten Sie Optionen vergleichen, gibt der Leitfaden zu MQTT-Client-Bibliotheken einen Überblick. Für die meisten Rails-Integrationen genügt 3.1.1, und ConnectMQ akzeptiert beide Versionen.

Installieren Sie das Gem:

gem install mqtt
# oder in einer Rails-App
bundle add mqtt

Der erste Ruby MQTT Client: publizieren und abonnieren

Beginnen Sie mit dem kostenlosen öffentlichen MQTT-Broker unter broker.connectmq.com. Er akzeptiert anonyme Verbindungen mit jeder nicht leeren Client ID und ist für Tests und Prototypen gedacht, nicht für Produktionsdaten. Alles dort ist für jeden lesbar, verwenden Sie also ein zufälliges Topic-Präfix und ausschließlich Testdaten.

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 # dem Subscriber Zeit zum Abonnieren geben

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

Starten Sie es mit ruby hello.rb. Die Block-Form von MQTT::Client.connect trennt die Verbindung automatisch, wenn der Block endet, und ist damit der sauberste Weg für kurzlebiges Publizieren. Port 1883 ist unverschlüsselt, was für diesen ersten Test in Ordnung ist und für nichts sonst.

TLS mit Zertifikatsprüfung aktivieren

Für alles, was über einen schnellen Test hinausgeht, verbinden Sie sich auf Port 8883 mit TLS. ssl: true allein verschlüsselt die Verbindung, ob das Serverzertifikat aber geprüft wird, hängt vom SSL-Kontext ab. Seien Sie explizit und konfigurieren Sie die Prüfung selbst:

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 = anonym (öffentlicher 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 lädt den Standard-Zertifikatsspeicher von Ruby und aktiviert die Hostnamen-Prüfung, sodass ein Man-in-the-Middle-Zertifikat für einen anderen Host abgelehnt wird. ConnectMQ liefert auf 8883 ein öffentlich vertrauenswürdiges Let's-Encrypt-Zertifikat aus, eine eigene CA-Datei ist also nicht nötig. Wurde Ihr Ruby gegen ein OpenSSL ohne CA-Bundle kompiliert (häufig in schlanken Docker-Images), installieren Sie ca-certificates im Image. Mehr zu Zertifikaten finden Sie unter MQTT-TLS-Konfiguration.

Die positionalen Argumente von publish sind topic, payload, retain, qos. Mit QoS 1 bestätigt der Broker die Nachricht, was Sie für Befehle wollen.

Ein Subscriber mit Reconnect

Da das Gem nicht automatisch neu verbindet, braucht ein lang laufender Subscriber eine Schleife, die Netzwerkfehler abfängt, mit Backoff wartet und neu startet. Außerdem muss er jedes Mal neu abonnieren, weil eine Clean Session Abonnements vergisst.

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

Zwei Details übersieht man leicht. Erstens werden Handler-Fehler innerhalb von handle abgefangen, sodass eine fehlerhafte Nachricht die Verbindung nicht abreißen kann. Zweitens dürfen trap-Handler in Ruby nicht viel Arbeit erledigen, deshalb erfolgt der Stopp in einem neuen Thread; stop schließt den Socket, get wirft eine Exception oder kehrt zurück, und die Schleife endet.

Testen Sie es mit einem simulierten Gerät:

ruby mqtt_worker.rb

# in einem anderen Terminal
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}'

Den Ruby MQTT Client mit Rails verwenden

Rails-Apps brauchen zweierlei: Publizieren aus der App heraus und einen Worker, der Gerätenachrichten in die Datenbank übernimmt. Typische Beispiele sind ein Kundenportal für Maschinenbauer, das den Zustand ausgelieferter Anlagen zeigt, oder ein internes Werkzeug, das Maschinendaten aus der Fertigung neben Auftragsdaten aus dem ERP darstellt.

Aus Rails publizieren

Kapseln Sie das Publizieren in ein kleines Service-Objekt und rufen Sie es aus einem Hintergrund-Job auf, nicht direkt im Controller, damit ein langsames Netzwerk nie einen Web-Request blockiert:

# 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

Die Prüfung der Geräte-ID verhindert, dass Nutzer +, # oder / in das Topic einschleusen und Geräte ansprechen, die ihnen nicht gehören. Auf einem privaten Broker, bei dem Client IDs an die ausgegebenen Zugangsdaten gebunden sind, verwenden Sie statt eines zufälligen Suffixes genau die Client ID, die Sie erhalten haben, und leiten Sie alle Publishes über einen einzigen Worker, wenn Sie nur einen Satz Zugangsdaten besitzen.

Der Consumer-Worker

Legen Sie tls_client.rb und mqtt_worker.rb in lib/ ab und ergänzen Sie ein Startskript bin/mqtt_worker, das Rails bootet, damit der Handler Models nutzen kann:

#!/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

Betreiben Sie ihn als eigenen Prozesstyp, zum Beispiel in einem Procfile:

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

Betreiben Sie pro Client ID genau eine Instanz des Workers. Bei zwei Prozessen mit derselben ID trennt der Broker den älteren, und beide werfen sich immer wieder gegenseitig hinaus. Brauchen Sie mehrere Consumer, die sich die Last teilen, sind Shared Subscriptions aus MQTT 5 die übliche Antwort, sie erfordern aber einen MQTT-5-Client; der Leitfaden zu Shared Subscriptions erklärt das Muster.

Effizient mit höheren Raten publizieren

Für jeden Publish eine TLS-Verbindung zu öffnen, ist einfach, doch jede kostet einen Handshake und eine neue Session auf dem Broker. Bei einigen Befehlen pro Minute spielt das keine Rolle. Bei Hunderten Nachrichten pro Minute halten Sie in einem lang laufenden Prozess eine Verbindung offen und speisen sie aus einer Queue. Ein eigener Publisher-Thread, der Einträge aus einer Ruby-Queue entnimmt, publiziert und sich bei Fehlern mit demselben Backoff wie der Worker neu verbindet, liefert Ihnen eine stabile Verbindung und eine einzige Client ID.

Leiten Sie alle Publishes über diesen einen Thread, statt ein Client-Objekt zwischen vielen Threads zu teilen. Das Gem dokumentiert keine Thread-Sicherheit für parallele Publishes, und die Serialisierung der Schreibvorgänge über eine Queue verhindert verschränkte Pakete auf dem Socket. Außerdem erhalten Sie so eine natürliche Stelle für Backpressure: Wächst die Queue über ein Limit hinaus, lehnen Sie neue Arbeit ab oder verwerfen Telemetrie mit niedriger Priorität, statt den Speicher wachsen zu lassen, während der Broker nicht erreichbar ist.

Checkliste für die Produktion

1. Erstellen Sie einen privaten Broker auf einem verwalteten MQTT-Broker und kopieren Sie den generierten Benutzernamen, das Passwort und die Client ID.
2. Speichern Sie MQTT_USERNAME, MQTT_PASSWORD, MQTT_CLIENT_ID und MQTT_TOPIC_PREFIX in den Rails Credentials oder im Secret Store Ihrer Plattform.
3. Verbinden Sie sich auf Port 8883 mit VERIFY_PEER; liefern Sie niemals VERIFY_NONE aus.
4. Betreiben Sie den Worker als separaten, überwachten Prozess mit automatischen Neustarts.
5. Machen Sie den Handler idempotent: QoS 1 kann eine Nachricht mehr als einmal zustellen.
6. Validieren Sie jede Nutzereingabe, die in einem Topic-Namen landet.

Fehlerbehebung

Für andere Backend-Sprachen mit denselben Mustern lesen Sie das PHP MQTT Client Tutorial und den Leitfaden MQTT-Clients testen.

Starten Sie mit ConnectMQ

Verbinden Sie Ihre Ruby-App in wenigen Minuten mit einem privaten, per TLS gesicherten Broker. Registrieren Sie sich unter app.connectmq.com für einen kostenlosen Serverless Broker ohne Kreditkarte, setzen Sie die generierten Zugangsdaten als Umgebungsvariablen und beobachten Sie die Topics Ihres Workers im Live-Topic-Explorer des Portals. ConnectMQ betreibt den Broker, sodass Ihr Team nur Ruby-Code pflegt.

Häufig gestellte Fragen

Unterstützt das Ruby-Gem mqtt MQTT 5?

Nein, es implementiert MQTT 3.1.1. Das ist voll kompatibel mit ConnectMQ, das sowohl 3.1.1 als auch 5.0 unterstützt.

Verbindet sich der Ruby MQTT Client automatisch neu?

Nein. Kapseln Sie connect, subscribe und get in eine Schleife, die Netzwerkfehler abfängt, mit Backoff schläft und neu abonniert, wie im Worker oben.

Kann ich MQTT innerhalb eines Rails-Web-Requests abonnieren?

Davon ist abzuraten. Abonnements brauchen einen lang laufenden Prozess. Betreiben Sie daher einen separaten Worker und beschränken Sie Web-Requests auf schnelle Publishes über Hintergrund-Jobs.

Wie prüfe ich das Broker-Zertifikat in Ruby?

Setzen Sie client.ssl = true und rufen Sie client.ssl_context.set_params(verify_mode: OpenSSL::SSL::VERIFY_PEER) auf. Das nutzt den CA-Store des Systems und prüft den Hostnamen.