LOVOL

php socket实现mqtt发布/发送消息到主题

Mon, 19 Aug 2024 10:13:31 +0800

php想要实现mqtt需要使用到php中的socket函数;

此次使用的是网上开源mqtt案例:其中使用的是 stream_socket_xxxx 系列函数

大概意思是:

正如你所指出的，'stream'是PHP核心（内置的，始终可用），而'套接字'是很少包含的扩展的一部分。除此之外，它们几乎完全相同。您可以同时使用TCP和UDP两种流，也可以使用阻塞和非阻塞模式，这些模式涵盖了所有用例的99％。
我能想到的唯一常见的例外是ICMP。例如，'ping'。但是，看起来目前还没有一种安全的方式来从PHP执行ICMP。这种调用需要通过套接字扩展来实现SOCK_RAW，这需要执行“root”权限。此外，并非所有路由器都会在TCP，UDP和ICMP之外路由其他数据包类型。这限制了套接字扩展的实用性。

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 \n");
            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\n";
        }else{
            error_log(sprintf("Connection failed! (Error: 0x%02x 0x%02x)\n",
                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\n";
    }
 
    /* 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\n";
    }
 
    /* 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\n";
                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\n";
 
                $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\n";
 
                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\n";
                $this->ping();
            }
 
 
            if($this->timesinceping<(time()-($this->keepalive*2))){
                if($this->debug) echo "not seen a package in a while, disconnecting\n";
                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 \n",$j,$num,$num,$chr);
        }
    }}

实现部分

发送到主题

 // 发送给订阅号信息,创建socket,无sam队列
 $server = "127.0.0.1";     // 服务代理地址(mqtt服务端地址)
 $port = 1883;                     // 通信端口
 $username = "";                   // 用户名(如果需要)
 $password = "";                   // 密码(如果需要
 $client_id = "clientx9293670xxctr"; // 设置你的连接客户端id
 $mqtt = new Mqtt($server, $port, $client_id); //实例化MQTT类
 if ($mqtt->connect(true, NULL, $username, $password)) { 
    //如果创建链接成功
     $mqtt->publish("xxx3809293670ctr", "setr=3xxxxxxxxx", 0); 
     // 发送到 xxx3809293670ctr 的主题 一个信息 内容为 setr=3xxxxxxxxx Qos 为 0 
     $mqtt->close();    //发送后关闭链接
 } else {
     echo "Time out!\n"; 
 }

订阅主题

/*// 订阅信息,接收一个信息后退出
$server = "127.0.0.1";     // 服务代理地址(mqtt服务端地址)
$port = 1883;                     // 通信端口
$username = "";                   // 用户名(如果需要)
$password = "";                   // 密码(如果需要
$client_id = "clientx9293670xxctr"; // 设置你的连接客户端id
 
$mqtt = new Mqtt($server, $port, $client_id);
 
if(!$mqtt->connect(true, NULL, $username, $password)) { //链接不成功再重复执行监听连接
    exit(1);
}
 
$topics['SN69143809293670state'] = array("qos" => 0, "function" => "procmsg");
// 订阅主题为 SN69143809293670state qos为0
$mqtt->subscribe($topics, 0);
 
while($mqtt->proc()){
 
}
//死循环监听
 
 
$mqtt->close();
 
function procmsg($topic, $msg){ //信息回调函数 打印信息
        echo "Msg Recieved: " . date("r") . "\n";
        echo "Topic: {$topic}\n\n";
        echo "\t$msg\n\n";
        $xxx = json_decode($msg);
        var_dump($xxxxxx->aa);
        die;
}

转发自phpMQTT详解以及处理使用过程中内存耗死问题 · php开发笔记 · 看云 (kancloud.cn)

phpMQTT-master.zip

asterisk 中文语音包

Sat, 25 May 2024 14:14:28 +0800

asterisk 中文语音包下载

在线病毒检测

Thu, 21 Mar 2024 17:38:53 +0800

VirSCAN：https://www.virscan.org/language/zh-cn/

腾讯哈勃分析系统：https://habo.qq.com

魔盾安全分析：https://www.maldun.com/submit/submit_file/

微步在线云沙箱：https://s.threatbook.cn/

Jotti的恶意软件扫描系统：https://virusscan.jotti.org/

Falcon Sandbox：https://www.hybrid-analysis.com/

VirusTotal：https://www.virustotal.com/gui/home

奇安信文件深度分析平台：https://sandbox.ti.qianxin.com/

ScanVir：http://www.scanvir.com/

以上几个网站使用方法都比较简单，直接打开网站，再选择上传你要分析的文件，上传后，稍等片刻就可以查看文件分析结果了。

平时在各种网站论坛下载一些从未用过的软件，对其安全性未知或不确定时，
可以将相关文件上传到以上几个病毒扫描网站，分析扫描一下，检查一下软件是否有恶意行为，

分享一些基于Tailwindcss的UI库,排名不分先后.

Sat, 09 Mar 2024 15:02:10 +0800

tailwindcss

ajax和PHP后端数据交互

Thu, 23 Dec 2021 10:49:40 +0800

HTML前端代码

<html xmlns="http://www.w3.org/1999/xhtml" xml:lang="en">
<head>
    <meta http-equiv="Content-Type" content="text/html;charset=UTF-8">

    <script  type="text/javascript" src="/luci/js/jquery-1.11.1.min.js"></script>
</head>
<body>
<div id="msg"></div>
<input id="btn" type="button" value="接收测试" />
<br>
<input id="btn-input-imei" type="text"  class="form-control-imei input-md" placeholder="输入IMEI" />
<input id="btn-input-msg" type="text"  class="form-control-msg input-md" placeholder="输入发送hex字符串" />
<button class="btn btn-success btn-md" id="btn-chat">发送回复测试</button>
<div id="send_msg"></div>
<script  type="text/javascript" >
    $(function(){
        $("#btn").bind("click",{btn:$("#btn")},function(evdata){
            $.ajax({
                type:"POST",
                dataType:"json",
                url:"back.php",
                timeout:80000,     //ajax请求超时时间80秒
                data:{IMEI_GET:"80"}, //40秒后无论结果服务器都返回数据
                success:function(data,textStatus){
                    //从服务器得到数据，显示数据并继续查询
                    if(data.success=="1"){
                        $("#msg").append("<br>[有数据]"+data.text);
                        evdata.data.btn.click();
                    }
                    //未从服务器得到数据，继续查询
                    if(data.success=="0"){
                        $("#msg").append("<br>[无数据]");
                        evdata.data.btn.click();
                    }
                },
                //Ajax请求超时，继续查询
                error:function(XMLHttpRequest,textStatus,errorThrown){
                    if(textStatus=="timeout"){
                        $("#msg").append("<br>[超时]");
                        evdata.data.btn.click();
                    }
                }
            });
        });
    });
</script>
<script type="text/javascript">
    $(function(){
     $("#btn-chat").click(function(){
     $.ajax({
                url:'back.php',
                type:'post',
                data:{"IMEI":$("#btn-input-imei").val(),"MSG":$("#btn-input-msg").val()},   //拼装json数组
                // data:$("#fm").serialize(),   //直接从form表单中取出数组
                dataType:"JSON",
                success:function(msg){   
                    if(msg) {
                        // $("p").append("IMEI：" +  msg.IMEI + "<br />" + "数据恢复：" + msg.MSG );
                        $("#send_msg").append("IMEI：" +  msg.IMEI + "<br />" + "数据恢复：" + msg.MSG + "<br>");
                    }
                    else {
                        alert("输入异常!");
                    }
                },
                error:function(){  
                    console.log("ERROR"); 
                }
            });
     });
    });
</script>
    <div id="example">
        <h3 id="example_title">将滚动条（scrollbar）保持在最底部的方法</h3>
        <div id="example_main">
            <script type="text/javascript">
                function add()
                {
                    var now = new Date();
                    var div = document.getElementById('scrolldIV');
                    div.innerHTML = div.innerHTML + 'time_' + now.getTime() + '<br />';
                    div.scrollTop = div.scrollHeight;//主要 需要指定高度度
                }
            </script>
            <span>请点击“插入一行”按钮，插入最新信息，当出现滚动条时，滚动条将自动保持在底部。</span><br />
 
            <div id="scrolldIV" style="overflow:auto; height: 100px; width: 400px; border: 1px solid #999;">
            </div>
            <input type="button" value="插入一行" onclick="add();">
        </div>
    </div>
</body>
</html>

PHP后端代码

server_get.php
<?php
if (isset($_POST['IMEI']) && isset($_POST['MSG'])) {

	function validate($data){
      $data = trim($data);
	   $data = stripslashes($data);
	   $data = htmlspecialchars($data);
	   return $data;
	}

	$IMEI = validate($_POST['IMEI']);
	$MSG = validate($_POST['MSG']);

    $json = array("IMEI"=>$IMEI, "MSG"=>$MSG);   //组合成json数组
	$data = json_encode($json);  //编译数组转化为json数据
    echo $data;  //将json数据传回网页
    
    }
    elseif(isset($_POST['IMEI_GET']))
    {
    set_time_limit(0);//无限请求超时时间        
    $i=0;
    while (true){
        //sleep(1);        
        usleep(500000);//0.5秒        
        $i++;
        //若得到数据则马上返回数据给客服端，并结束本次请求        
        $rand=rand(1,999);
        if($rand<=15){
            $arr=array('success'=>"1",'name'=>'xiaoyu','text'=>$rand);
            echo json_encode($arr);
            exit();
        }
        //服务器($_POST['time']*0.5)秒后告诉客服端无数据        
        if($i==$_POST['IMEI_GET']){
            $arr=array('success'=>"0",'name'=>'xiaoyu','text'=>$rand);
            echo json_encode($arr);
            exit();
        }
    }
    }
    else {
    exit();
    }
    

?>

php和HTML前端后端互相推送数据

Thu, 23 Dec 2021 10:25:35 +0800

Ajax+PHP实现网页实时更新

要想实现网页内容实时更新，我们经常采用的是长连接的方式。即用户浏览器与服务器之间在访问页面的时候建立长连接，在服务器端有新数据的时候，及时的推送至客户端。

示例代码如下：

前端网页部分源码：

<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd">
<html xmlns="http://www.w3.org/1999/xhtml" xml:lang="en">
<head>
<meta http-equiv="Content-Type" content="text/html;charset=UTF-8">
<script type="text/javascript" src="http://s1.hqbcdn.com/??lib/jquery/jquery-1.7.2.min.js"></script>
</head>
<body>
<div id="msg"></div>
<input id="btn" type="button" value="测试" />
<script type="text/javascript" >
$(function(){
$("#btn").bind("click",{btn:$("#btn")},function(evdata){
$.ajax({
type:"POST",
dataType:"json",
url:"data.php",
timeout:80000, //ajax请求超时时间80秒
data:{time:"80"}, //40秒后无论结果服务器都返回数据
success:function(data,textStatus){
//从服务器得到数据，显示数据并继续查询
if(data.success=="1"){
$("#msg").append("<br>[有数据]"+data.text);
evdata.data.btn.click();
}
//未从服务器得到数据，继续查询
if(data.success=="0"){
$("#msg").append("<br>[无数据]");
evdata.data.btn.click();
}
},
//Ajax请求超时，继续查询
error:function(XMLHttpRequest,textStatus,errorThrown){
if(textStatus=="timeout"){
$("#msg").append("<br>[超时]");
evdata.data.btn.click();
}
}
});
});
});
</script>
</body>
</html>

后端响应前端请求部分源码：

<?php
if(emptyempty($_POST['time']))exit();
set_time_limit(0);//无限请求超时时间
$i=0;
while (true){
//sleep(1);
usleep(500000);//0.5秒
$i++;
//若得到数据则马上返回数据给客服端，并结束本次请求
$rand=rand(1,999);
if($rand<=15){
$arr=array('success'=>"1",'name'=>'xiaoyu','text'=>$rand);
echo json_encode($arr);
exit();
}
//服务器($_POST['time']*0.5)秒后告诉客服端无数据
if($i==$_POST['time']){
$arr=array('success'=>"0",'name'=>'xiaoyu','text'=>$rand);
echo json_encode($arr);
exit();
}
}
?>

python接口服务两种方式（被调用、主动推送）

Wed, 18 Aug 2021 10:45:39 +0800

项目中用python开发，要对外接口时，根据场景不同，一般有两种方式：

1）外部调用python接口，返回数据

2）接口主动向外部推送数据

1）外部调用python接口，返回数据

这种方式一般可以通过http接口，定义接口的ip和端口，外部通过ip和端口调用，调用的程序再返回结果。

接口服务代码如下：

import flask
import json
from flask import request
 
'''
flask： web框架，通过flask提供的装饰器@server.route()将普通函数转换为服务
登录接口，需要传url、username、passwd
'''
# 创建一个服务，把当前这个python文件当做一个服务
server = flask.Flask(__name__)
@server.route('/login', methods=['get', 'post'])
def login():
    # 获取通过url请求传参的数据
    usename = request.args.get('name')
    password = request.args.get('pwd')
 
    #这里可以加入核心功能代码
 
    resu = {'code': 200,'result':'success!'}#返回数据
 
    return json.dumps(resu, ensure_ascii=False)  # 将字典转换为json串, json是字符串
     
  
if __name__ == '__main__':
    server.run(debug=True, port=80, host='0.0.0.0')# 指定端口、host设为0.0.0.0代表不管几个网卡，任何ip都可以访问