php使用Mosquitto扩展

订阅a.php

<?php

use Mosquitto\Client;

class MqttServer
{
    private $mqtt;

    public function __construct()
    {
        $this->mqtt = new Client();
        $base = str_replace('a.php', '', __FILE__);
        $this->mqtt->setTlsCertificates($base . 'key/ca.crt', $base . 'key/client.crt', $base . 'key/client.key');
        $this->mqtt->setTlsOptions(Client::SSL_VERIFY_NONE, null, null);
        $this->mqtt->setCredentials('root', '123456');
        $this->mqtt->connect("127.0.0.1", 8883, 5);
        $this->mqtt->onConnect([$this, 'handConnect']);
        $this->mqtt->onMessage([$this, 'handleMessage']);
        $this->mqtt->onDisconnect([$this, 'handleDisconnect']);
    }

    public function subscribeToTopics($topics)
    {
        foreach ($topics as $topic => $qos) {
            $this->mqtt->subscribe($topic, $qos);
        }
    }
    
        public function handConnect($rc)
    {
     echo "连接成功$rc\n";
    }
    
            public function publishData($msg,$topic,$qos)
    {
        $this->mqtt->publish($topic, $msg, $qos,false);
    }

    public function handleMessage($message)
    {
        printf("Got a message ID %d on topic %s with payload:\n%s\n\n", $message->mid, $message->topic, $message->payload);
    }

    public function handleDisconnect($rc)
    {
        echo "连接断开$rc\n";
    }

    public function start()
    {
        try {
            $this->mqtt->loop();
        } catch (\Throwable $e) {
            $this->mqtt->connect("127.0.0.1", 8883, 5);
        }
    }
}

$logger = new MqttServer();
$logger->subscribeToTopics(['test'=>0]);
while (true) {
    $logger->start();
}

发布b.php

<?php

use Mosquitto\Client;

class MqttServer
{
    private $mqtt;

    public function __construct()
    {
        $this->mqtt = new Client();
        $base = str_replace('b.php', '', __FILE__);
        $this->mqtt->setTlsCertificates($base . 'key/ca.crt', $base . 'key/client.crt', $base . 'key/client.key');
        $this->mqtt->setTlsOptions(Client::SSL_VERIFY_NONE, null, null);
        $this->mqtt->setCredentials('root', '123456');
        $this->mqtt->connect("127.0.0.1", 8883, 5);
        $this->mqtt->onConnect([$this, 'handConnect']);
        $this->mqtt->onMessage([$this, 'handleMessage']);
        $this->mqtt->onDisconnect([$this, 'handleDisconnect']);
    }

    public function subscribeToTopics($topics)
    {
        foreach ($topics as $topic => $qos) {
            $this->mqtt->subscribe($topic, $qos);
        }
    }

    public function handConnect($rc)
    {
        echo "连接成功$rc\n";
    }

    public function publishData($msg, $topic, $qos)
    {
        $mgid=$this->mqtt->publish($topic, $msg, $qos, false);
        if ($mgid) echo '推送成功'.$mgid."\r\n";
    }

    public function handleMessage($message)
    {
        printf("Got a message ID %d on topic %s with payload:\n%s\n\n", $message->mid, $message->topic, $message->payload);
    }

    public function handleDisconnect($rc)
    {
        echo "连接断开$rc\n";
    }

    public function start()
    {
        try {
            $this->mqtt->loop();
        } catch (\Throwable $e) {
            $this->mqtt->connect("127.0.0.1", 8883, 5);
        }
    }
}

$logger = new MqttServer();
while (true) {
    sleep(2);
    $msg = time() . '消息推送';
    $logger->publishData($msg, 'test', 0);
    $logger->start();
}

分别运行订阅与发布文件

image.png

image.png