golang 反向代理reverseproxy源碼分析

1 基於reverse proxy實現的反向代理例子

package main

import (
    "log"
    "net/http"
    "net/http/httputil"
    "net/url"
)

func main() {
    // 地址重寫實例
    // http://127.0.0.1:8888/test?id=1  =》 http://127.0.0.1:8081/reverse/test?id=1

    rs1 := "http://127.0.0.1:8081/reverse"
    targetUrl , err := url.Parse(rs1)
    if err != nil {
        log.Fatal("err")
    }
    proxy := httputil.NewSingleHostReverseProxy(targetUrl)
    log.Println("Reverse proxy server serve at : 127.0.0.1:8888")
    if err := http.ListenAndServe(":8888",proxy);err != nil{
        log.Fatal("Start server failed,err:",err)
    }
}

$ curl http://127.0.0.1:8888/hello?id=123 -s
http://127.0.0.1:8081/reverse/hello?id=123

2 reverse proxy源碼分析

主要結構體reverseproxy

// 處理進來的請求，併發送給另一臺server實現反向代理，並將請求回傳給客戶端
type ReverseProxy struct {
    // 經過transport 可修改請求，響應體將原封不動的返回
    Director func(*http.Request)

    // 鏈接池複用鏈接，用於執行請求，爲nil則默認使用http.DefaultTransport
    Transport http.RoundTripper

    // 刷新到客戶端的刷新時間間隔
    // 流式請求下該參數會被忽略，全部反向代理請求將被當即刷新
    FlushInterval time.Duration

    // 默認爲std.err，可用於自定義logger
    ErrorLog *log.Logger

    // 用於執行io.CopyBuffer 複製響應體，將其存放至byte切片
    BufferPool BufferPool

    // 用於修改響應結果及HTTP狀態碼，當返回結果error不爲空時，會調用ErrorHandler
    ModifyResponse func(*http.Response) error

    // 用於處理後端和ModifyResponse返回的錯誤信息，默認將返回傳遞過來的錯誤信息，並返回HTTP 502
    ErrorHandler func(http.ResponseWriter, *http.Request, error)
}

主要方法

// 實例化ReverseProxy
// 假設目標URI(target path)是/base ，請求的URI（target request）是/dir，那麼請求將被反向代理到http://x.x.x.x./base/dir
// ReverseProxy  不會rewrite Host header，須要重寫Host，可在Director函數中自定義

func NewSingleHostReverseProxy(target *url.URL) *ReverseProxy {
  // 獲取請求參數，例如請求的是/dir?id=123，那麼rawQuery ：id=123
    targetQuery := target.RawQuery

    // 實例化director
    director := func(req *http.Request) {
        req.URL.Scheme = target.Scheme   // http or https
        req.URL.Host = target.Host      // 主機名（ip+端口 或 域名+端口）
        req.URL.Path = singleJoiningSlash(target.Path, req.URL.Path) // 請求URL拼接

        // 使用"&"符號拼接請求參數
        if targetQuery == "" || req.URL.RawQuery == "" {
            req.URL.RawQuery = targetQuery + req.URL.RawQuery
        } else {
            req.URL.RawQuery = targetQuery + "&" + req.URL.RawQuery
        }

        // 若"User-Agent" 這個header不存在，則置空
        if _, ok := req.Header["User-Agent"]; !ok {
            // explicitly disable User-Agent so it's not set to default value
            req.Header.Set("User-Agent", "")
        }
    }
    return &ReverseProxy{Director: director}
}

url 拼接方法

func singleJoiningSlash(a, b string) string {
    aslash := strings.HasSuffix(a, "/")
    bslash := strings.HasPrefix(b, "/")
    switch {
    case aslash && bslash:      // 若是a,b都存在，則去掉後者第一個字符，也就是"/" 後拼接
        return a + b[1:]
    case !aslash && !bslash:  // 若是a,b都不存在，則在二者間添加"/"
        return a + "/" + b
    }
    return a + b  // 不然直接拼接到一塊
}

從上面的實例中咱們已經知道基本步驟是實例化一個reverseproxy對象，再傳入到http.ListenAndServe方法中

proxy := NewSingleHostReverseProxy(targetUrl)
http.ListenAndServe(":8888",proxy)

其中http.ListenAndServe 方法接收的是一個地址與handler，函數簽名以下：

func ListenAndServe(addr string, handler Handler) error {...}

這裏的handler 是一個接口，實現的方法是ServeHTTP

type Handler interface {
    ServeHTTP(ResponseWriter, *Request)
}

所以，咱們能夠確定實例化的reverseproxy對象也實現了ServeHTTP方法
主要步驟有：
一、拷貝上游請求的Header到下游請求
二、修改請求（例如協議、參數、url等）
三、判斷是否須要升級協議（Upgrade）
四、刪除上游請求中的hop-by-hop Header，即不須要透傳到下游的header
五、設置X-Forward-For Header，追加當前節點IP
六、使用鏈接池，向下遊發起請求
七、處理協議升級（httpcode 101）
八、刪除不須要返回給上游的逐跳Header
九、修改響應體內容（若有須要）
十、拷貝下游響應頭部到上游響應請求
十一、返回HTTP狀態碼
十二、定時刷新內容到response

