當需要訪問第三方服務時，Nginx提供了兩種全非同步方式來與第三方伺服器通訊：upstream與subrequest。本文將介紹upstream的使用方式，Nginx的HTTP反向代理模組就是基於upstream方式實現，當我們希望把第三方服務的內容幾乎原封不懂地返回給使用者時，一般使用upstream方式，它可以非常高效地透傳HTTP。

upstream的使用方式並不複雜，它提供了8個回撥方法，見檔案http/ngx_http_upstream_t.h，使用者需要實現的其中的幾個回撥方法，本例中實現了3個回撥方法：

struct ngx_http_upstream_s{
//...
 

//構造發往上游伺服器的請求
    ngx_int_t                      (*create_request)(ngx_http_request_t *r);
//收到上游伺服器的TCP流時回撥，直到不返回NGX_AGAIN，處理http響應
    ngx_int_t                      (*process_header)(ngx_http_request_t *r);
//銷燬upstream時回撥
    void                           (*finalize_request)(ngx_http_request_t *r,ngx_int_t rc);
//...
 

};

那麼upstream是如何嵌入到一個請求中的？
模組在處理任何一個請求時都有ngx_http_request_t結構物件r，該結構中有一個ngx_http_upstream_t型別的成員upstream，見檔案http/ngx_http_request.h

struct ngx_http_request_s
{
    //...
    ngx_http_upstream_t    *upstream;
    //...
}

如果沒有使用upstream機制，則需要將相應的成員設定為NULL指標，否則需要對r->upstream進行設定。
HTTP模組啟用upstream機制的步驟如下：

使用upstream模組提供的ngx_http_upstream_init方法後，HTTP如何執行upstream框架，大致流程如下：

2.1 ngx_http_upstream_t結構體設定

需要大致瞭解struct ngx_http_upstream_s結構體及其相關成員，詳情見檔案http/ngx_http_upstream.h。
除了之前介紹到回撥函式之外，還有一些需要了解的引數：

struct ngx_http_upstream_s {
    //...
    //發什麼樣的請求給上游伺服器，create_request中實現
    ngx_chain_t                     *request_bufs;
    //可直接指定上游伺服器的地址
    ngx_http_upstream_resolved_t    *resolved;

    //buffer儲存接收上游伺服器的響應內容。複用
    ngx_buf_t                        buffer;

    //向客戶端轉發上游伺服器的包體時才有用。
    //1-多個緩衝區以及磁碟檔案轉發（Nginx與上游間的網速遠大於Nginx與下游客戶端之間的網速）
    //0-只使用該結構體中的buffer緩衝區向下遊轉發響應包體
    unsigned                         buffering:1;
    //...
};

關於buffering引數，upstream提供了3種處理上游伺服器包體的方式：

1.交由HTTP模組使用input filter回撥方法直接處理包體。
2.以固定緩衝區（buffer成員）轉發包體（本文使用的方式 buffering設定為0）
3.以多個緩衝加磁碟檔案的方式轉發包體等（buffering設定為1）。

2.2 ngx_http_upstream_conf_t結構體設定

該結構是ngx_http_upstream_t中的成員之一，通過nginx.conf和設定commands使用預設的方法可以對該結構體賦值。在深入理解Nginx書中，為了簡便期間直接在create_loc_conf回撥函式中建立物件並硬編碼賦值。下列指出結構體中必須設定的成員。

// http/ngx_http_upstream.h

typedef struct{
//...
     //連線上游伺服器的超時時間
    ngx_msec_t                       connect_timeout;
    //傳送tcp包到上游伺服器的超時時間
    ngx_msec_t                       send_timeout;
    //接收Tcp包的超時時間
    ngx_msec_t                       read_timeout;
//...
}ngx_http_upstream_conf_t;

2.3 設定需要訪問的第三方服務地址resolved成員

這部分就是熟悉套接字程式設計了，看到resovlved結構的成員就清楚了（見檔案http/ngx_http_upstream.h）：

typedef struct{
    //...
    ngx_uint_t naddrs;
    //上游伺服器地址
    struct sockaddr *sockaddr;
    socklen_t socklen;
    //...
}ngx_http_upstream_resolved_t;

