Socket服务端

PHPSocket.IO，PHP跨平台实时通讯框架PHPSocket.IO是PHP版本的Socket.IO服务端实现，基于workerman开发，用于替换node.js版本Socket.IO服务端。PHPSocket.IO底层采用websocket协议通讯，如果客户端不支持websocket协议， 则会自动采用http长轮询的方式通讯。环境Ubuntu 18Laravel 5.8PHPSocket.IO 1.1安装依赖composer require workerman/phpsocket.iocomposer require guzzlehttp/guzzle启动程序整合到artisan命令中创建文件命令php artisan make:command MsgPush app/Console/Commands/MsgPush.php <?phpnamespace App\Console\Commands;use Illuminate\Console\Command;use Workerman\Worker;use Workerman\Lib\Timer;use PHPSocketIO\SocketIO;class MsgPush extends Command{ protected $signature = 'msg-push {action=start : start | restart | reload(平滑重启) | stop | status | connetions} {--d : deamon or debug}'; protected $description = 'web消息推送服务'; // 全局数组保存uid在线数据 private static $uidConnectionCounter = []; // 广播的在线用户数,一个uid代表一个用户 private static $onlineCount = 0; // 广播的在线页面数,同一个uid可能开启多个页面 private static $onlinePageCount = 0; //PHPSocketIO服务 private static $senderIo = null; public function __construct() { parent::__construct(); } /** * 根据脚本参数开启PHPSocketIO服务 * PHPSocketIO服务的端口是`2120` * 传递数据的端口是`2121` */ public function handle() { global $argv; //启动php脚本所需的命令行参数 $argv[0] = 'MsgPush'; $argv[1] = $this->argument('action'); // start | restart | reload(平滑重启) | stop | status | connetions $argv[2] = $this->option('d') ? '-d' : ''; // 守护进程模式或调试模式启动 // PHPSocketIO服务 self::$senderIo = new SocketIO(2120); // 客户端发起连接事件时，设置连接socket的各种事件回调 self::$senderIo->on('connection', function ($socket) { // 当客户端发来登录事件时触发,$uid目前由页面传值决定,当然也可以根据业务需要由服务端来决定 $socket->on('login', function ($uid) use ($socket) { // 已经登录过了 if (isset($socket->uid)) return; // 更新对应uid的在线数据 $uid = (string)$uid; // 这个uid有self::$uidConnectionCounter[$uid]个socket连接 self::$uidConnectionCounter[$uid] = isset(self::$uidConnectionCounter[$uid]) ? self::$uidConnectionCounter[$uid] + 1 : 1; // 将这个连接加入到uid分组，方便针对uid推送数据 $socket->join($uid); $socket->uid = $uid; // 更新这个socket对应页面的在线数据 self::emitOnlineCount(); }); // 当客户端断开连接是触发（一般是关闭网页或者跳转刷新导致） $socket->on('disconnect', function () use ($socket) { if (!isset($socket->uid)) { return; } // 将uid的在线socket数减一 if (--self::$uidConnectionCounter[$socket->uid] <= 0) { unset(self::$uidConnectionCounter[$socket->uid]); } }); }); // 当self::$senderIo启动后监听一个http端口，通过这个端口可以给任意uid或者所有uid推送数据 self::$senderIo->on('workerStart', function () { // 监听一个http端口 $innerHttpWorker = new Worker('http://0.0.0.0:2121'); // 当http客户端发来数据时触发 $innerHttpWorker->onMessage = function ($httpConnection, $data) { $type = $_REQUEST['type'] ?? ''; $content = htmlspecialchars($_REQUEST['content'] ?? ''); $to = (string)($_REQUEST['to'] ?? ''); // 推送数据的url格式 type=publish&to=uid&content=xxxx switch ($type) { case 'publish': // 有指定uid则向uid所在socket组发送数据 if ($to) { self::$senderIo->to($to)->emit('new_msg', $content); } else { // 否则向所有uid推送数据 self::$senderIo->emit('new_msg', $content); } // http接口返回，如果用户离线socket返回fail if ($to && !isset(self::$uidConnectionCounter[$to])) { return $httpConnection->send('offline'); } else { return $httpConnection->send('ok'); } } return $httpConnection->send('fail'); }; // 执行监听 $innerHttpWorker->listen(); // 一个定时器，定时向所有uid推送当前uid在线数及在线页面数 Timer::add(1, [self::class, 'emitOnlineCount']); });// Worker::$daemonize = true; Worker::runAll(); } /** * 将在线数变化推送给所有登录端 * 须是public方法,可供其它类调用 */ public static function emitOnlineCount() { $newOnlineCount = count(self::$uidConnectionCounter); $newOnlinePageCount = array_sum(self::$uidConnectionCounter); // 只有在客户端在线数变化了才广播，减少不必要的客户端通讯 if ($newOnlineCount != self::$onlineCount || $newOnlinePageCount != self::$onlinePageCount) {// var_dump('emitOnlineCount: ', self::$uidConnectionCounter); //将在线数变化推送给所有登录端 self::$senderIo->emit( 'update_online_count', [ 'onlineCount' => $newOnlineCount, 'onlinePageCount' => $newOnlinePageCount ] ); self::$onlineCount = $newOnlineCount; self::$onlinePageCount = $newOnlinePageCount; } }}启动PHPSocket.Io服务#守护进程模式启动php artisan msg-push start -d#调式模式启动php artisan msg-push startweb页面resources/views/socketio.blade.php ...

[原文地址：https://blog.ti-node.com/blog…]在＜PHP socket初探 — 先从一个简单的socket服务器开始＞中依次讲解了三个逐渐进步的服务器：只能服务于一个客户端的服务器利用fork可以服务于多个客户端的额服务器利用预fork派生进程服务于多个客户端的服务器最后一种服务器的进程模型基本上的大概原理其实跟我们常用的apache是非常相似的．其实这种模型最大的问题在于需要根据实际业务预估进程数量，依旧是需要大量进程来解决问题，可能会出现CPU浪费在进程间切换上，还有可能会出现惊群现象（简单理解就是100个进程在等带客户端连接，来了一个客户端但是所有进程都被唤醒了，但最终只有一个进程为这个客户端服务，其余99个白白折腾），那么，有没有一种解决方案可以使得少量进程服务于多个客户端呢？答案就是在＜PHP socket初探 — 关于IO的一些枯燥理论＞中提到的＂IO多路复用＂．多路是指多个客户端连接socket，复用就是指复用少数几个进程，多路复用本身依然隶属于同步通信方式，只是表现出的结果看起来像异步，这点值得注意．目前多路复用有三种常用的方案，依次是：select，最早的解决方案poll，算是select的升级版epoll，目前的最终解决版，解决c10k问题的功臣今天说的是select，这个东西本身是个Linux系统调用．在Linux中一切皆为文件，socket也不例外，每当Linux打开一个文件系统都会返回一个对应该文件的标记叫做文件描述符．文件描述符是一个非负整数，当文件描述数达到最大的时候，会重新回到小数重新开始（题外话：按照传统，一般情况下标准输入是0，标准输出是1，标准错误是2）．对文件的读写操作就是利用对文件描述符的读写操作．一个进程可以操作的文件描述符的数量是有限制的，不同系统有不同的数量，在linux中，可以通过调整ulimit来调整控制．先通过一个简单的例子说明下select的作用和功能．双11到了，你给少林足球队买了很多很多球鞋，分别有10个快递给你运送，然后你就不断地电话询问这10个快递员，你觉得有点儿累．阿梅很心疼你，于是阿梅就说：“这事儿你不用管了，你去专心练大力金刚腿吧，等任何一个快递到了，我告诉你”．当其中一个快递来了后，阿梅就喊你：＂下来啦，有快递！＂，但是，这个阿梅比较缺心眼，她不告诉你是具体哪双鞋子的快递，只告诉你有快递到了．所以，你只能依次查询一遍所有快递单的状态才能确认是哪个签收了．上面这个例子通过结合术语演绎一遍就是，你就是服务器软件，阿梅就是select，10个快递就是10个客户端（也就是10个连接socket fd）．阿梅负责替你管理着这10个连接socket fd，当其中任何一个fd有反应了也就是可以读数据或可以发送数据了，阿梅（select）就会告诉你有可以读写的fd了，但是阿梅（select）不会告诉你是哪个fd可读写，所以你必须轮循所有fd来看看是哪个fd，是可读还是可写．是时候机械记忆一波儿了：当你启动select后，需要将三组不同的socket fd加入到作为select的参数，传统意义上这种fd的集合就叫做fd_set，三组fd_set依次是可读集合，可写集合，异常集合．三组fd_set由系统内核来维护，每当select监控管理的三个fd_set中有可读或者可写或者异常出现的时候，就会通知调用方．调用方调用select后，调用方就会被select阻塞，等待可读可写等事件的发生．一旦有了可读可写或者异常发生，需要将三个fd_set从内核态全部copy到用户态中，然后调用方通过轮询的方式遍历所有fd，从中取出可读可写或者异常的fd并作出相应操作．如果某次调用方没有理会某个可操作的fd，那么下一次其余fd可操作时，也会再次将上次调用方未处理的fd继续返回给调用方，也就是说去遍历fd的时候，未理会的fd依然是可读可写等状态，一直到调用方理会．上面都是我个人的理解和汇总，有错误可以指出，希望不会误人子弟．下面通过php代码实例来操作一波儿select系统调用．在php中，你可以通过stream_select或者socket_select来操作select系统调用，下面演示socket_select进行代码演示：<?php// BEGIN 创建一个tcp socket服务器$host = ‘0.0.0.0’;$port = 9999;$listen_socket = socket_create( AF_INET, SOCK_STREAM, SOL_TCP );socket_bind( $listen_socket, $host, $port );socket_listen( $listen_socket );// END 创建服务器完毕 // 也将监听socket放入到read fd set中去，因为select也要监听listen_socket上发生事件$client = [ $listen_socket ];// 先暂时只引入读事件，避免有同学晕头$write = [];$exp = [];// 开始进入循环while( true ){ $read = $client; // 当select监听到了fd变化，注意第四个参数为null // 如果写成大于0的整数那么表示将在规定时间内超时 // 如果写成等于0的整数那么表示不断调用select，执行后立马返回，然后继续 // 如果写成null，那么表示select会阻塞一直到监听发生变化 if( socket_select( $read, $write, $exp, null ) > 0 ){ // 判断listen_socket有没有发生变化，如果有就是有客户端发生连接操作了 if( in_array( $listen_socket, $read ) ){ // 将客户端socket加入到client数组中 $client_socket = socket_accept( $listen_socket ); $client[] = $client_socket; // 然后将listen_socket从read中去除掉 $key = array_search( $listen_socket, $read ); unset( $read[ $key ] ); } // 查看去除listen_socket中是否还有client_socket if( count( $read ) > 0 ){ $msg = ‘hello world’; foreach( $read as $socket_item ){ // 从可读取的fd中读取出来数据内容，然后发送给其他客户端 $content = socket_read( $socket_item, 2048 ); // 循环client数组，将内容发送给其余所有客户端 foreach( $client as $client_socket ){ // 因为client数组中包含了 listen_socket 以及当前发送者自己socket，所以需要排除二者 if( $client_socket != $listen_socket && $client_socket != $socket_item ){ socket_write( $client_socket, $content, strlen( $content ) ); } } } } } // 当select没有监听到可操作fd的时候，直接continue进入下一次循环 else { continue; } }将文件保存为server.php，然后执行php server.php运行服务，同时再打开三个终端，执行telnet 127.0.0.1 9999，然后在任何一个telnet终端中输入"I am DOG!"，再看其他两个telnet窗口，是不是感觉很屌？不完全截图图下：还没意识到问题吗？如果我们看到有三个telnet客户端连接服务器并且可以彼此之间发送消息，但是我们只用了一个进程就可以服务三个客户端，如果你愿意，可以开更多的telnet，但是服务器只需要一个进程就可以搞定，这就是IO多路复用diao的地方！最后，我们重点解析一些socket_select函数，我们看下这个函数的原型：int socket_select ( array &$read , array &$write , array &$except , int $tv_sec [, int $tv_usec = 0 ] )值得注意的是$read，$write，$except三个参数前面都有一个&，也就是说这三个参数是引用类型的，是可以被改写内容的．在上面代码案例中，服务器代码第一次执行的时候，我们要把需要监听的所有fd全部放到了read数组中，然而在当系统经历了select后，这个数组的内容就会发生改变，由原来的全部read fds变成了只包含可读的read fds，这也就是为什么声明了一个client数组，然后又声明了一个read数组，然后read = client．如果我们直接将client当作socket_select的参数，那么client数组内容就被修改．假如有5个用户保存在client数组中，只有1个可读，在经过socket_select后client中就只剩下那个可读的fd了，其余4个客户端将会丢失，此时客户端的表现就是连接莫名其妙发生丢失了．[原文地址：https://blog.ti-node.com/blog…] ...

[原文地址：https://blog.ti-node.com/blog…]socket的中文名字叫做套接字，这种东西就是对TCP/IP的“封装”。现实中的网络实际上只有四层而已，从上至下分别是应用层、传输层、网络层、数据链路层。最常用的http协议则是属于应用层的协议，而socket，可以简单粗暴的理解为是传输层的一种东西。如果还是很难理解，那再粗暴地点儿tcp://218.221.11.23:9999，看到没？这就是一个tcp socket。socket赋予了我们操控传输层和网络层的能力，从而得到更强的性能和更高的效率，socket编程是解决高并发网络服务器的最常用解决和成熟的解决方案。任何一名服务器程序员都应当掌握socket编程相关技能。在php中，可以操控socket的函数一共有两套，一套是socket_系列的函数，另一套是stream_系列的函数。socket_是php直接将C语言中的socket抄了过来得到的实现，而stream_系则是php使用流的概念将其进行了一层封装。下面用socket_*系函数简单为这一系列文章开个篇。先来做个最简单socket服务器：<?php$host = ‘0.0.0.0’;$port = 9999;// 创建一个tcp socket$listen_socket = socket_create( AF_INET, SOCK_STREAM, SOL_TCP );// 将socket bind到IP：port上socket_bind( $listen_socket, $host, $port );// 开始监听socketsocket_listen( $listen_socket );// 进入while循环，不用担心死循环死机，因为程序将会阻塞在下面的socket_accept()函数上while( true ){ // 此处将会阻塞住，一直到有客户端来连接服务器。阻塞状态的进程是不会占据CPU的 // 所以你不用担心while循环会将机器拖垮，不会的 $connection_socket = socket_accept( $listen_socket ); // 向客户端发送一个helloworld $msg = “helloworldrn”; socket_write( $connection_socket, $msg, strlen( $msg ) ); socket_close( $connection_socket );}socket_close( $listen_socket );将文件保存为server.php，然后执行php server.php运行起来。客户端我们使用telnet就可以了，打开另外一个终端执行telnet 127.0.0.1 9999按下回车即可。运行结果如下：简单解析一下上述代码来说明一下tcp socket服务器的流程：1.首先，根据协议族（或地址族）、套接字类型以及具体的的某个协议来创建一个socket。2.第二，将上一步创建好的socket绑定（bind）到一个ip:port上。3.第三，开启监听linten。4.第四，使服务器代码进入无限循环不退出，当没有客户端连接时，程序阻塞在accept上，有连接进来时才会往下执行，然后再次循环下去，为客户端提供持久服务。上面这个案例中，有两个很大的缺陷：1.一次只可以为一个客户端提供服务，如果正在为第一个客户端发送helloworld期间有第二个客户端来连接，那么第二个客户端就必须要等待片刻才行。2.很容易受到攻击，造成拒绝服务。分析了上述问题后，又联想到了前面说的多进程，那我们可以在accpet到一个请求后就fork一个子进程来处理这个客户端的请求，这样当accept了第二个客户端后再fork一个子进程来处理第二个客户端的请求，这样问题不就解决了吗？OK！撸一把代码演示一下：<?php$host = ‘0.0.0.0’;$port = 9999;// 创建一个tcp socket$listen_socket = socket_create( AF_INET, SOCK_STREAM, SOL_TCP );// 将socket bind到IP：port上socket_bind( $listen_socket, $host, $port );// 开始监听socketsocket_listen( $listen_socket );// 进入while循环，不用担心死循环死机，因为程序将会阻塞在下面的socket_accept()函数上while( true ){ // 此处将会阻塞住，一直到有客户端来连接服务器。阻塞状态的进程是不会占据CPU的 // 所以你不用担心while循环会将机器拖垮，不会的 $connection_socket = socket_accept( $listen_socket ); // 当accept了新的客户端连接后，就fork出一个子进程专门处理 $pid = pcntl_fork(); // 在子进程中处理当前连接的请求业务 if( 0 == $pid ){ // 向客户端发送一个helloworld $msg = “helloworldrn”; socket_write( $connection_socket, $msg, strlen( $msg ) ); // 休眠5秒钟，可以用来观察时候可以同时为多个客户端提供服务 echo time().’ : a new client’.PHP_EOL; sleep( 5 ); socket_close( $connection_socket ); exit; }}socket_close( $listen_socket );将代码保存为server.php，然后执行php server.php，客户端依然使用telnet 127.0.0.1 9999，只不过这次我们开启两个终端来执行telnet。重点观察当第一个客户端连接上去后，第二个客户端时候也可以连接上去。运行结果如下：通过接受到客户端请求的时间戳可以看到现在服务器可以同时为N个客户端服务的。但是，接着想，如果先后有1万个客户端来请求呢？这个时候服务器会fork出1万个子进程来处理每个客户端连接，这是会死人的。fork本身就是一个很浪费系统资源的系统调用，1W次fork足以让系统崩溃，即便当下系统承受住了1W次fork，那么fork出来的这1W个子进程也够系统内存喝一壶了，最后是好不容易费劲fork出来的子进程在处理完毕当前客户端后又被关闭了，下次请求还要重新fork，这本身就是一种浪费，不符合社会主义主流价值观。如果是有人恶意攻击，那么系统fork的数量还会呈直线上涨一直到系统崩溃。所以，我们就再次提出增进型解决方案。我们可以预估一下业务量，然后在服务启动的时候就fork出固定数量的子进程，每个子进程处于无限循环中并阻塞在accept上，当有客户端连接挤进来就处理客户请求，当处理完成后仅仅关闭连接但本身并不销毁，而是继续等待下一个客户端的请求。这样，不仅避免了进程反复fork销毁巨大资源浪费，而且通过固定数量的子进程来保护系统不会因无限fork而崩溃。<?php$host = ‘0.0.0.0’;$port = 9999;// 创建一个tcp socket$listen_socket = socket_create( AF_INET, SOCK_STREAM, SOL_TCP );// 将socket bind到IP：port上socket_bind( $listen_socket, $host, $port );// 开始监听socketsocket_listen( $listen_socket );// 给主进程换个名字cli_set_process_title( ‘phpserver master process’ );// 按照数量fork出固定个数子进程for( $i = 1; $i <= 10; $i++ ){ $pid = pcntl_fork(); if( 0 == $pid ){ cli_set_process_title( ‘phpserver worker process’ ); while( true ){ $conn_socket = socket_accept( $listen_socket ); $msg = “helloworldrn”; socket_write( $conn_socket, $msg, strlen( $msg ) ); socket_close( $conn_socket ); } }}// 主进程不可以退出，代码演示比较粗暴，为了不保证退出直接走while循环，休眠一秒钟// 实际上，主进程真正该做的应该是收集子进程pid，监控各个子进程的状态等等while( true ){ sleep( 1 );}socket_close( $connection_socket );将文件保存为server.php后php server.php执行，然后再用ps -ef | grep phpserver | grep -v grep来看下服务器进程状态：可以看到master进程存在，除此之外还有10个子进程处于等待服务状态，再同一个时刻可以同时为10个客户端提供服务。我们通过telnet 127.0.0.1 9999来尝试一下，运行结果如下图：好啦，php新的征程系列就先通过一个简单的入门开始啦！下篇将会讲述一些比较深刻的理论基础知识。[原文地址：https://blog.ti-node.com/blog…] ...