下面咱們來分析下核心方法 serverHttp

func (p *ReverseProxy) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
    transport := p.Transport
    if transport == nil {
        transport = http.DefaultTransport
    }

    // 檢查請求是否被終止
    // 獲取請求的上下文，從responseWriter中獲取CloseNotify實例，起一個goroutine監聽notifyChan，收到請求結束通知後調用context cancel()方法
    // 關閉瀏覽器、網絡中斷、強行終止請求或是正常結束請求等都會收到請求結束通知
    ctx := req.Context()
    if cn, ok := rw.(http.CloseNotifier); ok {
        var cancel context.CancelFunc
        ctx, cancel = context.WithCancel(ctx)
        defer cancel()
        notifyChan := cn.CloseNotify()
        go func() {
            select {
            case <-notifyChan:
                cancel()
            case <-ctx.Done():
            }
        }()
    }

    // 設置context，這裏指的是想下游請求的request
    outreq := req.WithContext(ctx) // includes shallow copies of maps, but okay
    if req.ContentLength == 0 {
        outreq.Body = nil // Issue 16036: nil Body for http.Transport retries
    }

    // 深拷貝Header，即將上游的Header複製到下游request Header中
    outreq.Header = cloneHeader(req.Header)

    // 設置Director，修改request
    p.Director(outreq)
    outreq.Close = false

    // 升級http協議，HTTP Upgrade
    // 判斷header Connection 中是否有Upgrade
    reqUpType := upgradeType(outreq.Header)
    removeConnectionHeaders(outreq.Header)

    // Remove hop-by-hop headers to the backend. Especially
    // important is "Connection" because we want a persistent
    // connection, regardless of what the client sent to us.
    // 刪除 hop-by-hop headers，主要是一些規定的不須要向下遊傳遞的header
    for _, h := range hopHeaders {
        hv := outreq.Header.Get(h)
        if hv == "" {
            continue
        }
        // Te 和 trailers 這兩個Header 不作刪除處理
        if h == "Te" && hv == "trailers" {
            // Issue 21096: tell backend applications that
            // care about trailer support that we support
            // trailers. (We do, but we don't go out of
            // our way to advertise that unless the
            // incoming client request thought it was
            // worth mentioning)
            continue
        }
        outreq.Header.Del(h)
    }

    // After stripping all the hop-by-hop connection headers above, add back any
    // necessary for protocol upgrades, such as for websockets.
    // 若是reqUpType 不爲空，將Connection 、Upgrade值設置爲Upgrade ，例如websocket的場景
    if reqUpType != "" {
        outreq.Header.Set("Connection", "Upgrade")
        outreq.Header.Set("Upgrade", reqUpType)
    }

    // 設置X-Forwarded-For，追加節點IP
    if clientIP, _, err := net.SplitHostPort(req.RemoteAddr); err == nil {
        // If we aren't the first proxy retain prior
        // X-Forwarded-For information as a comma+space
        // separated list and fold multiple headers into one.
        if prior, ok := outreq.Header["X-Forwarded-For"]; ok {
            clientIP = strings.Join(prior, ", ") + ", " + clientIP
        }
        outreq.Header.Set("X-Forwarded-For", clientIP)
    }

    // 向下遊發起請求
    res, err := transport.RoundTrip(outreq)
    if err != nil {
        p.getErrorHandler()(rw, outreq, err)
        return
    }

    // Deal with 101 Switching Protocols responses: (WebSocket, h2c, etc)
    // 處理升級協議請求
    if res.StatusCode == http.StatusSwitchingProtocols {
        if !p.modifyResponse(rw, res, outreq) {
            return
        }
        p.handleUpgradeResponse(rw, outreq, res)
        return
    }
    // 刪除響應請求的逐跳 header
    removeConnectionHeaders(res.Header)

    for _, h := range hopHeaders {
        res.Header.Del(h)
    }

    // 修改響應內容
    if !p.modifyResponse(rw, res, outreq) {
        return
    }

    // 拷貝響應Header到上游
    copyHeader(rw.Header(), res.Header)

    // The "Trailer" header isn't included in the Transport's response,
    // at least for *http.Transport. Build it up from Trailer.
    announcedTrailers := len(res.Trailer)
    if announcedTrailers > 0 {
        trailerKeys := make([]string, 0, len(res.Trailer))
        for k := range res.Trailer {
            trailerKeys = append(trailerKeys, k)
        }
        rw.Header().Add("Trailer", strings.Join(trailerKeys, ", "))
    }
    // 寫入狀態碼
    rw.WriteHeader(res.StatusCode)

    // 週期刷新內容到response
    err = p.copyResponse(rw, res.Body, p.flushInterval(req, res))
    if err != nil {
        defer res.Body.Close()
        // Since we're streaming the response, if we run into an error all we can do
        // is abort the request. Issue 23643: ReverseProxy should use ErrAbortHandler
        // on read error while copying body.
        if !shouldPanicOnCopyError(req) {
            p.logf("suppressing panic for copyResponse error in test; copy error: %v", err)
            return
        }
        panic(http.ErrAbortHandler)
    }
    res.Body.Close() // close now, instead of defer, to populate res.Trailer
  ......
}

