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)
         
             })

结束

本文是个人对工作中遇到的问题的总结,不够全面和深入还请多多指教。谢谢!

    原文作者:Zero
    原文地址: https://segmentfault.com/a/1190000019577872
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