2.4 設定回撥方法

ngx_http_upstream_t結構體中有8個回撥方法，不過必須實現的有3個回撥方法：

create_request:建立請求
process_header:處理響應
finalize_request:銷燬請求

書中給出google的URL搜尋示例，由於牆的原因，我們使用baidu，其URL搜尋方法是www.baidu.com/s?wd=haha在nginx.conf配置location，例如：

location /test{
    mytest;
}

如果訪問URL是/test?lumia則可以通過upstream機制向www.baidu.com傳送搜尋請求。

完整程式碼

原始碼來源自深入理解Nginx，註釋包含自己的理解。

//ngx_http_mytest_module.c

#include <ngx_config.h>
#include <ngx_core.h>
#include <ngx_http.h>

/*上下文結構體*/
typedef struct
{
    ngx_http_status_t status;
    ngx_str_t backendServer;
}ngx_http_mytest_ctx_t;

/*uptream相關配置，例如超時等待時間等*/
typedef struct
{
    ngx_http_upstream_conf_t upstream;
} ngx_http_mytest_conf_t;


/*配置項處理函式*/
static char *
ngx_http_mytest(ngx_conf_t *cf, ngx_command_t *cmd, void *conf);


/*真正的處理函式*/
/*設定upstream的host、回撥函式、啟動upstream等*/
static ngx_int_t ngx_http_mytest_handler(ngx_http_request_t *r);


/*建立ngx_http_mytest_conf_t結構體，硬編碼引數*/
static void* ngx_http_mytest_create_loc_conf(ngx_conf_t *cf);


/*設定hide_headers_hash*/
static char *ngx_http_mytest_merge_loc_conf(ngx_conf_t *cf, void *parent, void *child);


/*解析http頭部*/
static ngx_int_t
mytest_upstream_process_header(ngx_http_request_t *r);


/*處理http響應行*/
static ngx_int_t
mytest_process_status_line(ngx_http_request_t *r);


/*安全考慮隱藏頭部*/
static ngx_str_t  ngx_http_proxy_hide_headers[] =
{
    ngx_string("Date"),
    ngx_string("Server"),
    ngx_string("X-Pad"),
    ngx_string("X-Accel-Expires"),
    ngx_string("X-Accel-Redirect"),
    ngx_string("X-Accel-Limit-Rate"),
    ngx_string("X-Accel-Buffering"),
    ngx_string("X-Accel-Charset"),
    ngx_null_string
};


/*command 結構體陣列*/
static ngx_command_t  ngx_http_mytest_commands[] =
{

    {
        ngx_string("mytest"),
        NGX_HTTP_MAIN_CONF | NGX_HTTP_SRV_CONF | NGX_HTTP_LOC_CONF | NGX_HTTP_LMT_CONF | NGX_CONF_NOARGS,
        ngx_http_mytest,
        NGX_HTTP_LOC_CONF_OFFSET,
        0,
        NULL
    },

    ngx_null_command
};


/*模組上下文*/
static ngx_http_module_t  ngx_http_mytest_module_ctx =
{
    NULL, /* preconfiguration */
    NULL, /* postconfiguration */

    NULL, /* create main configuration */
    NULL, /* init main configuration */

    NULL, /* create server configuration */
    NULL, /* merge server configuration */

    ngx_http_mytest_create_loc_conf,/* create location configuration */
    ngx_http_mytest_merge_loc_conf  /* merge location configuration */
};


/*nginx 模組*/
ngx_module_t  ngx_http_mytest_module =
{
    NGX_MODULE_V1,
    &ngx_http_mytest_module_ctx,           /* module context */
    ngx_http_mytest_commands,              /* module directives */
    NGX_HTTP_MODULE,                       /* module type */
    NULL,                                  /* init master */
    NULL,                                  /* init module */
    NULL,                                  /* init process */
    NULL,                                  /* init thread */
    NULL,                                  /* exit thread */
    NULL,                                  /* exit process */
    NULL,                                  /* exit master */
    NGX_MODULE_V1_PADDING
};


