CSAPP=并发编程

<?php// ============ CSAPP 第12章 并发编程 - PHP 大白话演示 ============// I/O多路复用用 stream_select 真做; 线程/信号量用逻辑模拟讲清语义declare(strict_types=1);functionT($n,$t){echo"\n===== {$n}. {$t} =====\n";}// 并发三大流派总览echo"并发编程三大流派:\n";echo" 1) 多进程 : 每客户端fork一个进程. 隔离好, 但开销大, 共享数据难\n";echo" 2) I/O多路复用: 单进程用select/epoll盯多个fd. 高效, 但逻辑复杂(状态机)\n";echo" 3) 多线程 : 单进程多线程. 共享内存方便, 但要小心同步\n";// 287 基于进程的并发T(287,"基于进程的并发");echo"服务器accept后fork: 子进程处理这个客户端, 父进程继续accept下一个\n";echo"C关键点:\n";echo" - 父进程必须close已连接socket, 子进程必须close监听socket (否则fd泄漏)\n";echo" - 必须回收僵尸子进程: SIGCHLD handler里 waitpid(-1,&s,WNOHANG)循环\n";echo"优点: 进程隔离, 一个崩了不影响别的; 共享文件表自然\n";echo"缺点: 进程开销大(fork~微秒级), 进程间共享状态要IPC\n";echo" -> 第8章的fork+第11章的accept = 经典并发服务器\n";// 288 I/O多路复用: 真做! 单进程同时服务多个客户端T(288,"I/O多路复用 - 真起一个并发echo服务器");echo"核心: select/poll/epoll 同时监视多个fd, 哪个就绪处理哪个, 单线程搞定并发\n\n";$port=9100+(getmypid()%800);$addr="127.0.0.1:$port";$server=stream_socket_server("tcp://$addr",$errno,$errstr);stream_set_blocking($server,false);if(!$server){echo"起服务器失败\n";}else{echo"并发服务器 listen @ $addr (非阻塞)\n";// 起3个客户端子进程, 几乎同时连入$clients=[];for($i=1;$i<=3;$i++){$code='usleep('.($i*20000).');$fp=stream_socket_client("tcp://'.$addr.'",$e,$s,3);if($fp){for($k=1;$k<=2;$k++){fwrite($fp,"client'.$i.'-msg$k\n");echotrim(fread($fp,256))."\n";usleep(10000);}fwrite($fp,"BYE\n");fclose($fp);}';$clients[$i]=proc_open([PHP_BINARY,'-r',$code],[1=>['pipe','w'],2=>['pipe','w']],$pipes);$clientPipes[$i]=$pipes;}// ===== 事件循环: 这就是I/O多路复用的核心 =====$conns=[];// 已连接客户端 fd 池$served=0;$activeClients=0;$totalMsgs=0;$startTime=time();while(time()-$startTime<5){// 构造要监视的读集合: 监听socket + 所有已连接socket$readSet=$conns;$readSet[]=$server;$write=null;$except=null;// select: 阻塞直到任意fd就绪 (有超时)$ready=@stream_select($readSet,$write,$except,1);if($ready===false||$ready===0){if(empty($conns)&&$served>=3)break;// 全部处理完continue;}foreach($readSet as $fd){if($fd===$server){// 监听socket就绪 = 有新连接$conn=@stream_socket_accept($server,0);if($conn){stream_set_blocking($conn,false);$conns[]=$conn;$served++;$activeClients++;$peer=stream_socket_get_name($conn,true);echo" [事件] 新连接 #$served 来自 $peer (当前活跃: $activeClients)\n";}}else{// 已连接socket就绪 = 客户端发来数据$key=array_search($fd,$conns,true);$line=@fgets($fd);if($line===false||$line===''){// 客户端断开fclose($fd);unset($conns[$key]);$activeClients--;continue;}$line=rtrim($line);if($line==='BYE'){fclose($fd);unset($conns[$key]);$activeClients--;echo" [事件] 一个客户端告别 (剩余活跃: $activeClients)\n";continue;}$totalMsgs++;fwrite($fd,"ECHO[$line]\n");// 回显}}$conns=array_values($conns);}echo"事件循环结束: 单进程服务了 $served 个客户端, 共 $totalMsgs 条消息\n";foreach($clients as $i=>$proc){$out=stream_get_contents($clientPipes[$i][1]);fclose($clientPipes[$i][1]);fclose($clientPipes[$i][2]);proc_close($proc);echo" 客户端$i 收到: ".str_replace("\n"," | ",trim($out))."\n";}fclose($server);echo"★ 关键: 始终只有1个进程1个线程, 却同时服务了3个客户端!\n";echo" 这就是 nginx / redis / Node.js 的并发模型 (epoll事件循环)\n";}// 289-292 基于线程的并发T(289,"基于线程的并发");echo"线程 = 进程内的多个执行流, 共享同一地址空间(代码/数据/堆)\n";echo"比进程轻: 创建快, 切换快, 共享数据直接读写(不用IPC)\n";echo"比进程险: 共享数据要同步, 一个线程崩可能拖垮整个进程\n";T(290,"线程执行模型");echo"一个进程 = 主线程 + 若干对等线程(peer threads)\n";echo"每个线程有自己的: 线程ID(tid) / 栈 / 栈指针 / PC / 通用寄存器\n";echo"所有线程共享: 代码 / 全局数据 / 堆 / 打开的文件\n";echo"对等关系: 任何线程都能kill/join其它线程, 没有严格父子树(不像进程)\n";T(291,"Posix线程 Pthreads");echo"标准C线程API:\n";echo" pthread_create(&tid, NULL, routine, arg) 创建\n";echo" pthread_join(tid, &retval) 等待并回收\n";echo" pthread_self() 获取自己tid\n";echo" pthread_detach(tid) 分离(自动回收,不用join)\n";echo" pthread_exit(retval) 线程退出\n";echo"PHP没内置线程, 用 parallel 扩展 或 多进程替代\n";T(292,"创建与回收线程 - 用并行进程模拟");// PHP无线程, 用并行子进程演示"多个执行流同时跑"$workers=[];$t0=microtime(true);for($i=1;$i<=4;$i++){$code='usleep(50000); echo "worker'.$i.' 算出 "'.'.('.$i.'*'.$i.');';$workers[$i]=proc_open([PHP_BINARY,'-r',$code],[1=>['pipe','w']],$p);$wp[$i]=$p;}echo"主线程创建了4个worker, 它们'同时'运行:\n";foreach($workers as $i=>$proc){// join: 等每个回收$out=stream_get_contents($wp[$i][1]);fclose($wp[$i][1]);proc_close($proc);echo" join worker$i -> $out\n";}printf("4个worker并行总耗时 %.0fms (若串行需200ms)\n",(microtime(true)-$t0)*1000);// 293 共享变量T(293,"共享变量");echo"全局变量、堆上对象 -> 所有线程可见可改 -> 这是线程通信的方式, 也是bug之源\n";echo"局部变量在各自栈上 -> 通常私有 (但传地址出去就可能被共享)\n";echo"判断一个变量是否共享: 看它是否被多个线程引用到\n";// 294-295 信号量与互斥锁: 逻辑模拟T(294,"信号量同步");echo"信号量 s = 一个非负整数, 两个原子操作:\n";echo" P(s): 等到 s>0, 然后 s-- (申请资源)\n";echo" V(s): s++ (释放资源), 可能唤醒等待者\n";echo"原子性是关键: P/V不可被打断, 否则计数会错\n\n";// 用类模拟信号量语义(单进程内顺序演示)class Semaphore{privateint$count;private array $waitLog=[];public function__construct(int$init){$this->count=$init;}public functionP(string $who):bool{// 申请if($this->count>0){$this->count--;echo" $who P() 成功, 剩余资源=$this->count\n";returntrue;}echo" $who P() 阻塞 (资源=0, 排队等待)\n";$this->waitLog[]=$who;returnfalse;}public functionV(string $who):void{// 释放$this->count++;$wake=$this->waitLog?" -> 唤醒 ".array_shift($this->waitLog):"";echo" $who V() 释放, 资源=$this->count$wake\n";}public functionval():int{return$this->count;}}T(295,"互斥锁 mutex = 二元信号量(初值1)");echo"保护临界区: 同一时刻只有一个线程能进\n";$mutex=newSemaphore(1);// 初值1 = 锁echo"两个线程争抢计数器 cnt++ (临界区):\n";$mutex->P("线程A");// A拿锁echo" 线程A 进入临界区, cnt: 0 -> 1\n";$mutex->P("线程B");// B想拿锁, 阻塞$mutex->V("线程A");// A放锁, 唤醒Becho" 线程B 进入临界区, cnt: 1 -> 2\n";$mutex->V("线程B");echo"结果 cnt=2 正确. 若无锁, 两个 cnt++ 可能交错 -> 丢更新, 结果=1\n";// 296 生产者-消费者: 真实演示有界缓冲T(296,"生产者-消费者 (有界缓冲区)");echo"三个信号量协作:\n";echo" mutex=1 : 保护缓冲区互斥访问\n";echo" slots=N : 还有几个空位 (生产者P它)\n";echo" items=0 : 已有几个产品 (消费者P它)\n\n";class BoundedBuffer{private array $buf=[];privateint$cap;private Semaphore $mutex,$slots,$items;public function__construct(int$n){$this->cap=$n;$this->mutex=newSemaphore(1);$this->slots=newSemaphore($n);// 初始n个空位$this->items=newSemaphore(0);// 初始0个产品}public functionproduce($x):void{// P(slots): 等空位; P(mutex): 锁; 放; V(mutex); V(items): 产品+1if($this->slots->val()<=0){echo" 生产者: 缓冲满, 等待消费\n";return;}$this->slots->P('生产者');$this->mutex->P('生产者');$this->buf[]=$x;echo" 生产 $x, 缓冲区: [".implode(',',$this->buf)."]\n";$this->mutex->V('生产者');$this->items->V('生产者');}public functionconsume():mixed{if($this->items->val()<=0){echo" 消费者: 缓冲空, 等待生产\n";returnnull;}$this->items->P('消费者');$this->mutex->P('消费者');$x=array_shift($this->buf);echo" 消费 $x, 缓冲区: [".implode(',',$this->buf)."]\n";$this->mutex->V('消费者');$this->slots->V('消费者');return$x;}}$bb=newBoundedBuffer(3);echo"容量3的缓冲区, 生产者生产5个, 消费者穿插消费:\n";$bb->produce('A');$bb->produce('B');$bb->consume();$bb->produce('C');$bb->produce('D');$bb->produce('E');$bb->produce('F');// 此时满, 触发等待$bb->consume();$bb->consume();// 297 线程池/预线程化T(297,"预线程化 (线程池)");echo"问题: 每来一个连接create一个线程, 创建/销毁开销大\n";echo"方案: 启动时预创建一批worker线程, 主线程accept后把连接放队列\n";echo" worker从队列取连接处理, 处理完不退出, 回去取下一个\n";echo" 主线程 --accept--> [连接队列] <--取-- worker池(N个线程)\n";echo"现代等价: nginx worker进程 / PHP-FPM进程池 / Java线程池 / Go的goroutine调度\n";echo"队列本身用 生产者-消费者 模型保护 (主线程生产, worker消费)\n";// 298 线程安全T(298,"线程安全");echo"线程安全函数: 被多个线程同时调用也能正确工作\n";echo"四类不安全函数:\n";echo" 1) 不保护共享变量 -> 加锁修复\n";echo" 2) 跨调用保持状态(如rand种子) -> 改成传入状态\n";echo" 3) 返回指向静态变量的指针(如ctime/gethostbyname) -> 用可重入版_r\n";echo" 4) 调用了线程不安全的函数 -> 传染, 整个链不安全\n";// 299 可重入性T(299,"可重入性 reentrant");echo"可重入函数 = 不引用任何共享数据 -> 天然线程安全(且更强)\n";echo" 显式可重入: 完全不碰静态/全局, 只用参数和局部变量\n";echo" 例: strtok(不可重入,有静态状态) vs strtok_r(可重入,状态由调用者传入)\n";echo" rand vs rand_r, gethostbyname vs getaddrinfo(可重入,所以推荐)\n";echo"关系: 可重入 ⊂ 线程安全 (可重入是线程安全的真子集)\n";// 300 竞争与死锁T(300,"竞争 race 与 死锁 deadlock");echo"【竞争】程序正确性依赖线程到达某点的相对顺序\n";echo" 经典坑: 主线程for循环里 pthread_create(&tid, ..., &i), 线程读*i\n";echo" -> 主线程的i可能已经变了 -> 传值而非传地址才对\n\n";echo"【死锁】一组线程互相等待对方持有的资源, 都卡死\n";// 死锁演示: 两个锁, 两个线程以相反顺序获取echo" 线程1: lock(A) 然后想 lock(B)\n";echo" 线程2: lock(B) 然后想 lock(A)\n";echo" 时序: T1锁A, T2锁B, T1等B(被T2占), T2等A(被T1占) -> 死锁!\n\n";echo"死锁四条件(全满足才死锁):\n";foreach(['互斥'=>'资源一次只能一个线程占','持有等待'=>'占着一个还要等另一个','不可抢占'=>'不能强夺别人的资源','循环等待'=>'形成等待环 T1->T2->T1',]as $k=>$v)echo" - $k: $v\n";echo"破解: 最常用'按固定全局顺序加锁'(破坏循环等待)\n";echo" 两个线程都先lock(A)再lock(B) -> 永不成环 -> 无死锁\n";echo"\n----- 全书总结 -----\n";echo"并发是现代系统的核心: 多核要并行, I/O要重叠, 服务器要同时服务海量用户\n";echo"三种模型各有取舍:\n";echo" 进程: 隔离最好,开销最大 (Chrome每标签一进程)\n";echo" I/O多路复用: 单线程高并发 (nginx/redis/Node)\n";echo" 线程: 共享方便,同步最难 (传统Java/C++服务器)\n";echo"现代趋势: 协程/async (Go goroutine, Rust async, PHP Swoole/Fiber) -- 兼顾三者优点\n";echo"\n========== 第12章演示完毕 / CSAPP全书完结 ==========\n";