181jk9090.php

用PHP写的网络通信服务器程序。用在远程数据采集系统

#!/usr/bin/php -q
<?php
// LSA3000系统服务端通信模块
$host="localhost";   
$user="bb2";        
$pw="bb2";          
$db="jiankong";         
$con=mysql_connect($host,$user,$pw)or die("connetction error");
mysql_select_db($db,$con)or die("could not open $db");

// 服务器IP、监听端口
$address = '192.168.101.181';
$port = '9090';

error_reporting(E_ALL);

// Allow the script to hang around waiting for connections. 
set_time_limit(0);

// Turn on implicit output flushing so we see what we're getting  as it comes in. 
ob_implicit_flush();

// tick use required as of PHP 4.3.0
declare(ticks = 1);

// signal handler function
function sig_handler($signo)
{

     switch ($signo) {
         case SIGTERM:
             // handle shutdown tasks
             exit;
             break;
         case SIGHUP:
             // handle restart tasks
             break;
         case SIGUSR1:
             echo "Caught SIGUSR1...\n";
             break;
         default:
             // handle all other signals
     }

}
// listen 
	if (($sock = socket_create(AF_INET, SOCK_STREAM, SOL_TCP)) < 0) {
	    echo "socket_create() failed: reason: " . socket_strerror($sock) . "\n";
	}
	if (($ret = socket_bind($sock, $address, $port)) < 0) {
	    echo "socket_bind() failed: reason: " . socket_strerror($ret) . "\n";
	}
	if (($ret = socket_listen($sock, 5)) < 0) {
	    echo "socket_listen() failed: reason: " . socket_strerror($ret) . "\n";
	}
