問(wèn)題描述

這個(gè)向網(wǎng)關(guān)發(fā)送請(qǐng)求并接收應(yīng)答，$client是從靜態(tài)屬性中取出，存在多進(jìn)程并用的情況，
那在進(jìn)程并發(fā)執(zhí)行 向網(wǎng)關(guān)發(fā)送消息時(shí)，能發(fā)確保收到的應(yīng)答是本進(jìn)程發(fā)送的請(qǐng)求應(yīng)答呢？

 /**
     * 批量向gateway發(fā)送并獲取數(shù)據(jù)
     * @param $gateway_buffer_array
     * @return array
     */
    protected static function getBufferFromGateway($gateway_buffer_array)
    {
        $client_array = $status_data = $client_address_map = $receive_buffer_array = $recv_length_array = array();
        // 批量向所有g(shù)ateway進(jìn)程發(fā)送請(qǐng)求數(shù)據(jù)
        foreach ($gateway_buffer_array as $address => $gateway_buffer) {
            $client = static::getGatewayConnection("tcp://$address");
            if (strlen($gateway_buffer) === stream_socket_sendto($client, $gateway_buffer)) {
                $socket_id                        = (int)$client;
                $client_array[$socket_id]         = $client;
                $client_address_map[$socket_id]   = explode(':', $address);
                $receive_buffer_array[$socket_id] = '';
            }
        }
        // 超時(shí)5秒
        $timeout    = 5;
        $time_start = microtime(true);
        // 批量接收請(qǐng)求
        while (count($client_array) > 0) {
            $write = $except = array();
            $read  = $client_array;
            if (@stream_select($read, $write, $except, $timeout)) {
                foreach ($read as $client) {
                    $socket_id = (int)$client;
                    $buffer    = stream_socket_recvfrom($client, 65535);
                    if ($buffer !== '' && $buffer !== false) {
                        $receive_buffer_array[$socket_id] .= $buffer;
                        $receive_length = strlen($receive_buffer_array[$socket_id]);
                        if (empty($recv_length_array[$socket_id]) && $receive_length >= 4) {
                            $recv_length_array[$socket_id] = current(unpack('N', $receive_buffer_array[$socket_id]));
                        }
                        if (!empty($recv_length_array[$socket_id]) && $receive_length >= $recv_length_array[$socket_id] + 4) {
                            unset($client_array[$socket_id]);
                        }
                    } elseif (feof($client)) {
                        unset($client_array[$socket_id]);
                    }
                }
            }
            if (microtime(true) - $time_start > $timeout) {
                static::$gatewayConnections = [];
                break;
            }
        }
        $format_buffer_array = array();
        foreach ($receive_buffer_array as $socket_id => $buffer) {
            $local_ip                                    = ip2long($client_address_map[$socket_id][0]);
            $local_port                                  = $client_address_map[$socket_id][1];
            $format_buffer_array[$local_ip][$local_port] = unserialize(substr($buffer, 4));
        }
        return $format_buffer_array;
    }