利用 Phalcon7 的异步功能实现，完整源码 https://github.com/dreamsxin/cphalcon7/blob/master/examples/async/Socks5Server.php。
1<?php 2 3class Pool 4{ 5 private $channel; 6 private $concurrency; 7 private $count = 0; 8 private $context; 9 10 public function __construct(int $concurrency = 1, int $capacity = 0) 11 { 12 $this->concurrency = max(1, $concurrency); 13 $this->channel = new \Phalcon\Async\Channel($capacity); 14 $this->context = \Phalcon\Async\Context::background(); 15 } 16 17 public function close(?\Throwable $e = null): void 18 { 19 $this->count = \PHP_INT_MAX; 20 $this->channel->close($e); 21 } 22 23 public function submit(callable $work, $socket, ...$args): \Phalcon\Async\Awaitable 24 { 25 if ($this->count < $this->concurrency) { 26 $this->count++; 27 Socks5Server::info('Pool count '.$this->count); 28 \Phalcon\Async\Task::asyncWithContext($this->context, static function (iterable $it) { 29 try { 30 foreach ($it as list ($defer, $context, $work, $socket, $args)) { 31 try { 32 $defer->resolve($context->run($work, $socket, ...$args)); 33 } catch (\Throwable $e) { 34 Socks5Server::err($e->getMessage()); 35 $defer->fail($e); 36 } finally { 37 } 38 } 39 } catch (\Throwable $e) { 40 Socks5Server::err($e->getMessage()); 41 } finally { 42 --$this->count; 43 } 44 }, $this->channel->getIterator()); 45 } 46 47 $this->channel->send([ 48 $defer = new \Phalcon\Async\Deferred(), 49 \Phalcon\Async\Context::current(), 50 $work, 51 $socket, 52 $args 53 ]); 54 55 return $defer->awaitable(); 56 } 57} 58 59class Socks5Server 60{ 61 static public $debug = false; 62 private $server; 63 64 private $host; 65 private $port; 66 67 private $clients; 68 69 public function __construct($host, $port, callable $callback = NULL, int $concurrency = 1, int $capacity = 0) 70 { 71 $this->host = $host; 72 $this->port = $port; 73 $this->callback = $callback; 74 $this->pool = new Pool($concurrency, $capacity); 75 } 76 77 public function start() 78 { 79 $callback = $this->callback; 80 $ws = $this; 81 $worker = static function ($socket) use ($ws, $callback) { 82 $socket->status = Socks5::STATUS_INIT; 83 $socket->is_closing = false; 84 try { 85 $buffer = ''; 86 while (!$socket->is_closing && null !== ($chunk = $socket->read())) { 87 88 $buffer .= $chunk; 89 switch ($socket->status) { 90 case Socks5::STATUS_INIT: 91 /** 92 * 协商版本以及验证方式 93 * +---------+-----------------+------------------+ 94 * |协议版本 |支持的验证式数量 |验证方式 | 95 * +---------+-----------------+------------------+ 96 * |1个字节 |1个字节 |1种式占一个字节 | 97 * +---------+-----------------+------------------+ 98 * |0x05 |0x02 |0x00,0x02 | 99 * +---------+-----------------+------------------+ 100 */ 101 /** 102 * 0x00 无验证需求 103 * 0x01 通用安全服务应用程序接口(GSSAPI) 104 * 0x02 用户名/密码(USERNAME/PASSWORD) 105 * 0x03 至 X’7F’ IANA 分配(IANA ASSIGNED) 106 * 0x80 至 X’FE’ 私人方法保留(RESERVED FOR PRIVATE METHODS) 107 * 0xFF 无可接受方法(NO ACCEPTABLE METHODS) 108 */ 109 $authtypes = Socks5::parseAuth($buffer); 110 if ($authtypes === false) { 111 $socket->is_closing = true; 112 break; 113 } 114 if ($authtypes === true) { // continue 115 break; 116 } 117 $socket->status = Socks5::STATUS_ADDR; 118 $socket->write("\x05\x00"); // TODO: 暂时 119 $buffer = ''; 120 break; 121 case Socks5::STATUS_AUTH: 122 123 if ($callback && \is_callable($callback)) { 124 // TODO 125 } 126 break; 127 case Socks5::STATUS_ADDR: 128 self::info('Socks5::STATUS_ADDR'); 129 /** 130 * 建立代理连接 131 * +----------+------------+---------+-----------+-----------------------+------------+ 132 * |协议版本 |请求的类型 |保留字段 |地址类型 |地址数据 |地址端口 | 133 * +----------+------------+---------+-----------+-----------------------+------------+ 134 * |1个字节 |1个字节 |1个字节 |1个字节 |变长 |2个字节 | 135 * +----------+------------+---------+-----------+-----------------------+------------+ 136 * |0x05 |0x01 |0x00 |0x01 |0x0a,0x00,0x01,0x0a |0x00,0x50 | 137 * +----------+------------+---------+-----------+-----------------------+------------+ 138 */ 139 /** 140 * 请求类型 141 * CONNECT : 0x01, 建立代理连接 142 * BIND : 0x02,告诉代理服务器监听目标机器的连接,也就是让代理服务器创建socket监听来自目标机器的连接。FTP这类需要服务端主动联接客户端的的应用场景。 143 * 1. 只有在完成了connnect操作之后才能进行bind操作 144 * 2. bind操作之后，代理服务器会有两次响应, 第一次响应是在创建socket监听完成之后，第二次是在目标机器连接到代理服务器上之后。 145 * UDP ASSOCIATE : 0x03, udp 协议请求代理。 146 */ 147 if (strlen($buffer) < 2) { 148 break; 149 } 150 $cmd = ord($buffer[1]); 151 if ($cmd != Socks5::CMD_CONNECT) { 152 self::err("Bad command ".$cmd); 153 $socket->is_closing = true; 154 break; 155 } 156 $headers = Socks5::parseAddr($buffer); 157 if (!$headers) { 158 self::err('Error header'); 159 $socket->is_closing = true; 160 break; 161 } 162 if ($headers === true) { // continue 163 break; 164 } 165 /** 166 * 数据包转发 167 * +----+------+------+----------+----------+----------+ 168 * |RSV | FRAG | ATYP | DST.ADDR | DST.PORT | DATA | 169 * +----+------+------+----------+----------+----------+ 170 * | 2 | 1 | 1 | Variable | 2 | Variable | 171 * +----+------+------+----------+----------+----------+ 172 */ 173 $buffer = substr($buffer, $headers[3]); 174 $socket->status = Socks5::STATUS_CONNECTING; 175 if (!in_array($headers[0], [Socks5::TYPE_IPV4, Socks5::TYPE_HOST, Socks5::TYPE_IPV6])) { 176 self::err('Addr type error'); 177 $socket->is_closing = true; 178 break; 179 } 180 $tls = NULL; 181 if ($headers[2] == 443) { 182 $tls = new \Phalcon\Async\Network\TlsClientEncryption(); 183 $tls = $tls->withAllowSelfSigned(true); 184 } 185 $socket->client = \Phalcon\Async\Network\TcpSocket::connect($headers[1], $headers[2], $tls); 186 $socket->write(Socks5::REPLY_ADDR); 187 $socket->status = Socks5::STATUS_STREAM; 188 189 \Phalcon\Async\Task::async(function ($client) use ($socket) { 190 while (!$socket->is_closing && null !== ($chunk = $client->read())) { 191 $socket->write($chunk); 192 } 193 }, $socket->client); 194 break; 195 case Socks5::STATUS_STREAM: 196 self::info('Socks5::STATUS_STREAM'); 197 $socket->client->write($buffer); 198 $buffer = ''; 199 break; 200 } 201 } 202 } catch (\Throwable $e) { 203 self::err($e->getMessage()); 204 } finally { 205 $socket->close(); 206 } 207 }; 208 209 try { 210 $this->server = \Phalcon\Async\Network\TcpServer::listen($this->host, $this->port); 211 echo Phalcon\Cli\Color::info('start server listen:'.$this->host.':'.$this->port).PHP_EOL; 212 while (true) { 213 $socket = $this->server->accept(); 214 if ($socket === false) { 215 continue; 216 } 217 $this->pool->submit($worker, $socket); 218 } 219 } catch (\Throwable $e) { 220 self::err($e->getMessage()); 221 } finally { 222 if ($this->server) { 223 $this->server->close(); 224 } 225 } 226 227 } 228 229 static public function info($message) 230 { 231 if (self::$debug) { 232 echo Phalcon\Cli\Color::info($message).PHP_EOL; 233 } 234 } 235 236 static public function err($message) 237 { 238 echo Phalcon\Cli\Color::error($message).PHP_EOL; 239 } 240} 241 242 243 244$opts = new \Phalcon\Cli\Options('Websocket CLI'); 245$opts->add([ 246 'type' => \Phalcon\Cli\Options::TYPE_STRING, 247 'name' => 'server', 248 'shortName' => 's', 249 'required' => false, // 可选，需要用=号赋值 250 'help' => "address" 251]); 252$opts->add([ 253 'type' => \Phalcon\Cli\Options::TYPE_INT, 254 'name' => 'port', 255 'shortName' => 'p', 256 'required' => false, 257 'help' => "port" 258]); 259$opts->add([ 260 'type' => \Phalcon\Cli\Options::TYPE_BOOLEAN, 261 'name' => 'concurrency', 262 'shortName' => 'c', 263 'required' => false 264]); 265$opts->add([ 266 'type' => \Phalcon\Cli\Options::TYPE_BOOLEAN, 267 'name' => 'capacity', 268 'shortName' => 'C', 269 'required' => false 270]); 271$opts->add([ 272 'type' => \Phalcon\Cli\Options::TYPE_BOOLEAN, 273 'name' => 'debug', 274 'shortName' => 'v', 275 'required' => false, 276 'help' => "enable debug" 277]); 278$vals = $opts->parse(); 279if ($vals === false ) { 280 exit; 281} 282/** 283 * 运行 php websocket-server.php 284 */ 285if (isset($vals['debug'])) { 286 Socks5Server::$debug = true; 287 echo Phalcon\Cli\Color::info('Use debug mode').PHP_EOL; 288} 289$sserver = new Socks5Server(\Phalcon\Arr::get($vals, 'server', '0.0.0.0'), \Phalcon\Arr::get($vals, 'port', 10002), function($socket, $status, $data) { 290 // TODO 291}, \Phalcon\Arr::get($vals, 'concurrency', 500), \Phalcon\Arr::get($vals, 'capacity', 1)); 292$sserver->start();