// 主进程
do {   
	if (($msgsock = socket_accept($sock)) < 0) {
        echo "socket_accept() failed: reason: " . socket_strerror($msgsock) . "\n";
        break;
        }

// fork a child process
	$pid = pcntl_fork();
	if ($pid == -1) {
	     die("could not fork");
	    } else if ($pid == 0 ) {
//  echo "we are the child"; 
// 设置定时器
// 90秒内没收到信号，系统关闭子进程
pcntl_alarm ( 90);

    // get FSU IP
socket_getpeername($msgsock, &$fsuIp);

$M_send = TRUE; //可以发送测量命令
$M_count = 1;   //测量大循环计数初始值
$C_send = TRUE; //可以发送控制命令
$C_count = 1;   //控制大循环计数初始值
$E_send = TRUE; //可以发送扩展命令
$E_count = 1;   //扩展大循环计数初始值
$cutbufcount = 20; // 裁减buf 循环最大次数

$lifetime = 600; // alarm signal life time
$M_countmax = 6 ;//子进程大循环计数器大于此值没收到应答，再次发送测量命令
$C_countmax = 6 ;//子进程大循环计数器大于此值没收到应答，再次发送控制命令
$E_countmax = 6 ;//子进程大循环计数器大于此值没收到应答，再次发送扩展命令


//测量历史数据变量初始化开始
$msinsert = TRUE;//允许插入测量历史数据
$msitem = "";
$ctrlitem = "";
$expitem = "";
$msstart = time(); 

// 相当于最大值最小值数组(20X4)
for ($i=0; $i < 20; $i++) { 
$msvalsum = "msvalsum".$i;
$$msvalsum = 0;
$msvalcount = "msvalcount".$i;
$$msvalcount = 0;
$msminval = "msminval".$i;
$$msminval = 0;
$msmaxval = "msmaxval".$i;
$$msmaxval = 0;
}

//测量历史数据变量初始化结束
$fsuId = "fsu003";

// detatch from the controlling terminal
if (!posix_setsid()) die("could not detach from terminal");

// loop forever performing client tasks 
// 子进程大循环开始
while (1) {
/*	
  //查询数据库，发送测量命令
	// #M=001,002,003,004*2,5#
  if ($M_send) {
	$query = sprintf("SELECT `ItemID` FROM `taskorder` WHERE fsuid='%s' AND attr='m' AND todo='Y' ",
			    mysql_real_escape_string($fsuId)); 
		$result = mysql_query($query) or die("Invalid query: " . mysql_error());
	
		//  Send measure item 
  	if (mysql_num_rows($result) > 0) {		
  			$msitem = "#";
  		    while ($row = mysql_fetch_array($result, MYSQL_ASSOC)) {  
				$msitem .= "M=".substr($row["ItemID"],7).",";
				}
			$msitem = substr($msitem ,0,-1)."#";
	   
	   // query freqency and length from fsuinit		
	   $query = sprintf("SELECT `freq`,`length` FROM `fsuinit` WHERE fsuid='%s' ",
			    mysql_real_escape_string($fsuId)); 
		 $result = mysql_query($query) or die("Invalid query: " . mysql_error()) ;
		 if (mysql_num_rows($result) > 0) {				
		 $row = mysql_fetch_array($result, MYSQL_ASSOC);	
		    $msitem .= $row["freq"].",".$row["length"]."#";
		  } else {
			$msitem .= "#";
			  }
		
    if (false === socket_write($msgsock, $msitem, strlen($msitem))) echo "socket_write() failed\n";
    // 发送成功，先将 todo 改成 N，否则由于大循环周期比收到FSU应答时间短，命令会重复发送
    $query = sprintf("UPDATE `taskorder` SET `todo` ='N' WHERE fsuid='%s' AND attr='m' ",
			    mysql_real_escape_string($fsuId)); 
			 mysql_query($query) or die("Invalid query: " . mysql_error());
		$M_send = FALSE; //收到响应前不可以发送新测量命令
	  }
   }
*/ 
  //查询数据库，发送控制命令
  //#C*001,002,005#
  if ($C_send) {
	$query = sprintf("SELECT `ItemID`  FROM taskorder WHERE FsuID='%s' AND TaskType='c' AND Task='Y'",
			    mysql_real_escape_string($fsuId)); 
	$result = mysql_query($query) or die("Invalid query: " . mysql_error());
		//  Send control item 
  if (mysql_num_rows($result) > 0 ) {
		  $ctrlitem = "#";
  		while ($row = mysql_fetch_array($result, MYSQL_ASSOC)) {  
			$ctrlitem .= "C*".substr($row["ItemID"],7).",";
			}
		$ctrlitem =substr($ctrlitem ,0,-1)."#";
    if (false === socket_write($msgsock, $ctrlitem, strlen($ctrlitem))) echo "socket_write() failed\n";
    // 发送成功，立即将 todo 改成 N，否则由于大循环周期比收到FSU应答时间短，命令会重复发送
    $query = sprintf("UPDATE `taskorder` SET `Task` ='N' WHERE FsuID='%s' AND TaskType='c' ",
		         mysql_real_escape_string($fsuId)); 
	  mysql_query($query) or die("Invalid query: " . mysql_error());
	  $C_send = FALSE; //收到响应前不可以发送新控制命令
    }
   }
/*     
  //查询数据库，发送扩展命令
  // #EXP*192.168.101.12*5632*001*#
  if ($E_send) {
	$query = sprintf("SELECT `ItemID`,`ip`,`port`  FROM taskorder WHERE fsuid='%s' AND attr='e' AND todo='Y'",
			    mysql_real_escape_string($fsuId)); 
		$result = mysql_query($query) or die("Invalid query: " . mysql_error());
		//  Send exp item 
  	if (mysql_num_rows($result) > 0 ) {
		  $expitem = "#EXP*";
  		$row = mysql_fetch_array($result, MYSQL_ASSOC);
				$expitem .= $row["ip"]."*";
				$expitem .= $row["port"]."*";
				$expItemID = $row["ItemID"];
				$expitem .= substr($expItemID,7)."*#";
    	if (false === socket_write($msgsock, $expitem, strlen($expitem))) echo "socket_write() failed\n";
    	// 发送成功，立即将 todo 改成 N，否则由于大循环周期比收到FSU应答时间短，命令会重复发送
    	$query = sprintf("UPDATE `taskorder` SET `todo` ='N' WHERE ItemID='%s' AND attr='e' ",
		            mysql_real_escape_string($expItemID)); 
		  mysql_query($query) or die("Invalid query: " . mysql_error());
      $E_send = FALSE; //收到响应前不可以发送新扩展命令
     }
   }         
*/  
//  接收 FSU 数据，根据TYPE 交各处理模块处理，分隔符：#	
		
   if (false === ($buf = socket_read($msgsock, 1024, PHP_NORMAL_READ))) {
       echo "socket_read() failed: reason: " . socket_strerror($ret) . "\n";
       break;
    }
	
// "#HeartBeat*fsu001#"
   if (substr($buf,0,11) == '#HeartBeat*') {
	  	pcntl_alarm ( 90);// 定时器复位

	  $nowtime = "#TIME=".date("Y-m-d H:i:s")."#";
      $fsuId = substr($buf,11,6); // 获得FSUID
      // echo $fsuId;
// 将FSU IP、时间等存入数据库，供监控程序判断通信链路状态
   $query = sprintf("UPDATE `stationinfo` SET `StationIP` = '%s',`HBTime`= NOW() WHERE `ItemID` = '%s'",
	 					mysql_real_escape_string($fsuIp),
	 					mysql_real_escape_string($fsuId)); 
	 		mysql_query($query) or die("Invalid query: " . mysql_error());  
/*
    $query = sprintf("SELECT  `fsuid`,`fsuip` FROM `fsulist` WHERE `fsuid` = '%s'",
	 					mysql_real_escape_string($fsuId)); 
      $result = mysql_query($query) or die("Invalid query: " . mysql_error());
    //  save IP if FSU use dyna ip address
			if (mysql_num_rows($result) == 0) {
			//insert IP into fsulist
			$query = sprintf("INSERT INTO `fsulist` (`fsuid`, `fsuip`) VALUES('%s', '%s')",
	 					mysql_real_escape_string($fsuId),
	 					mysql_real_escape_string($fsuIp)); 
	 		mysql_query($query) or die("Invalid query: " . mysql_error());  
			} else {
			//update ip data
			$query = sprintf("UPDATE `fsulist` SET `fsuip` = '%s' WHERE `fsuid` = '%s'",
	 					mysql_real_escape_string($fsuIp),
	 					mysql_real_escape_string($fsuId)); 
	 		mysql_query($query) or die("Invalid query: " . mysql_error());  
			}
*/		
	   socket_write($msgsock, $nowtime, strlen($nowtime)); //返回时间
    $buf = substr($buf, 18);//用掉一段，裁掉一段
			 continue;
    }     
//测试$buf为fhdsfh###MOK##COK##HeartBeat*fsu001##VAL*001=1.35*002=1.55*skadgfk##EOK#hsdfhsdf的情况,用掉一段，裁掉一段
//将分隔符加长，判断会更准确
   $aa = 0;

   while (strlen($buf) > 3) {
   $aa++;
   if ($aa > $cutbufcount ) break ;//避免$buf裁减失败，进入死循环  	
   $buf = strstr($buf, '#'); //去掉#之前的多余字符（如果有）
   $buftemp = substr($buf, 1 ); //去掉#
   if (substr($buftemp ,0 , 1 ) == "#") $buf = $buftemp ; //去掉$buf中##开头的第一个#
/*
// "#MOK#" 
   if (substr($buf,0,5) == '#MOK#') {
			 $M_send = TRUE; //可以发送新测量命令
			 $buf = substr($buf, 5);//用掉一段，裁掉一段
			 continue;
    }
   $M_count++;
   if ($M_count > $M_countmax) {
   	 // 如达到最大循环计数没收到#MOK#，重发测量命令
   	 if (false === socket_write($msgsock, $msitem, strlen($msitem))) echo "socket_write() failed\n";
     $M_count = 1 ;//计数器复位
    }
// "#COK#"
   if (substr($buf,0,5) == '#COK#') {
			 $C_send = TRUE; //可以发送新控制命令
			 $buf = substr($buf, 5);//用掉一段，裁掉一段
			 continue;
    }
   $C_count++;
   if ($C_count > $C_countmax) {
   	 // 如达到最大循环计数没收到#COK#，重发控制命令
   	 if (false === socket_write($msgsock, $ctrlitem, strlen($ctrlitem))) echo "socket_write() failed\n";
     $C_count = 1 ;//计数器复位
    }

// "#EOK#"
   if (substr($buf,0,5) == '#EOK#') {
   	 $E_send = TRUE; //可以发送新控制命令
   	 $buf = substr($buf, 5);//用掉一段，裁掉一段
			 continue;
    }
   $E_count++;
   if ($E_count > $E_countmax) {
   	 // 如达到最大循环计数没收到#EOK#，重发扩展命令
   	 if (false === socket_write($msgsock, $expitem, strlen($expitem))) echo "socket_write() failed\n";
     $E_count = 1 ;//计数器复位
    } 
*/

// "#VAL*001=1.35*002=1.55*#"
   if (substr($buf,0,5) == '#VAL*') {
   	  pcntl_alarm ( 90);// 定时器复位
   	  $msperiod = 30; //default 每隔30秒向数据库写入一个测量历史数据
   	                  // 查询 fsuinit表 msperiod字段 获得
      // 初始值 $msinsert = TRUE;
      if($msinsert) $msstart = time();  //临时标记，$msperiod 限定的时间内数据更新唯一标识

      
      $restbuf = substr($buf,5);
	    $measuresum = substr_count($restbuf, '=')  ; //收到的测量值数量

      for ($i=0; $i < $measuresum; $i++) {
	      $fsuitem = $fsuId.'m'.substr($restbuf,0,3);
	      $msval = substr($restbuf,4,(strpos($restbuf,"*")-4));
	      $nowtime = date("Y-m-d H:i:s");
	      $restbuf = substr($restbuf,((strpos($restbuf,"*"))+1));

// 临时使用倍数表开始
	
	      	if ($fsuitem =='fsu050m001') {
	      		$msval = 0.05306*$msval;
	      	}
	      	if ($fsuitem =='fsu050m002') {