php實(shí)現(xiàn)mqtt發(fā)布/發(fā)送消息到主題

eechen 發(fā)布于2019-06-28 18:38 / 2073人閱讀

摘要：實(shí)現(xiàn)發(fā)布發(fā)送消息到主題是啥我的博客有寫這個(gè)東西傳送門想要實(shí)現(xiàn)需要使用到中的函數(shù)函數(shù)是什么此次使用的是網(wǎng)上開源案例其中使用的是系列函數(shù)什么是系列函數(shù)大概意思是正如你所指出的，是核心內(nèi)置的，始終可用，而套接字是很少包含的擴(kuò)展的一部分。

php實(shí)現(xiàn)mqtt發(fā)布/發(fā)送消息到主題

mqtt是啥?我的博客有寫這個(gè)東西:傳送門

php想要實(shí)現(xiàn)mqtt需要使用到php中的socket函數(shù);

socket函數(shù)是什么?

此次使用的是網(wǎng)上開源mqtt案例:其中使用的是 stream_socket_xxxx 系列函數(shù)

什么是stream_socket_xxxx系列函數(shù)

大概意思是:

正如你所指出的，"stream"是PHP核心（內(nèi)置的，始終可用），而"套接字"是很少包含的擴(kuò)展的一部分。除此之外，它們幾乎完全相同。您可以同時(shí)使用TCP和UDP兩種流，也可以使用阻塞和非阻塞模式，這些模式涵蓋了所有用例的99％。我能想到的唯一常見的例外是ICMP。例如，"ping"。但是，看起來目前還沒有一種安全的方式來從PHP執(zhí)行ICMP。這種調(diào)用需要通過套接字?jǐn)U展來實(shí)現(xiàn)SOCK_RAW，這需要執(zhí)行“root”權(quán)限。此外，并非所有路由器都會(huì)在TCP，UDP和ICMP之外路由其他數(shù)據(jù)包類型。這限制了套接字?jǐn)U展的實(shí)用性。

MQTT類代碼:

/* phpMQTT */
class Mqtt {

    private $socket;             /* holds the socket    */
    private $msgid = 1;            /* counter for message id */
    public $keepalive = 10;        /* default keepalive timmer */
    public $timesinceping;        /* host unix time, used to detect disconects */
    public $topics = array();     /* used to store currently subscribed topics */
    public $debug = false;        /* should output debug messages */
    public $address;            /* broker address */
    public $port;                /* broker port */
    public $clientid;            /* client id sent to brocker */
    public $will;                /* stores the will of the client */
    private $username;            /* stores username */
    private $password;            /* stores password */

    public $cafile;

    function __construct($address, $port, $clientid, $cafile = NULL){
        $this->broker($address, $port, $clientid, $cafile);
    }

    /* sets the broker details */
    function broker($address, $port, $clientid, $cafile = NULL){
        $this->address = $address;
        $this->port = $port;
        $this->clientid = $clientid;
        $this->cafile = $cafile;
    }

    function connect_auto($clean = true, $will = NULL, $username = NULL, $password = NULL){
        while($this->connect($clean, $will, $username, $password)==false){
            sleep(10);
        }
        return true;
    }

