MQTT auf dem Raspberry Pi: vom Sensor zum Broker in 15 Minuten

Um Sensorwerte von einem Raspberry Pi in MQTT zu bringen, brauchst du drei Bausteine: eine Sensor-Bibliothek, den paho-mqtt-Client und einen Dienstmanager, der das Skript am Leben hält. Dieses Tutorial schließt einen BME280 (oder einen DHT22) an den Pi an, publiziert Temperatur, Luftfeuchte und Luftdruck alle 10 Sekunden als JSON und betreibt den Publisher als systemd-Dienst, der sich selbst neu verbindet und beim Booten startet.

Alles Folgende ist für Raspberry Pi OS Bookworm (oder neuer) mit Python 3.11+, paho-mqtt 2.x und den CircuitPython-Sensortreibern von Adafruit geschrieben. Es läuft auf Pi Zero 2 W, Pi 3, Pi 4 und Pi 5.

Was du brauchst

Verbinde den BME280 mit 3V3 (Pin 1), GND (Pin 6), SDA (Pin 3, GPIO2) und SCL (Pin 5, GPIO3). Ein DHT22 kommt an 3V3, GND und einen Daten-Pin wie GPIO4 (Pin 7), mit einem 10-kΩ-Pull-up-Widerstand, falls dein Modul keinen eingebaut hat.

Auf dem Pi muss kein Broker laufen. Wenn du trotzdem einen dort willst, lies einen Broker auf dem Raspberry Pi betreiben; dieser Artikel behandelt nur die Client-Seite.

1. I2C aktivieren und Sensor prüfen

sudo raspi-config nonint do_i2c 0
sudo apt update && sudo apt install -y i2c-tools python3-venv
i2cdetect -y 1

i2cdetect gibt ein Raster aus. Dort sollte 76 oder 77 stehen – die Adresse des BME280. Ein leeres Raster bedeutet ein Verdrahtungsproblem (vertauschte SDA- und SCL-Leitungen sind der Klassiker) oder dass I2C noch nicht aktiv ist; starte nach raspi-config bei Bedarf einmal neu.

2. Virtuelle Umgebung anlegen

Bookworm verweigert ein systemweites pip install mit error: externally-managed-environment. Nutze stattdessen ein venv in einem Projektordner:

mkdir -p ~/sensors && cd ~/sensors
python3 -m venv .venv
.venv/bin/pip install paho-mqtt adafruit-circuitpython-bme280
# für einen DHT22 stattdessen:
# .venv/bin/pip install paho-mqtt adafruit-circuitpython-dht

Die Adafruit-Treiber ziehen Blinka mit, die Kompatibilitätsschicht, die board und busio unter Linux bereitstellt.

Ein kurzer Test vor jeglichem MQTT-Code:

.venv/bin/python -c "import board; from adafruit_bme280 import basic as b; s=b.Adafruit_BME280_I2C(board.I2C(), address=0x76); print(s.temperature, s.relative_humidity, s.pressure)"

Bei ValueError: No I2C device at address: 0x76 probierst du address=0x77.

3. Das Publisher-Skript

Das Skript verbindet sich mit einem Last Will, damit Subscriber merken, wenn der Pi wegfällt, publiziert einen retained online-Status und sendet dann alle 10 Sekunden einen JSON-Messwert. Die Zugangsdaten kommen aus Umgebungsvariablen, sodass dieselbe Datei auf deinem Schreibtisch und in Produktion funktioniert.

import json, os, socket, time
import board
import paho.mqtt.client as mqtt
from adafruit_bme280 import basic as adafruit_bme280

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.getenv("MQTT_PORT", "8883"))
DEVICE = socket.gethostname()
BASE = f"testmqtt/pi/{DEVICE}"
INTERVAL = 10

sensor = adafruit_bme280.Adafruit_BME280_I2C(board.I2C(), address=0x76)

def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"Verbindung fehlgeschlagen: {reason_code}", flush=True)
        return
    print("verbunden", flush=True)
    client.publish(f"{BASE}/status", "online", qos=1, retain=True)

def on_disconnect(client, userdata, flags, reason_code, properties):
    print(f"getrennt: {reason_code}", flush=True)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"pi-{DEVICE}")
client.username_pw_set(os.getenv("MQTT_USER"), os.getenv("MQTT_PASS"))
if PORT == 8883:
    client.tls_set()                      # CA-Speicher des Systems
client.will_set(f"{BASE}/status", "offline", qos=1, retain=True)
client.reconnect_delay_set(min_delay=1, max_delay=60)
client.on_connect = on_connect
client.on_disconnect = on_disconnect

client.connect_async(HOST, PORT, keepalive=60)
client.loop_start()                       # Netzwerk-Thread, versucht es bis zur Verbindung

while True:
    reading = {
        "temperature": round(sensor.temperature, 2),
        "humidity": round(sensor.relative_humidity, 1),
        "pressure": round(sensor.pressure, 1),
        "ts": int(time.time()),
    }
    info = client.publish(f"{BASE}/bme280", json.dumps(reading), qos=1)
    print(f"publish mid={info.mid} rc={info.rc} {reading}", flush=True)
    time.sleep(INTERVAL)

Ein paar Entscheidungen, die eine Erklärung verdienen:

