The clean way to use MQTT in Laravel is to split the two directions. Publish from queued jobs, so HTTP requests return immediately and failed sends are retried. Subscribe in an Artisan command that runs as a supervised long-lived process and dispatches a job for each message. Both sides use the official php-mqtt/laravel-client package and the same .env configuration.
This guide builds both halves for a small device-control app, deploys the listener with Supervisor and shows how to test it with Laravel's fakes and a real broker.
Install and configure
composer require php-mqtt/laravel-client
php artisan vendor:publish --provider="PhpMqtt\Client\MqttClientServiceProvider" --tag="config"This creates config/mqtt-client.php with a default connection that reads from environment variables:
MQTT_HOST=broker.connectmq.com
MQTT_PORT=8883
MQTT_TLS_ENABLED=true
MQTT_AUTH_USERNAME="${MQTT_USER}"
MQTT_AUTH_PASSWORD="${MQTT_PASS}"
# Leave MQTT_CLIENT_ID empty: each publisher connection gets a random ID
MQTT_CLIENT_ID=Add a second connection for the listener. It needs a fixed client ID so you can recognize it in broker logs, while publishers keep random IDs so that parallel queue workers never collide (see duplicate client ID):
'connections' => [
'default' => [ /* as published */ ],
'subscriber' => [
'host' => env('MQTT_HOST'),
'port' => env('MQTT_PORT', 8883),
'client_id' => env('MQTT_LISTENER_CLIENT_ID', 'testmqtt-laravel-listener'),
'use_clean_session' => true,
'connection_settings' => [
'tls' => ['enabled' => env('MQTT_TLS_ENABLED', true)],
'auth' => [
'username' => env('MQTT_LISTENER_USERNAME', env('MQTT_AUTH_USERNAME')),
'password' => env('MQTT_LISTENER_PASSWORD', env('MQTT_AUTH_PASSWORD')),
],
'keep_alive_interval' => 30,
],
],
],Copy the remaining keys from the default block of your published file; key names can differ slightly between package versions.
Design the topics first
Laravel code gets simpler when the topic layout encodes what you need for routing. This tutorial uses:
| Topic | Direction | Payload |
|---|---|---|
testmqtt/laravel/devices/{id}/cmd | Laravel → device | {"action":"on"} |
testmqtt/laravel/devices/{id}/telemetry | device → Laravel | {"temp":21.5,"ts":1760700000} |
testmqtt/laravel/devices/{id}/status | device → everyone (retained) | online / offline |
The device ID sits at a fixed position, so the listener can extract it without parsing the payload. More guidance is in topic design best practices.
Publish from a queued job
<?php
namespace App\Jobs;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use PhpMqtt\Client\Facades\MQTT;
use PhpMqtt\Client\MqttClient;
class SendDeviceCommand implements ShouldQueue
{
use Queueable;
public int $tries = 5;
public array $backoff = [2, 10, 30, 60];
public function __construct(public string $deviceId, public array $command) {}
public function handle(): void
{
try {
$mqtt = MQTT::connection();
$mqtt->publish(
"testmqtt/laravel/devices/{$this->deviceId}/cmd",
json_encode($this->command),
MqttClient::QOS_AT_LEAST_ONCE,
);
$mqtt->loop(true, true); // wait for the PUBACK
} finally {
MQTT::disconnect();
}
}
}loop(true, true) waits until the broker acknowledged the QoS 1 message; without it the job could finish before delivery is confirmed. If the broker is unreachable, MQTT::connection() throws, the job fails and Laravel retries it after 2, 10, 30 and 60 seconds.
Disconnecting in finally costs one connection per job but avoids a subtle problem: an idle cached connection in a long-running queue worker is dropped by the broker after the keep-alive expires, and the next job would fail on a dead socket. If you send many messages at once, publish them in one job rather than one job per message.
Dispatch it from a controller or an API route:
use App\Jobs\SendDeviceCommand;
use Illuminate\Http\Request;
use Illuminate\Support\Facades\Route;
Route::post('/devices/{device}/commands', function (Request $request, string $device) {
$data = $request->validate(['action' => 'required|in:on,off']);
SendDeviceCommand::dispatch($device, $data);
return response()->json(['queued' => true], 202);
});Subscribe in an Artisan command
The listener connects once, subscribes, and turns each message into a queued job. Keep the callback short: heavy work inside it delays every following message and can push the client past its keep-alive.
<?php
namespace App\Console\Commands;
use App\Jobs\StoreTelemetry;
use Illuminate\Console\Command;
use PhpMqtt\Client\Facades\MQTT;
use PhpMqtt\Client\MqttClient;
class MqttListen extends Command
{
protected $signature = 'mqtt:listen';
protected $description = 'Subscribe to device telemetry and dispatch jobs';
public function handle(): int
{
$mqtt = MQTT::connection('subscriber');
// Stop cleanly when Supervisor or Docker sends SIGTERM
$this->trap([SIGTERM, SIGINT], fn () => $mqtt->interrupt());
$mqtt->subscribe('testmqtt/laravel/devices/+/telemetry', function (string $topic, string $message) {
$deviceId = explode('/', $topic)[3];
StoreTelemetry::dispatch($deviceId, $message);
}, MqttClient::QOS_AT_LEAST_ONCE);
$this->info('Listening on testmqtt/laravel/devices/+/telemetry');
$mqtt->loop(true);
MQTT::disconnect('subscriber');
return self::SUCCESS;
}
}The job does the validation and the database write, with the queue's retries and failure handling:
<?php
namespace App\Jobs;
use App\Models\Telemetry;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Support\Facades\Log;
class StoreTelemetry implements ShouldQueue
{
use Queueable;
public function __construct(public string $deviceId, public string $payload) {}
public function handle(): void
{
$data = json_decode($this->payload, true);
if (!is_array($data) || !isset($data['temp'])) {
Log::warning('Invalid telemetry', ['device' => $this->deviceId, 'payload' => $this->payload]);
return;
}
Telemetry::create([
'device_id' => $this->deviceId,
'temp' => $data['temp'],
'measured_at' => isset($data['ts']) ? now()->setTimestamp($data['ts']) : now(),
]);
}
}Run the listener in production
The command blocks forever, so a process manager must start it, restart it when it exits and stop it on deploy. With Supervisor:
[program:mqtt-listen]
command=php /var/www/app/artisan mqtt:listen
user=www-data
autostart=true
autorestart=true
stopsignal=TERM
stopwaitsecs=15
redirect_stderr=true
stdout_logfile=/var/www/app/storage/logs/mqtt-listen.logThis also gives you the reconnect strategy for free: if the connection drops, loop() throws, the command exits and Supervisor starts a fresh one. Two points people miss:
- Restart it on every deploy (
supervisorctl restart mqtt-listen). Like queue workers, it loaded your code once at start-up;php artisan queue:restartdoes not touch custom commands. - Clean session and offline messages. With
use_clean_sessionset to true, messages published while the listener is down are not queued for it. If you need that, use a persistent session with a fixed client ID on your own private broker (the sharedbroker.connectmq.comendpoint keeps no offline sessions) — see persistent sessions.
Scaling to more than one listener
Starting the same command on two servers means both receive every message and every row is stored twice. Subscribe with a shared subscription such as $share/laravel/testmqtt/laravel/devices/+/telemetry (supported by most current brokers) so each message goes to one listener, give each instance its own client ID, and make StoreTelemetry idempotent. Details in MQTT shared subscriptions.
If browsers need the telemetry live, you can either broadcast from the job with Laravel's broadcasting or let the front end subscribe directly over secure WebSockets (port 8084, path /mqtt) with read-only credentials — see MQTT over WebSockets.
Test it
Unit tests should never need a broker. Fake the queue for the HTTP layer and mock the facade for the job:
<?php
namespace Tests\Feature;
use App\Jobs\SendDeviceCommand;
use Illuminate\Support\Facades\Queue;
use Mockery;
use PhpMqtt\Client\Contracts\MqttClient;
use PhpMqtt\Client\Facades\MQTT;
use Tests\TestCase;
class DeviceCommandTest extends TestCase
{
public function test_command_is_queued(): void
{
Queue::fake();
$this->postJson('/api/devices/42/commands', ['action' => 'on'])->assertAccepted();
Queue::assertPushed(SendDeviceCommand::class, fn ($job) => $job->deviceId === '42');
}
public function test_job_publishes_to_the_device_topic(): void
{
$client = Mockery::mock(MqttClient::class);
$client->shouldReceive('publish')->once()
->with('testmqtt/laravel/devices/42/cmd', '{"action":"on"}', 1);
$client->shouldReceive('loop')->once();
MQTT::shouldReceive('connection')->andReturn($client);
MQTT::shouldReceive('disconnect');
(new SendDeviceCommand('42', ['action' => 'on']))->handle();
}
}Then run one end-to-end check against a real broker. Start the listener and a queue worker, watch the command topic, and play the device from the terminal or the online MQTT client:
php artisan mqtt:listen &
php artisan queue:work &
# 1. Laravel -> device: watch commands, then call the API
mosquitto_sub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
-u "$MQTT_USER" -P "$MQTT_PASS" -t 'testmqtt/laravel/devices/+/cmd' -v &
curl -X POST localhost:8000/api/devices/42/commands -H 'Content-Type: application/json' -d '{"action":"on"}'
# 2. device -> Laravel: send telemetry and check the telemetry table
mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
-u "$MQTT_USER" -P "$MQTT_PASS" -t 'testmqtt/laravel/devices/42/telemetry' -m '{"temp":22.4}' -q 1Finally, run supervisorctl stop mqtt-listen and confirm it exits within stopwaitsecs — proof that the trap() handler works. The plain-PHP details behind all of this (QoS 1 and loop(true, true), TLS options, Last Will) are covered in MQTT in PHP.
Watch the Laravel topics in the online client
Laravel and MQTT fit together well once each side has its own process: queued jobs publish with retries and random client IDs, one supervised Artisan command subscribes and hands every message to the queue, and tests use fakes so only one end-to-end check needs a real broker.
Frequently asked questions
Where should the MQTT subscriber run in Laravel?
In a dedicated Artisan command (php artisan mqtt:listen) that calls loop(true) and is kept alive by Supervisor, systemd or a container restart policy. Never subscribe in a controller, middleware or service provider: web requests end, and every PHP-FPM worker would open its own connection.
Why publish MQTT messages from a queued job instead of the controller?
The HTTP response no longer waits for the broker, and if the broker is briefly unreachable the job is retried with back-off instead of failing the request. It also keeps all MQTT credentials and connection code in one place.
How do I run more than one Laravel MQTT listener without duplicate processing?
Two listeners with the same subscription both receive every message. Use a shared subscription ($share/<group>/<filter>) on a broker that supports it so each message goes to one member, give each listener its own client ID, and make the jobs idempotent.
