请问socket如何非阻塞？

//类文件其它部分就不贴出来了
$server = new Locus_push();
$server->run();

public function run()
    {
        while (true) {
            if (!self::$redis) {
                self::$redis = new redis();
                self::$redis->connect('127.0.0.1');
            }
            $changes = $this->_sockets;
            $write  = NULL;
            $except = NULL;
            socket_select($changes, $write, $except, NULL);
            foreach ($changes as $key => $_sock) {
                socket_set_nonblock($_sock);
                if ($this->_socket == $_sock) {
                    //判断是不是新接入的socket

                    $this->e("--------------new---socket------------");
                    $this->e($_sock);
                    $this->e("--------------new---socket------------");


                    $newClient = socket_accept($_sock);
                    $line      = trim(socket_read($newClient, 1024));
                    $this->handshaking($newClient, $line);
                    $key                 = uniqid();
                    $this->_sockets[]    = $newClient;
                    $this->clients[$key] = [
                        'socket' => $newClient,
                        'hand'   => true
                    ];


                    $queue_key = "car_1_locus_queue";
                    while ($item = self::$redis->rPop($queue_key)) {
                        //根据每段坐标行驶时间 休息对应的秒数
                        $item_arr = json_decode($item, true);
                        sleep(ceil($item_arr['dur'] / 1000));
                        $this->send($newClient, $item);

                    }

                    echo "--------------new--socket--user---------\n";
                    var_dump($this->clients);
                    echo "--------------new--socket--user---------\n";
                } else {
                    $len    = 0;
                    $buffer = '';
                    //读取该socket的信息，注意：第二个参数是引用传参即接收数据，第三个参数是接收数据的长度
                    do {
                        $l      = socket_recv($_sock, $buf, 1000, 0);
                        $len    += $l;
                        $buffer .= $buf;
                    } while ($l == 1000);
                    $k = $this->getClient($_sock);
                    //如果接收的信息长度小于7，则该client的socket为断开连接
                    if ($len < 7) {
                        //给该client的socket进行断开操作，并在$this->sockets和$this->users里面进行删除
                        $this->close($k);
                        continue;
                    }
                    if (!$this->clients[$k]['hand']) {
                        $this->handshaking($_sock, $buffer);
                    } else {
                        //走到这里就是该client发送信息了，对接受到的信息进行处理
                        $client_msg = $this->message($buffer);
                        $client_msg_array = json_decode($client_msg, true);
                        if ($client_msg_array['type'] == 'get_car_locus') {
                            $queue_key = "car_1_locus_queue";
                           //这段就是行队列里面取数据，然后循环推送给用户
                           while ($item = self::$redis->rPop($queue_key)) {
                                //根据每段坐标行驶时间 休息对应的秒数
                                $item_arr = json_decode($item, true);
                                sleep(ceil($item_arr['dur'] / 1000));
                                foreach ($this->clients as $key => $client) {
                                    $this->send($client['socket'], $item);
                                }
                            }
                        } elseif ($client_msg_array['type'] == 'get_aircraft_locus') {}
                    }
                }
            }
        }
    }Copy