摘要：支持多路復(fù)用支持對(duì)和已建立連接的復(fù)用，如果舊連接已失效則主動(dòng)關(guān)閉舊連接，如果連接有效則嘗試使用已有連接傳輸數(shù)據(jù)。

對(duì)于同一服務(wù)可能存在多次調(diào)用的情況，然而每次調(diào)用都需要建立一次tcp連接導(dǎo)致大量重復(fù)工作的同時(shí)還增加了連接超時(shí)或連接錯(cuò)誤的概率，為了減少tcp連接次數(shù)最大限度的提高連接利用率，需要能夠重復(fù)利用每個(gè)tcp連接。

HTTP1.1與HTTP2.0支持對(duì)于一次TCP連接建立的通道重復(fù)使用。

HTTP2.0支持多路復(fù)用

CURL支持對(duì)HTTP1.1和HTTP2.0已建立連接的復(fù)用，如果舊連接已失效則主動(dòng)關(guān)閉舊連接，如果連接有效則嘗試使用已有連接傳輸數(shù)據(jù)。關(guān)鍵代碼如下：

// php/ext/url/interface.c
/* {{{ proto bool curl_exec(resource ch)
   Perform a cURL session */
PHP_FUNCTION(curl_exec)
{
    CURLcode    error;
    zval        *zid;
    php_curl    *ch;

    if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "r", &zid) == FAILURE) {
        return;
    }

    ZEND_FETCH_RESOURCE(ch, php_curl *, &zid, -1, le_curl_name, le_curl);

    _php_curl_verify_handlers(ch, 1 TSRMLS_CC);

    _php_curl_cleanup_handle(ch);

   // 調(diào)用CURL方法
    error = curl_easy_perform(ch->cp);
    SAVE_CURL_ERROR(ch, error);
    /* CURLE_PARTIAL_FILE is returned by HEAD requests */
    if (error != CURLE_OK && error != CURLE_PARTIAL_FILE) {
        if (ch->handlers->write->buf.len > 0) {
            smart_str_free(&ch->handlers->write->buf);
        }
        RETURN_FALSE;
    }

    if (ch->handlers->std_err) {
        php_stream  *stream;
        stream = (php_stream*)zend_fetch_resource(&ch->handlers->std_err TSRMLS_CC, -1, NULL, NULL, 2, php_file_le_stream(), php_file_le_pstream());
        if (stream) {
            php_stream_flush(stream);
        }
    }

    if (ch->handlers->write->method == PHP_CURL_RETURN && ch->handlers->write->buf.len > 0) {
        smart_str_0(&ch->handlers->write->buf);
        RETURN_STRINGL(ch->handlers->write->buf.c, ch->handlers->write->buf.len, 1);
    }

    /* flush the file handle, so any remaining data is synched to disk */
    if (ch->handlers->write->method == PHP_CURL_FILE && ch->handlers->write->fp) {
        fflush(ch->handlers->write->fp);
    }
    if (ch->handlers->write_header->method == PHP_CURL_FILE && ch->handlers->write_header->fp) {
        fflush(ch->handlers->write_header->fp);
    }

    if (ch->handlers->write->method == PHP_CURL_RETURN) {
        RETURN_EMPTY_STRING();
    } else {
        RETURN_TRUE;
    }
}
/* }}} */