    /* connects to the broker
        inputs: $clean: should the client send a clean session flag */
    function connect($clean = true, $will = NULL, $username = NULL, $password = NULL){

        if($will) $this->will = $will;
        if($username) $this->username = $username;
        if($password) $this->password = $password;


        if ($this->cafile) {
            $socketContext = stream_context_create(["ssl" => [
                "verify_peer_name" => true,
                "cafile" => $this->cafile
            ]]);
            $this->socket = stream_socket_client("tls://" . $this->address . ":" . $this->port, $errno, $errstr, 60, STREAM_CLIENT_CONNECT, $socketContext);
        } else {
            $this->socket = stream_socket_client("tcp://" . $this->address . ":" . $this->port, $errno, $errstr, 60, STREAM_CLIENT_CONNECT);
        }

        if (!$this->socket ) {
            if($this->debug) error_log("stream_socket_create() $errno, $errstr 
");
            return false;
        }

        stream_set_timeout($this->socket, 5);
        stream_set_blocking($this->socket, 0);

        $i = 0;
        $buffer = "";

        $buffer .= chr(0x00); $i++;
        $buffer .= chr(0x06); $i++;
        $buffer .= chr(0x4d); $i++;
        $buffer .= chr(0x51); $i++;
        $buffer .= chr(0x49); $i++;
        $buffer .= chr(0x73); $i++;
        $buffer .= chr(0x64); $i++;
        $buffer .= chr(0x70); $i++;
        $buffer .= chr(0x03); $i++;

        //No Will
        $var = 0;
        if($clean) $var+=2;

        //Add will info to header
        if($this->will != NULL){
            $var += 4; // Set will flag
            $var += ($this->will["qos"] << 3); //Set will qos
            if($this->will["retain"])    $var += 32; //Set will retain
        }

        if($this->username != NULL) $var += 128;    //Add username to header
        if($this->password != NULL) $var += 64;    //Add password to header

        $buffer .= chr($var); $i++;

        //Keep alive
        $buffer .= chr($this->keepalive >> 8); $i++;
        $buffer .= chr($this->keepalive & 0xff); $i++;

        $buffer .= $this->strwritestring($this->clientid,$i);

        //Adding will to payload
        if($this->will != NULL){
            $buffer .= $this->strwritestring($this->will["topic"],$i);
            $buffer .= $this->strwritestring($this->will["content"],$i);
        }

        if($this->username) $buffer .= $this->strwritestring($this->username,$i);
        if($this->password) $buffer .= $this->strwritestring($this->password,$i);

        $head = "  ";
        $head{0} = chr(0x10);
        $head{1} = chr($i);

        fwrite($this->socket, $head, 2);
        fwrite($this->socket,  $buffer);

        $string = $this->read(4);

        if(ord($string{0})>>4 == 2 && $string{3} == chr(0)){
            if($this->debug) echo "Connected to Broker
";
        }else{
            error_log(sprintf("Connection failed! (Error: 0x%02x 0x%02x)
",
                ord($string{0}),ord($string{3})));
            return false;
        }

        $this->timesinceping = time();

        return true;
    }

    /* read: reads in so many bytes */
    function read($int = 8192, $nb = false){

        //    print_r(socket_get_status($this->socket));

        $string="";
        $togo = $int;

        if($nb){
            return fread($this->socket, $togo);
        }

        while (!feof($this->socket) && $togo>0) {
            $fread = fread($this->socket, $togo);
            $string .= $fread;
            $togo = $int - strlen($string);
        }




        return $string;
    }

    /* subscribe: subscribes to topics */
    function subscribe($topics, $qos = 0){
        $i = 0;
        $buffer = "";
        $id = $this->msgid;
        $buffer .= chr($id >> 8);  $i++;
        $buffer .= chr($id % 256);  $i++;

        foreach($topics as $key => $topic){
            $buffer .= $this->strwritestring($key,$i);
            $buffer .= chr($topic["qos"]);  $i++;
            $this->topics[$key] = $topic;
        }

        $cmd = 0x80;
        //$qos
        $cmd +=    ($qos << 1);


        $head = chr($cmd);
        $head .= chr($i);

        fwrite($this->socket, $head, 2);
        fwrite($this->socket, $buffer, $i);
        $string = $this->read(2);

        $bytes = ord(substr($string,1,1));
        $string = $this->read($bytes);
    }

    /* ping: sends a keep alive ping */
    function ping(){
        $head = " ";
        $head = chr(0xc0);
        $head .= chr(0x00);
        fwrite($this->socket, $head, 2);
        if($this->debug) echo "ping sent
";
    }

    /* disconnect: sends a proper disconect cmd */
    function disconnect(){
        $head = " ";
        $head{0} = chr(0xe0);
        $head{1} = chr(0x00);
        fwrite($this->socket, $head, 2);
    }

    /* close: sends a proper disconect, then closes the socket */
    function close(){
        $this->disconnect();
        stream_socket_shutdown($this->socket, STREAM_SHUT_WR);
    }

    /* publish: publishes $content on a $topic */
    function publish($topic, $content, $qos = 0, $retain = 0){

        $i = 0;
        $buffer = "";

        $buffer .= $this->strwritestring($topic,$i);

        //$buffer .= $this->strwritestring($content,$i);

        if($qos){
            $id = $this->msgid++;
            $buffer .= chr($id >> 8);  $i++;
            $buffer .= chr($id % 256);  $i++;
        }

        $buffer .= $content;
        $i+=strlen($content);


        $head = " ";
        $cmd = 0x30;
        if($qos) $cmd += $qos << 1;
        if($retain) $cmd += 1;

        $head{0} = chr($cmd);
        $head .= $this->setmsglength($i);

        fwrite($this->socket, $head, strlen($head));
        fwrite($this->socket, $buffer, $i);

    }

    /* message: processes a recieved topic */
    function message($msg){
        $tlen = (ord($msg{0})<<8) + ord($msg{1});
        $topic = substr($msg,2,$tlen);
        $msg = substr($msg,($tlen+2));
        $found = 0;
        foreach($this->topics as $key=>$top){
            if( preg_match("/^".str_replace("#",".*",
                    str_replace("+","[^/]*",
                        str_replace("/","/",
                            str_replace("$","$",
                                $key))))."$/",$topic) ){
                if(is_callable($top["function"])){
                    call_user_func($top["function"],$topic,$msg);
                    $found = 1;
                }
            }
        }

        if($this->debug && !$found) echo "msg recieved but no match in subscriptions
";
    }

    /* proc: the processing loop for an "allways on" client
        set true when you are doing other stuff in the loop good for watching something else at the same time */
    function proc( $loop = true){

        if(1){
            $sockets = array($this->socket);
            $w = $e = NULL;
            $cmd = 0;

            //$byte = fgetc($this->socket);
            if(feof($this->socket)){
                if($this->debug) echo "eof receive going to reconnect for good measure
";
                fclose($this->socket);
                $this->connect_auto(false);
                if(count($this->topics))
                    $this->subscribe($this->topics);
            }

            $byte = $this->read(1, true);

            if(!strlen($byte)){
                if($loop){
                    usleep(100000);
                }

            }else{

                $cmd = (int)(ord($byte)/16);
                if($this->debug) echo "Recevid: $cmd
";

                $multiplier = 1;
                $value = 0;
                do{
                    $digit = ord($this->read(1));
                    $value += ($digit & 127) * $multiplier;
                    $multiplier *= 128;
                }while (($digit & 128) != 0);

                if($this->debug) echo "Fetching: $value
";

                if($value)
                    $string = $this->read($value);

                if($cmd){
                    switch($cmd){
                        case 3:
                            $this->message($string);
                            break;
                    }

                    $this->timesinceping = time();
                }
            }

            if($this->timesinceping < (time() - $this->keepalive )){
                if($this->debug) echo "not found something so ping
";
                $this->ping();
            }


            if($this->timesinceping<(time()-($this->keepalive*2))){
                if($this->debug) echo "not seen a package in a while, disconnecting
";
                fclose($this->socket);
                $this->connect_auto(false);
                if(count($this->topics))
                    $this->subscribe($this->topics);
            }

        }
        return 1;
    }

    /* getmsglength: */
    function getmsglength(&$msg, &$i){

        $multiplier = 1;
        $value = 0 ;
        do{
            $digit = ord($msg{$i});
            $value += ($digit & 127) * $multiplier;
            $multiplier *= 128;
            $i++;
        }while (($digit & 128) != 0);

        return $value;
    }


    /* setmsglength: */
    function setmsglength($len){
        $string = "";
        do{
            $digit = $len % 128;
            $len = $len >> 7;
            // if there are more digits to encode, set the top bit of this digit
            if ( $len > 0 )
                $digit = ($digit | 0x80);
            $string .= chr($digit);
        }while ( $len > 0 );
        return $string;
    }

    /* strwritestring: writes a string to a buffer */
    function strwritestring($str, &$i){
        $ret = " ";
        $len = strlen($str);
        $msb = $len >> 8;
        $lsb = $len % 256;
        $ret = chr($msb);
        $ret .= chr($lsb);
        $ret .= $str;
        $i += ($len+2);
        return $ret;
    }

    function printstr($string){
        $strlen = strlen($string);
        for($j=0;$j<$strlen;$j++){
            $num = ord($string{$j});
            if($num > 31)
                $chr = $string{$j}; else $chr = " ";
            printf("%4d: %08b : 0x%02x : %s 
",$j,$num,$num,$chr);
        }
    }
}

實(shí)現(xiàn)部分 發(fā)送到主題

 // 發(fā)送給訂閱號(hào)信息,創(chuàng)建socket,無sam隊(duì)列
 $server = "127.0.0.1";     // 服務(wù)代理地址(mqtt服務(wù)端地址)
 $port = 1883;                     // 通信端口
 $username = "";                   // 用戶名(如果需要)
 $password = "";                   // 密碼(如果需要
 $client_id = "clientx9293670xxctr"; // 設(shè)置你的連接客戶端id
 $mqtt = new Mqtt($server, $port, $client_id); //實(shí)例化MQTT類
 if ($mqtt->connect(true, NULL, $username, $password)) { 
    //如果創(chuàng)建鏈接成功
     $mqtt->publish("xxx3809293670ctr", "setr=3xxxxxxxxx", 0); 
     // 發(fā)送到 xxx3809293670ctr 的主題 一個(gè)信息 內(nèi)容為 setr=3xxxxxxxxx Qos 為 0 
     $mqtt->close();    //發(fā)送后關(guān)閉鏈接
 } else {
     echo "Time out!
"; 
 }

訂閱主題

/*// 訂閱信息,接收一個(gè)信息后退出
$server = "127.0.0.1";     // 服務(wù)代理地址(mqtt服務(wù)端地址)
$port = 1883;                     // 通信端口
$username = "";                   // 用戶名(如果需要)
$password = "";                   // 密碼(如果需要
$client_id = "clientx9293670xxctr"; // 設(shè)置你的連接客戶端id

$mqtt = new Mqtt($server, $port, $client_id);

if(!$mqtt->connect(true, NULL, $username, $password)) { //鏈接不成功再重復(fù)執(zhí)行監(jiān)聽連接
    exit(1);
}

$topics["SN69143809293670state"] = array("qos" => 0, "function" => "procmsg");
// 訂閱主題為 SN69143809293670state qos為0
$mqtt->subscribe($topics, 0);

while($mqtt->proc()){

}
//死循環(huán)監(jiān)聽


$mqtt->close();

function procmsg($topic, $msg){ //信息回調(diào)函數(shù) 打印信息
        echo "Msg Recieved: " . date("r") . "
";
        echo "Topic: {$topic}

";
        echo "	$msg

";
        $xxx = json_decode($msg);
        var_dump($xxxxxx->aa);
        die;
}

這是php實(shí)現(xiàn)方法,如果用php做發(fā)送端還是不錯(cuò)的.但是

我被這個(gè)圖片打擊了,區(qū)塊鏈應(yīng)用還真提莫的是js寫起來跟簡單;

我最終寫出的mqtt api 使用的是node;為什么?

node.js實(shí)現(xiàn)mqtt

GPU云服務(wù)器云服務(wù)器手機(jī)發(fā)送消息到服務(wù)器 php 發(fā)送延遲消息 php給進(jìn)程發(fā)送消息 php實(shí)現(xiàn)發(fā)布信息

文章版權(quán)歸作者所有，未經(jīng)允許請(qǐng)勿轉(zhuǎn)載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。

轉(zhuǎn)載請(qǐng)注明本文地址：http://specialneedsforspecialkids.com/yun/28481.html

發(fā)表評(píng)論

登陸后可評(píng)論

0條評(píng)論

eechen

男|高級(jí)講師

我要關(guān)注我要私信

TA的文章

五、C++11新特性: 指針空值--nullptr及nullptr_t

閱讀 1614·2021-11-16 11:45
美國高防服務(wù)器怎么樣?美國高防服務(wù)器的注意事項(xiàng)

閱讀 2544·2021-09-29 09:48
Gcore(gcorelabs)阿什本(弗吉尼亞州)VPS簡單測試

閱讀 3269·2021-09-07 10:26
搬瓦工，DC3機(jī)房CN2 VPS匯總，免費(fèi)自動(dòng)備份+快照，G口帶寬

閱讀 1840·2021-08-16 10:50
純css3漂亮的checkbox和radio美化效果插件Pretty.css插件

閱讀 1866·2019-08-30 15:44
CSS學(xué)習(xí)筆記(八) 頁面布局之三欄-固定寬度布局

閱讀 2698·2019-08-28 18:03
MyBatis插件使用--通用Mapper

閱讀 1898·2019-08-27 10:54
如何讓你的 GitHub 項(xiàng)目表面上更專業(yè)

閱讀 1823·2019-08-26 14:01

国产xxxx99真实实拍_久久不雅视频_高清韩国a级特黄毛片_嗯老师别我我受不了了小说

資訊專欄INFORMATION COLUMN

上云采購季！| 2核2G4M爆款云服務(wù)器低至59元/年，更有多臺(tái)、長期優(yōu)惠，快來選購！

php實(shí)現(xiàn)mqtt發(fā)布/發(fā)送消息到主題

相關(guān)文章

MQTT協(xié)議(1)-簡介

MQTT

MQTT協(xié)議介紹

MQTT如何快速助你產(chǎn)品化

發(fā)表評(píng)論

0條評(píng)論

eechen

男|高級(jí)講師

TA的文章

五、C++11新特性: 指針空值--nullptr及nullptr_t

美國高防服務(wù)器怎么樣?美國高防服務(wù)器的注意事項(xiàng)

Gcore(gcorelabs)阿什本(弗吉尼亞州)VPS簡單測試

搬瓦工，DC3機(jī)房CN2 VPS匯總，免費(fèi)自動(dòng)備份+快照，G口帶寬

純css3漂亮的checkbox和radio美化效果插件Pretty.css插件

CSS學(xué)習(xí)筆記(八) 頁面布局之三欄-固定寬度布局

MyBatis插件使用--通用Mapper

如何讓你的 GitHub 項(xiàng)目表面上更專業(yè)

最新活動(dòng)

資訊專欄INFORMATION COLUMN

上云采購季！| 2核2G4M爆款云服務(wù)器低至59元/年，更有多臺(tái)、長期優(yōu)惠，快來選購！

php實(shí)現(xiàn)mqtt發(fā)布/發(fā)送 消息到主題

相關(guān)文章

發(fā)表評(píng)論

0條評(píng)論

男|高級(jí)講師

TA的文章

最新活動(dòng)

上云采購季！| 2核2G4M爆款云服務(wù)器低至59元/年，更有多臺(tái)、長期優(yōu)惠，快來選購！

php實(shí)現(xiàn)mqtt發(fā)布/發(fā)送消息到主題