go http請求轉發

go http請求轉發

1.說明

• 平常開發中會遇到須要將請求轉發到其它服務器的需求：

  • 1.若是是前端進行轉發，須要解決跨域的問題；
  • 2.後端轉發到目標服務器，並返回數據到client；

咱們只討論後端如何處理轉發。

2. 原理

• 轉發須要對數據流解決的問題：

  • 1.向目標服務器發送請求，並獲取數據
  • 2.將數據返回到client
  • 3.對於client整個過程透明，不被感知到請求被轉發

3. 方案

• 解決方案：

  • 1.服務端使用tcp鏈接模擬http請求
  • 2.使用標準庫提供的設置項進行代理

• 實現

• • 1.tcp模擬轉發

    • 先和目標服務器創建tcp鏈接
    • 開啓goroutinue，而後將請求數據發送到目標服務器，同時將接受到的數據進行回寫
    • 鏈接採用鏈接池的方式進行處理，代碼較爲簡單，不做過多贅述
    • 此方法由於徹底屏蔽了http細節，因此有數據讀取超時問題，其實就是不知道數據流是否讀寫結束。在http協議中通常經過Content-Length可知數據體長度，可是在未設置此頭部信息時（流傳輸），就很難肯定數據是否讀寫結束。看到有文提出經過判斷最後的「\r\n\r\n\r\n\r\n」來肯定結束，可是這並不嚴謹。在沒有絕對的協商約束下，會不經意的截斷body中內容，致使數據丟失。
    • 可採用設置讀寫超時來結束讀寫操做阻塞的問題，可是時間設置長短可能影響併發性能。（SetDeadLine，SetReadDeadLine，SetWriteDeadLine）

    package proxy
     
     import (
         "io"
         "log"
         "net"
         "sync"
         "sync/atomic"
         "time"
     )
     
     /**
     封裝代理服務，對於http鏈接反饋又超時處理，注意超時問題
     */
     
     var pool = make(chan net.Conn, 100)
     
     type conn struct {
         conn  net.Conn
         wg    *sync.WaitGroup
         lock  sync.Mutex
         state int32
     }
     
     const (
         maybeValid = iota
         isValid
         isInvalid
         isInPool
         isClosed
     )
     
     type timeoutErr interface {
         Timeout() bool
     }
     
     func isTimeoutError(err error) bool {
         timeoutErr, _ := err.(timeoutErr)
         if timeoutErr == nil {
             return false
         }
         return timeoutErr.Timeout()
     }
     
     func (cn *conn) Read(b []byte) (n int, err error) {
         n, err = cn.conn.Read(b)
         if err != nil {
             if !isTimeoutError(err) {
                 atomic.StoreInt32(&cn.state, isInvalid)
             }
         } else {
             atomic.StoreInt32(&cn.state, isValid)
         }
         return
     }
     
     func (cn *conn) Write(b []byte) (n int, err error) {
         n, err = cn.conn.Write(b)
         if err != nil {
             if !isTimeoutError(err) {
                 atomic.StoreInt32(&cn.state, isInvalid)
             }
         } else {
             atomic.StoreInt32(&cn.state, isValid)
         }
         return
     }
     
     func (cn *conn) Close() error {
         atomic.StoreInt32(&cn.state, isClosed)
         return cn.conn.Close()
     }
     
     func getConn() (*conn, error) {
         var cn net.Conn
         var err error
         select {
         case cn = <-pool:
             //service.Logger.Info().Msg("get conn from pool")
         default:
             cn, err = net.Dial("tcp", "127.0.0.1:8090")
             //service.Logger.Info().Msg("get conn by new")
         }
         if err != nil {
             service.Logger.Error().Err(err).Msgf("dial to dest %s failed ", "127.0.0.1:8090")
             return nil, err
         }
         return &conn{
             conn:  cn,
             wg:    &sync.WaitGroup{},
             state: maybeValid,
         }, nil
     }
     
     func release(cn *conn) error {
         state := atomic.LoadInt32(&cn.state)
         switch state {
         case isInPool, isClosed:
             return nil
         case isInvalid:
             return cn.conn.Close()
         }
         cn.lock.Lock()
         defer cn.lock.Unlock()
         select {
         case pool <- cn.conn:
             //service.Logger.Info().Msgf("%d  %d put conn to pool",os.Getpid(),os.Getppid())
             atomic.StoreInt32(&cn.state, isInPool)
             return nil
         default:
             return cn.Close()
         }
     }
     
     func Handle(conn net.Conn) {
         if conn == nil {
             return
         }
         defer conn.Close()
         conn.SetDeadline(time.Now().Add(time.Millisecond * 100))  //設置讀寫超時
         client, err := getConn()
         if err != nil {
             return
         }
     
         defer release(client)
         client.conn.SetDeadline(time.Now().Add(time.Millisecond * 100)) //設置讀寫超時
     
         client.wg.Add(2)
         //進行轉發
         go func() {
             if _, err := io.Copy(client, conn); err != nil {
                 service.Logger.Err(err).Msg("copy data to svr")
             }
             client.wg.Done()
         }()
         go func() {
             if _, err := io.Copy(conn, client); err != nil {
                 service.Logger.Err(err).Msg("copy data to conn")
             }
             client.wg.Done()
         }()
     
         client.wg.Wait()
     }
     
     func StartProxySvr() <-chan struct{} {
         exit := make(chan struct{}, 1)
         proxy_server, err := net.Listen("tcp", "8889")
         if err != nil {
             log.Printf("proxy server listen error: %v\n", err)
             exit <- struct{}{}
             return exit
         }
     
         for {
             conn, err := proxy_server.Accept()
             if err != nil {
                 log.Printf("proxy server accept error: %v\n", err)
                 exit <- struct{}{}
                 return exit
             }
             go Handle(conn)
         }
     }