// curl/lib/url.c line 4328

  // 主動(dòng)關(guān)閉已失效的連接
  prune_dead_connections(data);

  /*************************************************************
   * Check the current list of connections to see if we can
   * re-use an already existing one or if we have to create a
   * new one.
   *************************************************************/

  /* reuse_fresh is TRUE if we are told to use a new connection by force, but
     we only acknowledge this option if this is not a re-used connection
     already (which happens due to follow-location or during a HTTP
     authentication phase). */
  if(data->set.reuse_fresh && !data->state.this_is_a_follow)
    reuse = FALSE;
  else
    // 從已存在的鏈接中查找出可以復(fù)用的連接（如果是不支持多路復(fù)用且正在使用中的連接會(huì)被忽略）
    reuse = ConnectionExists(data, conn, &conn_temp, &force_reuse, &waitpipe);

  /* If we found a reusable connection, we may still want to
     open a new connection if we are pipelining. */
  if(reuse && !force_reuse && IsPipeliningPossible(data, conn_temp)) {
    size_t pipelen = conn_temp->send_pipe.size + conn_temp->recv_pipe.size;
    if(pipelen > 0) {
      infof(data, "Found connection %ld, with requests in the pipe (%zu)
",
            conn_temp->connection_id, pipelen);

      if(conn_temp->bundle->num_connections < max_host_connections &&
         data->state.conn_cache->num_connections < max_total_connections) {
        /* We want a new connection anyway */
        reuse = FALSE;

        infof(data, "We can reuse, but we want a new connection anyway
");
      }
    }
  }

  if(reuse) {
    /*
     * We already have a connection for this, we got the former connection
     * in the conn_temp variable and thus we need to cleanup the one we
     * just allocated before we can move along and use the previously
     * existing one.
     */
    conn_temp->inuse = TRUE; /* mark this as being in use so that no other
                                handle in a multi stack may nick it */
    reuse_conn(conn, conn_temp);
    free(conn);          /* we don"t need this anymore */
    conn = conn_temp;
    *in_connect = conn;

    infof(data, "Re-using existing connection! (#%ld) with %s %s
",
          conn->connection_id,
          conn->bits.proxy?"proxy":"host",
          conn->socks_proxy.host.name ? conn->socks_proxy.host.dispname :
          conn->http_proxy.host.name ? conn->http_proxy.host.dispname :
                                       conn->host.dispname);
  }
  else {
    /* We have decided that we want a new connection. However, we may not
       be able to do that if we have reached the limit of how many
       connections we are allowed to open. */
    struct connectbundle *bundle = NULL;

    if(conn->handler->flags & PROTOPT_ALPN_NPN) {
      /* The protocol wants it, so set the bits if enabled in the easy handle
         (default) */
      if(data->set.ssl_enable_alpn)
        conn->bits.tls_enable_alpn = TRUE;
      if(data->set.ssl_enable_npn)
        conn->bits.tls_enable_npn = TRUE;
    }

    if(waitpipe)
      /* There is a connection that *might* become usable for pipelining
         "soon", and we wait for that */
      connections_available = FALSE;
    else
      bundle = Curl_conncache_find_bundle(conn, data->state.conn_cache);

    if(max_host_connections > 0 && bundle &&
       (bundle->num_connections >= max_host_connections)) {
      struct connectdata *conn_candidate;

      /* The bundle is full. Let"s see if we can kill a connection. */
      conn_candidate = find_oldest_idle_connection_in_bundle(data, bundle);

      if(conn_candidate) {
        /* Set the connection"s owner correctly, then kill it */
        conn_candidate->data = data;
        (void)Curl_disconnect(conn_candidate, /* dead_connection */ FALSE);
      }
      else {
        infof(data, "No more connections allowed to host: %d
",
              max_host_connections);
        connections_available = FALSE;
      }
    }

    if(connections_available &&
       (max_total_connections > 0) &&
       (data->state.conn_cache->num_connections >= max_total_connections)) {
      struct connectdata *conn_candidate;

      /* The cache is full. Let"s see if we can kill a connection. */
      conn_candidate = Curl_conncache_oldest_idle(data);

      if(conn_candidate) {
        /* Set the connection"s owner correctly, then kill it */
        conn_candidate->data = data;
        (void)Curl_disconnect(conn_candidate, /* dead_connection */ FALSE);
      }
      else {
        infof(data, "No connections available in cache
");
        connections_available = FALSE;
      }
    }

    if(!connections_available) {
      infof(data, "No connections available.
");

      conn_free(conn);
      *in_connect = NULL;

      result = CURLE_NO_CONNECTION_AVAILABLE;
      goto out;
    }
    else {
      /*
       * This is a brand new connection, so let"s store it in the connection
       * cache of ours!
       */
      Curl_conncache_add_conn(data->state.conn_cache, conn);
    }

#if defined(USE_NTLM)
    /* If NTLM is requested in a part of this connection, make sure we don"t
       assume the state is fine as this is a fresh connection and NTLM is
       connection based. */
    if((data->state.authhost.picked & (CURLAUTH_NTLM | CURLAUTH_NTLM_WB)) &&
       data->state.authhost.done) {
      infof(data, "NTLM picked AND auth done set, clear picked!
");
      data->state.authhost.picked = CURLAUTH_NONE;
      data->state.authhost.done = FALSE;
    }

    if((data->state.authproxy.picked & (CURLAUTH_NTLM | CURLAUTH_NTLM_WB)) &&
       data->state.authproxy.done) {
      infof(data, "NTLM-proxy picked AND auth done set, clear picked!
");
      data->state.authproxy.picked = CURLAUTH_NONE;
      data->state.authproxy.done = FALSE;
    }
#endif
  }

// curl/lib/multi.c
/*
 * This function scans the connection cache for half-open/dead connections,
 * closes and removes them.
 * The cleanup is done at most once per second.
 */
static void prune_dead_connections(struct Curl_easy *data)
{
  struct curltime now = Curl_now();
  time_t elapsed = Curl_timediff(now, data->state.conn_cache->last_cleanup);

  if(elapsed >= 1000L) {
    Curl_conncache_foreach(data, data->state.conn_cache, data,
                           call_disconnect_if_dead);
    data->state.conn_cache->last_cleanup = now;
  }
}

class Curl
{
    protected $ch = null;
    protected $errorCode = 0;
    protected $errorMsg = "";
    protected $curlInfo = array();
    protected $verbose = null;
    private static $instance = null;

    public function getLastErrorCode()
    {
        return $this->errorCode;
    }

    public function getLastErrorMsg()
    {
        return $this->errorMsg;
    }

    public function getLastCurlInfo()
    {
        return $this->curlInfo;
    }

    private function __construct()
    {
        $this->ch = curl_init();
    }

    /*
     * 單例模式防止被clone
     */
    private function __clone(){
        throw new CurlException("The Curl library can"t be cloned");
    }

    /*
     * 使用單例模式調(diào)用
     */
    public static function getInstance(){
        if(!self::$instance instanceof self){
            self::$instance = new self();
        }
        return self::$instance;
    }

    /**
     * curl以get的方式訪問
     * @param $url
     * @param int $timeout
     * @param array $params get請(qǐng)求的參數(shù)，可以在url中直接帶參數(shù)，也可以在這里傳
     * @param array $headers 支持["Accept" => "application/json"]和["Accept: application/json"]兩種方式
     * @return mixed
     */
    public function get($url, $timeout = 3, $params = [], $headers = [])
    {
        $url = $this->buildQuery($url,$params);
        $this->setGeneralOption($url,$timeout,$headers);
        $result = $this->execute();
        return $result;
    }

    /**
     * curl以post的方式訪問
     * @param $url
     * @param array $params
     * @param array $headers 支持["Accept" => "application/json"]和["Accept: application/json"]兩種方式
     * @param bool $withHttpBuildQuery
     * @param int $timeout
     * @return mixed
     */
    public function post($url, $params = [], $headers = [], $withHttpBuildQuery = true, $timeout=3)
    {
        if ($withHttpBuildQuery) {
            if (!is_array($params)) {
                $params = [$params];
            }
            $params = http_build_query($params);
        }
        
        curl_setopt($this->ch, CURLOPT_POST, 1);
        curl_setopt($this->ch, CURLOPT_POSTFIELDS, $params);
        $this->setGeneralOption($url,$timeout,$headers);
        $result = $this->execute();
        return $result;
    }

    /**
     * curl以HTTP2.0 get的方式訪問
     * @param string $url 請(qǐng)求URL
     * @param int $timeout 超時(shí)時(shí)間，單位秒
     * @param array $params get請(qǐng)求的參數(shù)，可以在url中直接帶參數(shù)，也可以在這里傳
     * @param array $headers 支持["Accept" => "application/json"]和["Accept: application/json"]兩種方式
     * @return mixed
     */
    public function get2($url, $timeout = 3, $params = [], $headers = [])
    {
        $url = $this->buildQuery($url,$params);
        $this->setGeneralOption($url,$timeout,$headers,CURL_HTTP_VERSION_2_0);
        $result = $this->execute();
        return $result;
    }

    /**
     * curl以HTTP2.0 post的方式訪問
     * @param string $url 請(qǐng)求URL
     * @param array $params
     * @param array $headers 支持["Accept" => "application/json"]和["Accept: application/json"]兩種方式
     * @param bool $withHttpBuildQuery
     * @param int $timeout 超時(shí)時(shí)間，單位秒
     * @return mixed
     */
    public function post2($url, $params = [], $headers = [], $withHttpBuildQuery = true, $timeout=3)
    {
        if ($withHttpBuildQuery) {
            if (!is_array($params)) {
                $params = [$params];
            }
            $params = http_build_query($params);
        }

        curl_setopt($this->ch, CURLOPT_POST, 1);
        curl_setopt($this->ch, CURLOPT_POSTFIELDS, $params);
        $this->setGeneralOption($url,$timeout,$headers,CURL_HTTP_VERSION_2_0);
        $result = $this->execute();
        return $result;
    }

    /**
     * 實(shí)例銷毀前主動(dòng)關(guān)閉所有連接
     */
    public function __destruct()
    {
        $this->close();
    }

    /**
     * 關(guān)閉所有連接
     * Description: 這一步在php-fpm中可以省略，實(shí)例結(jié)束后php-fpm的垃圾回收機(jī)制會(huì)關(guān)閉
     */
    public function close()
    {
        if (is_resource($this->ch)) {
            curl_close($this->ch);
            $this->ch = null;
        }
    }

    /**
     * 拼接請(qǐng)求URL
     * @param string $url 請(qǐng)求URL
     * @param array $params 待拼接參數(shù)
     * @return string
     */
    protected function buildQuery($url,$params)
    {
        if (!$params) {
            return $url;
        }

        if (strpos($url, "?") === false) {
            $url .= "?";
        } else {
            $url .= "&";
        }

        $url .= http_build_query($params);
        return $url;
    }

    /**
     * 設(shè)置通用curl配置
     * @param string $url 請(qǐng)求URL
     * @param int $timeout 超時(shí)時(shí)間，單位秒
     * @param array $headers 請(qǐng)求header
     * @param int $httpVersion 使用的http協(xié)議，默認(rèn)為1.1
     */
    protected function setGeneralOption($url,$timeout,$headers=array(),$httpVersion=CURL_HTTP_VERSION_1_1)
    {
        curl_setopt($this->ch, CURLOPT_URL, $url);
        curl_setopt($this->ch, CURLOPT_SSL_VERIFYPEER, true); //讓CURL支持HTTPS訪問
        curl_setopt($this->ch, CURLOPT_SSL_VERIFYHOST, 2);
        curl_setopt($this->ch, CURLOPT_RETURNTRANSFER, 1);
        curl_setopt($this->ch, CURLOPT_TIMEOUT, $timeout);
        curl_setopt($this->ch, CURLOPT_HTTP_VERSION, $httpVersion);
        // 啟用debug獲取更詳細(xì)的連接信息，與CURLOPT_HEADER互斥
        curl_setopt($this->ch, CURLOPT_VERBOSE, 1);
        $this->verbose = fopen("php://temp", "w+");
        curl_setopt($this->ch, CURLOPT_STDERR, $this->verbose);

        if ($headers && is_array($headers)) {
            $realHeader = [];
            foreach ($headers as $key => $val) {
                if (is_string($key)) {
                    $realHeader[] = $key. ": ". $val;
                } else {
                    $realHeader[] = $val;
                }
            }
            curl_setopt($this->ch, CURLOPT_HTTPHEADER, $realHeader);
        }
    }

    /**
     * 執(zhí)行請(qǐng)求
     * @return mixed
     */
    protected function execute()
    {
        $result = curl_exec($this->ch);

        // 記錄詳細(xì)的debug信息
        $this->curlInfo = curl_getinfo($this->ch);
        rewind($this->verbose);
        $this->curlInfo["verbose"] = stream_get_contents($this->verbose);
        $this->verbose = null;

        if ($result === false) {
            $this->errorCode = curl_errno($this->ch);
            $this->errorMsg = curl_error($this->ch);
            $this->curlInfo["error_code"] = $this->errorCode;
            $this->curlInfo["error_message"] = $this->errorMsg;
        }

        curl_reset($this->ch);
        return $result;
    }
}

class CurlException extends Exception {}

由于PHP-FPM的回收機(jī)制，一次請(qǐng)求結(jié)束后CURL的資源將會(huì)被回收，這意味著這次請(qǐng)求建立的TCP連接將會(huì)被關(guān)閉，在這種情況下就無法達(dá)到垮請(qǐng)求復(fù)用的目的。因此可以利用獨(dú)立進(jìn)程的方式來維護(hù)已建立的TCP連接專門負(fù)責(zé)CURL的請(qǐng)求。

對(duì)于HTTP2.0而言，由于支持多路復(fù)用，因此對(duì)于一個(gè)域名的請(qǐng)求建立一次tcp連接后可以支持同時(shí)多個(gè)請(qǐng)求的處理（HTTP1.1一個(gè)tcp連接同時(shí)只支持一個(gè)請(qǐng)求，如果第二個(gè)請(qǐng)求同時(shí)到達(dá)則CURL將建立新的tcp連接以便完成請(qǐng)求），利用這一特性使用獨(dú)立進(jìn)程配合協(xié)程可以達(dá)到對(duì)于單一場(chǎng)景的curl高并發(fā)的支撐。

同理除PHP外可擴(kuò)展到其他語言。

源地址?By佐柱

轉(zhuǎn)載請(qǐng)注明出處，也歡迎偶爾逛逛我的小站，謝謝 ：）