static void*
ngx_http_mytest_create_loc_conf(ngx_conf_t *cf)
{
    ngx_http_mytest_conf_t *mycf;

    mycf = (ngx_http_mytest_conf_t  *)ngx_pcalloc(cf->pool, sizeof(ngx_http_mytest_conf_t));
    if (mycf == NULL)
    {
        return NULL;
    }

    /*對結構體變數中的成員進行硬編碼，超時連線相關*/
    mycf->upstream.connect_timeout = 60000;
    mycf->upstream.send_timeout = 60000;
    mycf->upstream.read_timeout = 60000;
    mycf->upstream.store_access = 0600;

    /*處理上游伺服器包體方式相關，這裡以固定buffer轉發包體*/
    mycf->upstream.buffering = 0;
    mycf->upstream.bufs.num = 8;
    mycf->upstream.bufs.size = ngx_pagesize;
    mycf->upstream.buffer_size = ngx_pagesize;
    mycf->upstream.busy_buffers_size = 2 * ngx_pagesize;
    mycf->upstream.temp_file_write_size = 2 * ngx_pagesize;
    mycf->upstream.max_temp_file_size = 1024 * 1024 * 1024;


    /*upstream模組要求hide_headers必須初始化*/
    mycf->upstream.hide_headers = NGX_CONF_UNSET_PTR;
    mycf->upstream.pass_headers = NGX_CONF_UNSET_PTR;

    return mycf;
}


static char *
ngx_http_mytest_merge_loc_conf(ngx_conf_t *cf, void *parent, void *child)
{
    ngx_http_mytest_conf_t *prev = (ngx_http_mytest_conf_t *)parent;
    ngx_http_mytest_conf_t *conf = (ngx_http_mytest_conf_t *)child;

    ngx_hash_init_t hash;
    hash.max_size = 100;
    hash.bucket_size = 1024;
    hash.name = "proxy_headers_hash";
    /*hide header*/
    if (ngx_http_upstream_hide_headers_hash(cf, &conf->upstream,
     &prev->upstream, ngx_http_proxy_hide_headers, &hash)
        != NGX_OK)
    {
        return NGX_CONF_ERROR;
    }
    return NGX_CONF_OK;
}

/*upstream  create_request 回撥，構建到上游伺服器的請求*/
static ngx_int_t
mytest_upstream_create_request(ngx_http_request_t *r)
{
    /*模仿baidu搜尋請求 /s?wd= */
    static ngx_str_t backendQueryLine =
        ngx_string("GET /s?wd=%V HTTP/1.1\r\nHost: www.baidu.com\r\nConnection: close\r\n\r\n");

    /*請求長度 -2表示 %V*/
    ngx_int_t queryLineLen = backendQueryLine.len + r->args.len - 2;

    /*記憶體池申請記憶體，請求結束時記憶體被自動釋放*/
    ngx_buf_t* b = ngx_create_temp_buf(r->pool, queryLineLen);
    if (b == NULL)
        return NGX_ERROR;

    /*last要指向請求的末尾*/
    b->last = b->pos + queryLineLen;

    /*作用相當於snprintf*/
    ngx_snprintf(b->pos, queryLineLen ,(char*)backendQueryLine.data, &r->args);

    /*傳送給上游伺服器的請求*/
    r->upstream->request_bufs = ngx_alloc_chain_link(r->pool);
    if (r->upstream->request_bufs == NULL)
        return NGX_ERROR;

    /*request_bufs這裡只包含1個ngx_buf_t緩衝區*/
    r->upstream->request_bufs->buf = b;
    r->upstream->request_bufs->next = NULL;

    r->upstream->request_sent = 0;
    r->upstream->header_sent = 0;

    /*header_hash不可以為0*/
    r->header_hash = 1;
    return NGX_OK;
}