• 2.使用原生提供的http代理

  • http.Client中的Transport可用來設置目標服務的addr

  • 詳細內容請看源碼說明，下文提供一箇中間件樣例來進行請求轉發

    type Client struct {
         // Transport specifies the mechanism by which individual
         // HTTP requests are made.
         // If nil, DefaultTransport is used.
         Transport RoundTripper
     
         // CheckRedirect specifies the policy for handling redirects.
         // If CheckRedirect is not nil, the client calls it before
         // following an HTTP redirect. The arguments req and via are
         // the upcoming request and the requests made already, oldest
         // first. If CheckRedirect returns an error, the Client's Get
         // method returns both the previous Response (with its Body
         // closed) and CheckRedirect's error (wrapped in a url.Error)
         // instead of issuing the Request req.
         // As a special case, if CheckRedirect returns ErrUseLastResponse,
         // then the most recent response is returned with its body
         // unclosed, along with a nil error.
         //
         // If CheckRedirect is nil, the Client uses its default policy,
         // which is to stop after 10 consecutive requests.
         CheckRedirect func(req *Request, via []*Request) error
     
         // Jar specifies the cookie jar.
         //
         // The Jar is used to insert relevant cookies into every
         // outbound Request and is updated with the cookie values
         // of every inbound Response. The Jar is consulted for every
         // redirect that the Client follows.
         //
         // If Jar is nil, cookies are only sent if they are explicitly
         // set on the Request.
         Jar CookieJar
     
         // Timeout specifies a time limit for requests made by this
         // Client. The timeout includes connection time, any
         // redirects, and reading the response body. The timer remains
         // running after Get, Head, Post, or Do return and will
         // interrupt reading of the Response.Body.
         //
         // A Timeout of zero means no timeout.
         //
         // The Client cancels requests to the underlying Transport
         // as if the Request's Context ended.
         //
         // For compatibility, the Client will also use the deprecated
         // CancelRequest method on Transport if found. New
         // RoundTripper implementations should use the Request's Context
         // for cancelation instead of implementing CancelRequest.
         Timeout time.Duration
     }
     
     //中間件樣例
         http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
                 proxy := func(_ *http.Request) (*url.URL, error) {
                     return url.Parse("target ip:port")//127.0.0.1:8099
                 }
                 transport := &http.Transport{
                     Proxy: proxy,
                     DialContext: (&net.Dialer{
                         Timeout:   30 * time.Second,
                         KeepAlive: 30 * time.Second,
                         DualStack: true,
                     }).DialContext,
                     MaxIdleConns:          100,
                     IdleConnTimeout:       90 * time.Second,
                     TLSHandshakeTimeout:   10 * time.Second,
                     ExpectContinueTimeout: 1 * time.Second,
                     MaxIdleConnsPerHost:   100,
                 }
     
                 client := &http.Client{Transport: transport}
                 url := "http://" + r.RemoteAddr + r.RequestURI
                 req, err := http.NewRequest(r.Method, url, r.Body)
                 //注： 設置Request頭部信息
                 for k, v := range r.Header {
                     for _, vv := range v {
                         req.Header.Add(k, vv)
                     }
                 }
     
                 resp, err := client.Do(req)
                 if err != nil {
                     return
                 }
                 defer resp.Body.Close()
                 //注： 設置Response頭部信息
                 for k, v := range resp.Header {
                     for _, vv := range v {
                         w.Header().Add(k, vv)
                     }
                 }
                 data, _ := ioutil.ReadAll(resp.Body)
                 w.Write(data)
     
         })

結束

本文是我的對工做中遇到的問題的總結，不夠全面和深刻還請多多指教。謝謝！