发表新帖

swoole如何客户端间点对点的通信

我现在需要客户端点对点的通信。

我的思路是这样的，连接后以接受第一句话作为客户端的名字，并且每个客户端的第一句话都是唯一的，作为鉴别。

作为接受第一句话，以$task_id==0来做的。

但是客户端掉线了，再连接，fd变了，但是接受的任务$task_id还是原来的，并没有清零。

请问下$server->on('finish',function($serv,$task_id, $data) {});中的$task_id该如何修改呢？谢谢！

下面是我的代码：
<?php 
/****
简单的TCP通信task
****/

$server = new swoole_server('120.25.60.18',9527);

$server->on('connect',function($serv, $fd, $from_id){
    echo "Connected\n";
    echo "fd = {$fd},from_id={$from_id}\n";
    $task_id =0;
    $param = array(
        'fd' => $fd,
        'data' => $data,
        'task_id' => $task_id          
    );    
    $serv->task( json_encode( $param ) );
    echo "Continue Handle Worker\n"; 
});

$server->on('receive',function (swoole_server $serv, $fd, $from_id, $data){
    echo "Get Message From Client from_id:{$from_id} fd:{$fd}:data:{$data}\n";
    // send a task to task worker.
    $param = array(
        'fd' => $fd,
        'data' => $data        
    );    
    $serv->task( json_encode( $param ) );
    echo "Continue Handle Worker\n";
});

$server->on('close',function($serv, $fd, $from_id ){
    echo "Closed\n";
    //删掉数据库
    $data = "close";
    echo "Get Message From Client closed{$fd}:{$data}\n";    
   
    // send a task to task worker.
    $param = array(
        'fd' => $fd,
        'data' => $data               
    );
    $serv->task( json_encode( $param ) );
    echo "Continue Handle Worker\n";
});

$server->on('task',function($serv,$task_id,$from_id, $data) {
	echo "This Task {$task_id} from Worker {$from_id}\n";
	echo "Data: {$data}\n";
	/*
	for($i = 0 ; $i < 10 ; $i ++ ) {
		sleep(1);
		echo "Taks {$task_id} Handle {$i} times...\n";
	}*/

$fd = json_decode( $data , true )['fd'];
	$data = json_decode( $data , true )['data'];
    //$task_id = json_decode( $data , true )['task_id'];
    //
    if($data == 'close')
    {
        client_delete_name($task_id,$fd);
    }
    //
    else
    {
        client_name_v($task_id,$fd,$data);//连接数据库    
    }
	echo "fd: {$fd}\n";
	var_dump($fd);
	//$serv->send( $fd , "@{$data}");
	return "Task {$task_id}'s result";
});

//通过onTask函数return一个字符串将执行结果返回给Worker进程。Worker进程将通过onFinish回调函数接收这个处理结果。
$server->on('finish',function($serv,$task_id, $data) {
    echo "Task {$task_id} finish\n";
    echo "Result: {$data}\n";
});

//连接数据库，客户端第一次对话的内容，作为客户端的名字插入到数据库中    
function client_name_v($task_id,$fd,$data) {
    //连接数据库
    $link = mysql_connect("127.0.0.1","root","cmd.123")or die("不能连接到数据库服务器！".mysql_error());
    if($link){
        echo "数据库连接正常\n";
    }
    echo "client_name_v-data:{$data}\n";
    //设置客户端字符集和连接字符集
    mysql_query('set names utf8');
    //切换数据库
    mysql_select_db("car_lock_client_infor_db");
    if($task_id == 1)
    {
        echo "插入数据库\n";
        //组合Sql语句
        //$sql="insert into car_lock_client_infor_table values({$fd},'{$data}')";
        //把Sql语句发送到Mysql数据器
        $result=mysql_query("insert into car_lock_client_infor_table values({$fd},'{$data}')");
        //如果插入失败，则更新数据库
        if($result == false)
        {
            echo "插入失败，则更新数据库\n";
           //$sql="update car_lock_client_infor_table set client_name = '{$data}' where fd = {$fd}"; 
           $result=mysql_query("update car_lock_client_infor_table set client_name = '{$data}' where fd = {$fd}");
           var_dump($result);
        }
        else
        {
            echo "插入成功\n";  
        }

}
    else if($task_id >= 2)
    {
        echo "查询数据库\n";
    }

//删掉某行数据库
    //$result=mysql_query("delete from car_lock_client_infor_table where fd = 1");
    //var_dump($result);

/*
	 query.prepare("update car_flow_table set car_free =:number where park_id = :id");
				query.bindValue(":id",park_id);
				query.bindValue(":number",park_free);
				if(!query.exec())
				{
					qDebug()<<query.lastError();
				}

//插入数据库
    $result=mysql_query("insert into car_lock_client_infor_table values(1,'asd')");
    var_dump($result);
    */
    /*
    //查找某数值
    //组合Sql语句
    $sql="select * from car_lock_client_infor_table";
    //把Sql语句发送到Mysql数据器
    $result=mysql_query($sql);
    var_dump($result);
    while($row=mysql_fetch_array($result)){         
        if($row['client_name'] == 'cmd2') 
        {             
            echo "find data:{$row['fd']}\n";  
        }    
            
        print_r($row);        
    } 
    */

}

}

$server->set(array(
    'reactor_num' => 1,//通过此参数来调节poll线程的数量，以充分利用多核.默认设置为CPU核数
    'worker_num' => 3,//建议开启的worker进程数为cpu核数的1-4倍。
    'daemonize' => false,//设置程序进入后台作为守护进程运行
    'max_request' => 10000,//max_request => 2000，此参数表示worker进程在处理完n次请求后结束运行。manager会重新创建一个worker进程。此选项用来防止worker进程内存溢出。
    'dispatch_mode' => 2,//指定数据包分发策略:1 => 轮循模式,2 => 固定模式,3 => 抢占模式   
    'task_worker_num' => 2  
));

$server->start();

发布于7年前 · 2 次浏览 · 来自提问