static ngx_int_t
mytest_process_status_line(ngx_http_request_t *r)
{
    size_t len;
    ngx_int_t rc;
    ngx_http_upstream_t *u;

    /*取出http請求的上下文*/
    ngx_http_mytest_ctx_t* ctx = ngx_http_get_module_ctx(r, ngx_http_mytest_module);
    if (ctx == NULL)
    {
        return NGX_ERROR;
    }

    u = r->upstream;

    /*解析http響應行*/
    rc = ngx_http_parse_status_line(r, &u->buffer, &ctx->status);

    /*NGX_AGAIN表示繼續解析*/
    if (rc == NGX_AGAIN)
    {
        return rc;
    }

    /*NGX_ERROR沒有接收到合法的http響應行*/
    if (rc == NGX_ERROR)
    {
        ngx_log_error(NGX_LOG_ERR, r->connection->log, 0,
                      "upstream sent no valid HTTP/1.0 header");

        r->http_version = NGX_HTTP_VERSION_9;
        u->state->status = NGX_HTTP_OK;

        return NGX_OK;
    }

    /*解析到完整到響應行，將解析出來的內容賦值到headers_in結構體中*/
    if (u->state)
    {
        u->state->status = ctx->status.code;
    }

    /*賦值操作*/
    u->headers_in.status_n = ctx->status.code;

    len = ctx->status.end - ctx->status.start;

    u->headers_in.status_line.len = len;

    u->headers_in.status_line.data = ngx_pnalloc(r->pool, len);
    if (u->headers_in.status_line.data == NULL)
    {
        return NGX_ERROR;
    }
    ngx_memcpy(u->headers_in.status_line.data, ctx->status.start, len); 


    /*開始解析http響應頭部*/
    /*之後收到的新字元流將有新的回撥函式解析*/
    u->process_header = mytest_upstream_process_header;

    /*如果本次收到的字元流除了http響應行外，還有多餘的字元*/
    /*將由mytest_upstream_process_header方法解析*/
    return mytest_upstream_process_header(r);
}


