php-mqtt客户端在发布端长时间离线后停止接收消息

环境与配置 #

• 系统：Ubuntu 22.04
• 服务：systemd托管的PHP MQTT消息接收服务
• 依赖：php-mqtt/client库、Mosquitto MQTT代理

PHP业务代码 #

<?php 
header("Access-Control-Allow-Origin: *");
require 'database.php'; 
require('vendor/autoload.php');

use PhpMqtt\Client\MqttClient;
use PhpMqtt\Client\ConnectionSettings;

ini_set('display_errors', 1);
error_reporting(E_ALL);

$server   = '<server>';
$port     = 1883;
$clientId = '<id>';
$username = '<user>';
$password = '<password>';

$connectionSettings = (new ConnectionSettings)
  ->setUsername($username)
  ->setPassword($password)
  ->setKeepAliveInterval(60)
  ->setLastWillQualityOfService(1)
  ->setConnectTimeout(60)
  ->setMaxReconnectAttempts(PHP_INT_MAX)
  ->setReconnectAutomatically(true);

$mqtt = new MqttClient($server, $port, $clientId, MqttClient::MQTT_3_1);
$mqtt->connect($connectionSettings, false);

function append($db, $sample) {
  try {
    $stmt = $db->prepare("INSERT INTO sensors (sampled, temperature, humidity) VALUES (CURRENT_TIMESTAMP(), :temperature, :humidity);");
    $stmt->execute($sample);
  } catch (PDOException $e) {
    echo "Error: " . $e->getMessage();
  }    
}

$mqtt->subscribe('<topic1>', function ($topic, $message) use (&$data, &$dbApp) {
  printf("Received message on topic [%s]: %s\n", $topic, $message);

  if ($message === 'Offline') {
    $sample = [
      'temperature' => NULL,
      'humidity' => NULL
    ];

    append($dbApp, $sample);
  }

}, 0);

$mqtt->subscribe('<topic2>', function ($topic, $message) use (&$data, &$dbApp) {
  printf("Received message on topic [%s]: %s\n", $topic, $message);
  $obj = json_decode($message);

  $sample = [
    'temperature' => floatval($obj->Temperature),
    'humidity' => floatval($obj->Humidity)
  ];
  
  append($dbApp, $sample);
}, 0);
   
$mqtt->loop(true);
$mqtt->close();
$dbApp = null;      

systemd单元配置 #

[Unit]
Description=MQTT receiver
 
[Service]
Type=simple
ExecStart=/usr/bin/php /usr/share/nginx/html/mqtt.php
Restart=always
RestartSec=10
 
[Install]
WantedBy=multi-user.target

问题现象 #

1. 初始测试流程正常：启动代理、服务、发布端后，可正常接收<topic1>的LWT离线消息和<topic2>的业务数据；关闭发布端能收到离线消息，短时间重启发布端后仍可正常接收。
2. 异常情况：当发布端离线数小时（如过夜）后重启，PHP客户端无法再接收任何消息，无Received message on topic...输出；但用mosquitto_sub订阅相同主题可正常接收，systemd服务状态活跃且无报错，仅显示前一日日志，重启服务后恢复正常。

排查方向与解决方案 #

1. 重连后订阅丢失 #

原因：php-mqtt客户端在MQTT 3.1协议下自动重连时，默认不会重新注册之前的订阅（MQTT 3.1.1及5.0部分客户端会保留，但3.1无此机制），长时间离线重连后，客户端未重新订阅主题，导致无法接收消息。

解决：注册重连回调函数，在每次重连成功后重新执行订阅逻辑：

// 定义可复用的订阅逻辑
function subscribeToTopics($mqtt, $dbApp) {
    $mqtt->subscribe('<topic1>', function ($topic, $message) use ($dbApp) {
        printf("Received message on topic [%s]: %s\n", $topic, $message);
        if ($message === 'Offline') {
            $sample = ['temperature' => NULL, 'humidity' => NULL];
            append($dbApp, $sample);
        }
    }, 0);

    $mqtt->subscribe('<topic2>', function ($topic, $message) use ($dbApp) {
        printf("Received message on topic [%s]: %s\n", $topic, $message);
        $obj = json_decode($message);
        $sample = [
            'temperature' => floatval($obj->Temperature),
            'humidity' => floatval($obj->Humidity)
        ];
        append($dbApp, $sample);
    }, 0);
}

// 首次连接后订阅
subscribeToTopics($mqtt, $dbApp);

// 注册重连回调，重连后重新订阅
$mqtt->registerReconnectHandler(function (MqttClient $client, ConnectionSettings $settings) use ($dbApp) {
    error_log("MQTT client reconnected, re-subscribing to topics");
    subscribeToTopics($client, $dbApp);
});

2. 数据库连接失效 #

原因：长时间空闲后，数据库服务器会主动关闭闲置的PDO连接，重连后PHP客户端未重建数据库连接，导致订阅回调中执行数据库操作时静默失败（无报错输出），但不会影响MQTT连接状态。

解决：在append函数中添加连接有效性检查，自动重建连接：

function append(&$db, $sample) {
    try {
        // 检查连接是否有效
        if ($db instanceof PDO) {
            $db->query('SELECT 1');
        } else {
            // 重新初始化数据库连接（需对应database.php中的初始化逻辑）
            require 'database.php';
        }
        
        $stmt = $db->prepare("INSERT INTO sensors (sampled, temperature, humidity) VALUES (CURRENT_TIMESTAMP(), :temperature, :humidity);");
        $stmt->execute($sample);
    } catch (PDOException $e) {
        error_log("Database error: " . $e->getMessage());
        // 尝试重新连接并重试
        require 'database.php';
        append($db, $sample);
    }    
}

3. 增强日志排查能力 #

原因：当前仅用printf输出消息，systemd可能未完整捕获输出，且缺少重连事件、错误信息的日志，难以定位问题。

解决：

• 修改systemd服务配置，确保日志输出到journal并同步到控制台：

[Service]
Type=simple
ExecStart=/usr/bin/php /usr/share/nginx/html/mqtt.php
Restart=always
RestartSec=10
StandardOutput=journal+console
StandardError=journal+console

• 在代码中添加关键事件的错误日志，比如重连、连接失败、数据库错误等，使用error_log()函数，方便通过journalctl -u mqtt.service查看完整日志。

4. 优化连接超时与心跳配置 #

原因：网络中间设备（如路由器、防火墙）可能会提前断开长时间空闲的TCP连接，即使MQTT设置了KeepAlive，也可能因网络层超时导致连接失效但客户端未感知。

解决：

• 缩短MQTT KeepAlive间隔至30秒，增加心跳频率，并启用TCP层心跳：

$connectionSettings = (new ConnectionSettings)
  // ... 其他配置
  ->setKeepAliveInterval(30)
  ->setTcpKeepAlive(true); // 启用TCP层心跳，防止中间设备断开连接

• 调整setConnectTimeout为30秒，避免重连等待过久。

内容的提问来源于stack exchange，提问作者Mark