• connect_async() + loop_start() – beim Booten ist das Netzwerk eventuell noch nicht bereit. connect() würde sofort eine Exception werfen; mit connect_async() versucht es der Hintergrund-Thread von paho weiter, und derselbe Thread verbindet sich nach jedem späteren Abbruch neu, mit dem Backoff aus reconnect_delay_set().
• QoS 1 für Messwerte, damit ein kurzer Ausfall keine Daten stillschweigend verschluckt. paho puffert QoS-1-Nachrichten während der Trennung und sendet sie, sobald die Verbindung wieder steht. QoS 0 wäre leichter, verwirft aber alles, was während des Ausfalls gesendet wird.
• Retained Status + Last Will – wer später abonniert, sieht sofort online oder offline. Das Muster erklärt MQTT Last Will.
• Client-ID aus dem Hostnamen – stabil und eindeutig pro Pi. Zwei Geräte mit derselben ID werfen sich gegenseitig in einer Schleife vom Broker.

Stattdessen einen DHT22 verwenden

Das DHT22-Protokoll ist unter Linux zeitkritisch, deshalb scheitern Messungen regelmäßig mit RuntimeError (zum Beispiel "Checksum did not validate" oder "A full buffer was not returned"). Das ist normal – verwirf den Wert und versuche es erneut:

import adafruit_dht
dht = adafruit_dht.DHT22(board.D4)

def read_dht():
    for _ in range(5):
        try:
            return {"temperature": dht.temperature, "humidity": dht.humidity}
        except RuntimeError:
            time.sleep(2)              # der DHT22 braucht ~2 s zwischen Messungen
    return None                        # in dieser Runde nichts publizieren

Rufe read_dht() in der Hauptschleife auf und publiziere nur, wenn ein Wert zurückkommt.

4. Von Hand testen

Lege die Zugangsdaten in einer Datei ab, die nur dein Benutzer lesen darf, und starte das Skript im Vordergrund:

MQTT_HOST=broker.connectmq.com
MQTT_PORT=8883
MQTT_USER=username
MQTT_PASS=password

chmod 600 ~/sensors/mqtt.env
set -a; . ~/sensors/mqtt.env; set +a
.venv/bin/python publish.py

Du solltest verbunden sehen und danach alle 10 Sekunden eine Zeile publish mid=.... Jetzt beobachtest du von einem anderen Rechner aus:

mosquitto_sub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
  -u "$MQTT_USER" -P "$MQTT_PASS" -t 'testmqtt/pi/#' -v

Oder du sparst dir die Installation, öffnest den Online-MQTT-Client im Browser, verbindest dich mit demselben Broker und abonnierst testmqtt/pi/#. Zuerst erscheint die retained status-Nachricht, danach in jedem Intervall ein frischer bme280-JSON-Payload.

Messwerte deines Pi im Browser ansehen

Kommt nichts an, prüfe den ausgegebenen Reason Code. Bad user name or password und Not authorized sind Ablehnungen durch den Broker; schlag sie im MQTT-Reason-Code-Tool nach. Ein Timeout bedeutet meist einen blockierten Port – der Leitfaden zu Verbindungsfehlern geht jeden Fall durch.

5. Als systemd-Dienst betreiben

Ein Skript aus einer SSH-Sitzung stirbt beim Abmelden. systemd startet es beim Booten, startet es nach einem Absturz neu und sammelt die Ausgabe im Journal.

[Unit]
Description=BME280 to MQTT publisher
Wants=network-online.target
After=network-online.target

[Service]
User=pi
WorkingDirectory=/home/pi/sensors
EnvironmentFile=/home/pi/sensors/mqtt.env
ExecStart=/home/pi/sensors/.venv/bin/python -u /home/pi/sensors/publish.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

Ersetze pi durch deinen echten Benutzernamen – aktuelle Raspberry-Pi-OS-Images lassen dich bei der Einrichtung einen wählen. Der Benutzer muss in der Gruppe i2c sein (bzw. gpio für einen DHT22); der Standardbenutzer ist es bereits.

sudo systemctl daemon-reload
sudo systemctl enable --now bme280-mqtt
journalctl -u bme280-mqtt -f

-u in der Python-Kommandozeile schaltet die Ausgabepufferung ab, damit Logzeilen sofort im Journal erscheinen (das flush=True im Skript bewirkt dasselbe).

Beweise, dass es Ausfälle übersteht

1. sudo reboot – die Messwerte sollten innerhalb einer Minute wieder kommen, ganz ohne offene SSH-Sitzung.
2. Zieh für eine Minute das Netzwerk. Subscriber sehen offline, sobald der Broker das fehlende Keep-Alive bemerkt (bis zu 1,5 × 60 Sekunden), nach dem Reconnect wieder online, gefolgt von den während des Ausfalls gepufferten Messwerten.
3. sudo systemctl kill bme280-mqtt – systemd startet das Skript nach 5 Sekunden neu.

Wie es weitergeht

• Bring den Payload in Form für Dashboards und Datenbanken: JSON-Payloads in Python.
• Mehr Sensoren und ein Display: Die MQTT-Wetterstation baut auf diesem Skript auf.
• Versorge Home Assistant über Discovery-Nachrichten, beschrieben in Home Assistant MQTT.

Ein Pi, ein Sensor und vierzig Zeilen Python reichen für zuverlässige Telemetrie, solange die langweiligen Teile erledigt sind: ein venv für die Abhängigkeiten, eine stabile Client-ID, ein Last Will für die Präsenz und systemd für Neustarts. Teste jeden Ausfall einmal auf dem Schreibtisch, dann verhält sich der Pi im Garten genauso.

Häufig gestellte Fragen