/*處理http頭部*/
static ngx_int_t
mytest_upstream_process_header(ngx_http_request_t *r)
{
    ngx_int_t rc;

    /*為http頭部量身定製*/
    /*例如 key 儲存"Content-Length"*/
    /*value 儲存 "1024" */
    ngx_table_elt_t *h;

    ngx_http_upstream_header_t     *hh;

    ngx_http_upstream_main_conf_t  *umcf;

    /*將upstream模組配置項ngx_http_upstream_main_conf_t取了*/
    /*該結構體中儲存了需要做統一處理的http頭部名稱和回撥方法*/
    umcf = ngx_http_get_module_main_conf(r, ngx_http_upstream_module);

    /*迴圈的解析所有的http頭部*/
    for ( ;; )
    {
    /*解析http頭部*/
        rc = ngx_http_parse_header_line(r, &r->upstream->buffer, 1);

        /*NGX_OK解析出一行http頭部*/
        if (rc == NGX_OK)
        {
            /*向headers_in.headers這個ngx_list_t連結串列中新增http頭部*/
        /*ngx_list_t是先插入再賦值*/
            h = ngx_list_push(&r->upstream->headers_in.headers);
            if (h == NULL)
            {
                return NGX_ERROR;
            }

            /*構造剛剛新增到headers連結串列中的http頭部*/
            h->hash = r->header_hash;

        /*key-頭部名稱 value-對應的值*/
            h->key.len = r->header_name_end - r->header_name_start;
            h->value.len = r->header_end - r->header_start;


        /*zx:作者這裡分配這麼大空間的原因在於，在結構體中有三個變數*/
        //1.key
        //2.value
        //3.lowcase_case
        /*一次申請這一整段空間，+1用來'\0'來區分*/
        /*這也在於ngx_str_t的特性在於data欄位記錄的只是字串地址*/
            h->key.data = ngx_pnalloc(r->pool,
                                      h->key.len + 1 + h->value.len + 1 + h->key.len);
            if (h->key.data == NULL)
            {
                return NGX_ERROR;
            }

        /*賦值操作*/
            h->value.data = h->key.data + h->key.len + 1;
            h->lowcase_key = h->key.data + h->key.len + 1 + h->value.len + 1;

            ngx_memcpy(h->key.data, r->header_name_start, h->key.len);
            h->key.data[h->key.len] = '\0';
            ngx_memcpy(h->value.data, r->header_start, h->value.len);
            h->value.data[h->value.len] = '\0';

            if (h->key.len == r->lowcase_index)
            {
                ngx_memcpy(h->lowcase_key, r->lowcase_header, h->key.len);
            }
            else
            {
                ngx_strlow(h->lowcase_key, h->key.data, h->key.len);
            }

            /*upstream模組會對一些http頭部做特殊處理*/
            hh = ngx_hash_find(&umcf->headers_in_hash, h->hash,
                               h->lowcase_key, h->key.len);

            if (hh && hh->handler(r, h, hh->offset) != NGX_OK)
            {
                return NGX_ERROR;
            }

            continue;
        }

        /*返回NGX_HTTP_PARSE_HEADER_DONE表示響應中所有的http頭部都解析*/
        if (rc == NGX_HTTP_PARSE_HEADER_DONE)
        {
            /*如果之前解析http頭部時沒有發現server和date頭部*/
        /*據http協議新增這兩個頭部*/
            if (r->upstream->headers_in.server == NULL)
            {
                h = ngx_list_push(&r->upstream->headers_in.headers);
                if (h == NULL)
                {
                    return NGX_ERROR;
                }

        /*zx:檢視ngx_hash_key_lc/ngx_hash_key原始碼可知，可以改寫*/
        ngx_str_t str=ngx_string("server");
        h->hash=ngx_hash_key_lc(str.data,str.len);
#if 0
                h->hash = ngx_hash(ngx_hash(ngx_hash(ngx_hash(
                                                      ngx_hash('s', 'e'), 'r'), 'v'), 'e'), 'r');
#endif

                ngx_str_set(&h->key, "Server");
                ngx_str_null(&h->value);
                h->lowcase_key = (u_char *) "server";
            }

            if (r->upstream->headers_in.date == NULL)
            {
                h = ngx_list_push(&r->upstream->headers_in.headers);
                if (h == NULL)
                {
                    return NGX_ERROR;
                }

        ngx_str_t str=ngx_string("date");
        /*zx:同上*/
        h->hash=ngx_hash_key_lc(str.data,str.len);
#if 0
                h->hash = ngx_hash(ngx_hash(ngx_hash('d', 'a'), 't'), 'e');
#endif

                ngx_str_set(&h->key, "Date");
                ngx_str_null(&h->value);
                h->lowcase_key = (u_char *) "date";
            }

            return NGX_OK;
        }

    /*沒有解析到完整的http頭部，繼續接收新的字元流*/
        if (rc == NGX_AGAIN)
        {
            return NGX_AGAIN;
        }

        /*其他返回值都是非法的*/
        ngx_log_error(NGX_LOG_ERR, r->connection->log, 0,
                      "upstream sent invalid header");
        return NGX_HTTP_UPSTREAM_INVALID_HEADER;
    }
}

/*finalize_request回撥 請求結束前會呼叫*/
static void
mytest_upstream_finalize_request(ngx_http_request_t *r, ngx_int_t rc)
{
    ngx_log_error(NGX_LOG_DEBUG, r->connection->log, 0,
                  "mytest_upstream_finalize_request");
}


static char *
ngx_http_mytest(ngx_conf_t *cf, ngx_command_t *cmd, void *conf)
{
    ngx_http_core_loc_conf_t  *clcf;

    /*首先找到mytest配置項所屬的配置塊，clcf貌似是location塊內的資料*/
    /*結構，其實不然，它可以是main、srv或者loc級別配置項，也就是說在每個*/
    /*http{}和server{}內也都有一個ngx_http_core_loc_conf_t結構體*/
    clcf = ngx_http_conf_get_module_loc_conf(cf, ngx_http_core_module);

    /*http框架在處理使用者請求進行到NGX_HTTP_CONTENT_PHASE階段時，如果*/
    /*請求的主機域名、URI與mytest配置項所在的配置塊相匹配，就將呼叫我們*/
    /*實現的ngx_http_mytest_handler方法處理這個請求*/
    clcf->handler = ngx_http_mytest_handler;

    return NGX_CONF_OK;
}