3 修改返回內容實例

核心在於修改 reverseproxy 中的ModifyResponse 方法中的響應體內容和內容長度

package main

import (
    "bytes"
    "fmt"
    "io/ioutil"
    "log"
    "net/http"
    "net/http/httputil"
    "net/url"
    "strings"
)

func main() {
    // 地址重寫實例
    // http://127.0.0.1:8888/test?id=1  =》 http://127.0.0.1:8081/reverse/test?id=1

    rs1 := "http://127.0.0.1:8081/reverse"
    targetUrl , err := url.Parse(rs1)
    if err != nil {
        log.Fatal("err")
    }
    proxy := NewSingleHostReverseProxy(targetUrl)
    log.Println("Reverse proxy server serve at : 127.0.0.1:8888")
    if err := http.ListenAndServe(":8888",proxy);err != nil{
        log.Fatal("Start server failed,err:",err)
    }
}

func singleJoiningSlash(a, b string) string {
    aslash := strings.HasSuffix(a, "/")
    bslash := strings.HasPrefix(b, "/")
    switch {
    case aslash && bslash:
        return a + b[1:]
    case !aslash && !bslash:
        return a + "/" + b
    }
    return a + b
}

func NewSingleHostReverseProxy(target *url.URL) *httputil.ReverseProxy {
    targetQuery := target.RawQuery
    director := func(req *http.Request) {
        req.URL.Scheme = target.Scheme
        req.URL.Host = target.Host
        req.URL.Path = singleJoiningSlash(target.Path, req.URL.Path)
        if targetQuery == "" || req.URL.RawQuery == "" {
            req.URL.RawQuery = targetQuery + req.URL.RawQuery
        } else {
            req.URL.RawQuery = targetQuery + "&" + req.URL.RawQuery
        }
        if _, ok := req.Header["User-Agent"]; !ok {
            // explicitly disable User-Agent so it's not set to default value
            req.Header.Set("User-Agent", "")
        }
    }

    // 自定義ModifyResponse
    modifyResp := func(resp *http.Response) error{
        var oldData,newData []byte
        oldData,err := ioutil.ReadAll(resp.Body)
        if err != nil{
            return err
        }
        // 根據不一樣狀態碼修改返回內容
        if resp.StatusCode == 200 {
            newData = []byte("[INFO] " + string(oldData))

        }else{
            newData = []byte("[ERROR] " + string(oldData))
        }

        // 修改返回內容及ContentLength
        resp.Body = ioutil.NopCloser(bytes.NewBuffer(newData))
        resp.ContentLength = int64(len(newData))
        resp.Header.Set("Content-Length",fmt.Sprint(len(newData)))
        return nil
    }
    // 傳入自定義的ModifyResponse
    return &httputil.ReverseProxy{Director: director,ModifyResponse:modifyResp}
}

測試結果

$ curl http://127.0.0.1:8888/test?id=123
[INFO] http://127.0.0.1:8081/reverse/test?id=123

4 返回客戶端真實IP

處於安全性的考慮，一般咱們不會將真實服務器也就是realserver 直接對外部用戶暴露，而是經過反向代理的方式對外暴露服務，以下圖所示：

帶來的問題是，在用戶與真實服務器之間通過一臺或多臺反向代理服務器後，真實服務器究竟應該如何獲取到用戶的真實IP，換句話說，中間的反向代理服務器應如何將用戶真實IP原封不動的透傳到後端真實服務器。

一般咱們會基於HTTP header實現，經常使用的有X-Real-IP 和 X-Forward-For 兩個字段。
X-Real-IP ： 一般在離用戶最近的代理點上設置，用於記錄用戶的真實IP，日後的反向代理節點不須要設置，不然將覆蓋爲上一個反向代理的IP
X-Forward-For：記錄每一個通過的節點IP，以","分隔，例如請求鏈路是client -> proxy1 -> proxy2 -> webapp，那麼獲得的值爲clientip,proxy1,proxy2

if clientIP, _, err := net.SplitHostPort(req.RemoteAddr); err == nil {
        // If we aren't the first proxy retain prior
        // X-Forwarded-For information as a comma+space
        // separated list and fold multiple headers into one.
        if prior, ok := outreq.Header["X-Forwarded-For"]; ok {
            clientIP = strings.Join(prior, ", ") + ", " + clientIP
        }
        outreq.Header.Set("X-Forwarded-For", clientIP)
    }