static ngx_int_t
ngx_http_mytest_handler(ngx_http_request_t *r)
{
    /*首先建立http上下文結構體ngx_http_mytest_ctx_t*/
    ngx_http_mytest_ctx_t* myctx = ngx_http_get_module_ctx(r, ngx_http_mytest_module);
    if (myctx == NULL)
    {
        myctx = ngx_palloc(r->pool, sizeof(ngx_http_mytest_ctx_t));
        if (myctx == NULL)
        {
            return NGX_ERROR;
        }
        /*將新建的上下文與請求關聯起來*/
        ngx_http_set_ctx(r, myctx, ngx_http_mytest_module);
    }

    /*對每1個要使用upstream的請求，必須呼叫且只能呼叫1次*/
    /*ngx_http_upstream_create方法，它會初始化r->upstream成員*/
    if (ngx_http_upstream_create(r) != NGX_OK)
    {
        ngx_log_error(NGX_LOG_ERR, r->connection->log, 0, "ngx_http_upstream_create() failed");
        return NGX_ERROR;
    }

    /*得到配置upstream結構體ngx_http_mytest_conf_t*/
    ngx_http_mytest_conf_t  *mycf = (ngx_http_mytest_conf_t*) ngx_http_get_module_loc_conf(r, ngx_http_mytest_module);

    /*配置upstream*/
    ngx_http_upstream_t *u = r->upstream;

    /*這裡用配置檔案中的結構體來賦給r->upstream->conf成員*/
    /*ngx_http_upstream_conf_t結構*/
    u->conf = &mycf->upstream;

    /*決定轉發包體時使用的緩衝區*/
    u->buffering = mycf->upstream.buffering;

    /*以下程式碼開始初始化resolved結構體，用來儲存上游伺服器的地址*/

    u->resolved = (ngx_http_upstream_resolved_t*) ngx_pcalloc(r->pool, sizeof(ngx_http_upstream_resolved_t));
    if (u->resolved == NULL)
    {
        ngx_log_error(NGX_LOG_ERR, r->connection->log, 0,
                      "ngx_pcalloc resolved error. %s.", strerror(errno));
        return NGX_ERROR;
    }

    /*這裡的上游伺服器就是www.baidu.com*/
    static struct sockaddr_in backendSockAddr;
    struct hostent *pHost = gethostbyname((char*) "www.baidu.com");
    if (pHost == NULL)
    {
        ngx_log_error(NGX_LOG_ERR, r->connection->log, 0,
                      "gethostbyname fail. %s", strerror(errno));

        return NGX_ERROR;
    }

    /*訪問上游伺服器的80埠*/
    backendSockAddr.sin_family = AF_INET;
    backendSockAddr.sin_port = htons((in_port_t) 80);

    memcpy(&backendSockAddr.sin_addr,pHost->h_addr_list[0],sizeof(struct in_addr));
    char* pDmsIP = inet_ntoa(*(struct in_addr*) (pHost->h_addr_list[0]));
    /*inet_addr*/
    //backendSockAddr.sin_addr.s_addr = inet_addr(pDmsIP);
    myctx->backendServer.data = (u_char*)pDmsIP;
    myctx->backendServer.len = strlen(pDmsIP);

    /*將地址設定到resolved成員中*/
    u->resolved->sockaddr = (struct sockaddr *)&backendSockAddr;
    u->resolved->socklen = sizeof(struct sockaddr_in);
    u->resolved->naddrs = 1;

    /*設定三個必須實現的回撥方法:建立請求、處理頭部、請求銷燬*/
    u->create_request = mytest_upstream_create_request;
    u->process_header = mytest_process_status_line;
    u->finalize_request = mytest_upstream_finalize_request;

    /*這裡必須將count成員加1*/
    r->main->count++;
    /*啟動upstream*/
    ngx_http_upstream_init(r);
    /*必須返回NGX_DONE*/
    return NGX_DONE;
}

將模組編入nginx，啟動nginx服務，curl一下，百度就出來了。

curl -v localhost:1024/test?lumia

如圖：

深入理解Nginx