HTTP/2 in GO(四)

女主宣言

上篇文章我们了解了如何在HTTP/2 server端进行Header信息的发送，同时保持连接不断开。这次我们在这个基础上，实现自动下发PUSH。 本文来自公众号“ 360搜索技术团队 ”的投稿，作者付坤。

PS：丰富的一线技术、多元化的表现形式，尽在“ 3 60云计算 ”，点关注哦！

• HTTP/2 in GO(一)

• HTTP/2 in GO(二)

• HTTP/2 in GO(三)

Start

上篇文章我们了解了如何在HTTP/2 server端进行Header信息的发送，同时保持连接不断开。这次我们在这个基础上，实现自动下发 PUSH 。

先来实现一个最简单的 Server Push 的例子, 我们在上次的demo基础上继续改进

package main

import (
    "html/template"
    "log"
    "net/http"
)

func main() {
    http.HandleFunc("/header", func(w http.ResponseWriter, r *http.Request) {
        w.Header().Add("X-custom-header", "custom header")
        w.WriteHeader(http.StatusNoContent)

        if f, ok := w.(http.Flusher); ok {
            f.Flush()
        }
        select {}
    })

    // 用于push的 handler
    http.HandleFunc("/crt", func(w http.ResponseWriter, r *http.Request) {
        tpl := template.Must(template.ParseFiles("server.crt"))
        tpl.Execute(w, nil)
    })

    // 请求该Path会触发Push
    http.HandleFunc("/push", func(w http.ResponseWriter, r *http.Request) {
        pusher, ok := w.(http.Pusher)
        if !ok {
            log.Println("not support server push")
        } else {
            err := pusher.Push("/crt", nil)
            if err != nil {
                log.Printf("Failed for server push: %v", err)
            }
        }
        w.WriteHeader(http.StatusOK)
    })

    log.Println("start listen on 8080...")
    log.Fatal(http.ListenAndServeTLS(":8080", "server.crt", "server.key", nil))
}

以上代码添加了两个 Hanlder ，一个是  /crt ，返回我们的证书内容，这个是用来给做客户端push的内容。另一个是  /push ，请求该链接时，我们会将  /crt 的内容主动  push 到客户端。

GO服务启动后，我们通过h2c来访问下 /push ：

先在一个终端通过  h2c start -d 启动进行输出显示，然后另外开一个终端窗口发起请求  h2c connect localhost:8080 和  h2c get /push :

来解读下这个请求中都发生了什么:

1. 客户端通过 stream id=1 发送  HEADERS FRAME 进行请求，请求Path是  /push

2. 服务端在 stream id=1 中返回一个  PUSH_PROMISE (配合下表食用) ,携带了部分  Header 信息，承诺会在  stream id=2 中返回  path: /crt 的相关信息，这里相当于告诉客户端，如果你接下来需要请求  /crt 的时候，就不要请求了，这个内容我一会就给你发过去了。

3. 服务端正常响应 get /push 的请求，返回了对应的  Header 信息，并通过  END_STREAM 表示此  stream 的交互完成了。

4. 服务端通过 stream id=2 下发  /crt 的相关信息，第四步是返回的  Header 信息.

5. 服务端通过 stream id=2 下发  /crt 的相关  DATA 信息, 并通过  END_STREAM 表示承诺的  /crt 的内容发送完毕。

 // PUSH_PROMISE Frame结构
 +---------------+
 |Pad Length? (8)|
 +-+-------------+-----------------------------------------------+
 |R|                  Promised Stream ID (31)                    |
 +-+-----------------------------+-------------------------------+
 |                   Header Block Fragment (*)                 ...
 +---------------------------------------------------------------+
 |                           Padding (*)                       ...
 +---------------------------------------------------------------+

通过这个例子，我们应该就掌握了 Server Push 的用法，在此基础上，我们结合上一章讲到的内容，再改进一下，实现 "服务端定时主动PUSH":

// 服务端定时 "主动" push内容
http.HandleFunc("/autoPush", func(w http.ResponseWriter, r *http.Request) {
    w.Header().Add("X-custom-header", "custom")
    w.WriteHeader(http.StatusNoContent)

    if f, ok := w.(http.Flusher); ok {
        f.Flush()
    }
    pusher, ok := w.(http.Pusher)
    if ok {
        for {
            select {
            case <-time.Tick(5 * time.Second):
                err := pusher.Push("/crt", nil)
                if err != nil {
                    log.Printf("Failed for server push: %v", err)
                }
            }
        }
    }
})

效果如图：

服务端一直发送 PUSH_PROMISE 消息给客户端，每次间隔5s，并且每次  Promised Strea Id 都在偶数范围内进行递增  2,4,6,8,10…

这个例子里，我们用了一个 for 循环 和一个定时器  time.Tick ，在服务端返回不带  END_STREAM 的  Headers 后，每隔5s向客户端主动  Push 一个内容，这里我们  Push 的内容是固定的，在实际应用场景中，可以从一个特定的  channel 中取出需要下发的消息，然后再动态的构造请求的path，可以是携带参数的，来实现动态的控制需要  Push 什么内容。这样就实现了 "服务端主动PUSH" 的功能。

HTTP/2 PUSH in Go

接下来看下 Server Push 在 Go 中的实现：

// Push implements http.Pusher.
func (w *http2responseWriter) Push(target string, opts *PushOptions) error {
    internalOpts := http2pushOptions{}
    if opts != nil {
        internalOpts.Method = opts.Method
        internalOpts.Header = opts.Header
    }
    return w.push(target, internalOpts)
}

func (w *http2responseWriter) push(target string, opts http2pushOptions) error {
    // ...
    // Push只能是对 GET or HEAD 方法
    if opts.Method != "GET" && opts.Method != "HEAD" {
        return fmt.Errorf("method %q must be GET or HEAD", opts.Method)
    }
    // 构造要Push的内容的请求
    msg := &http2startPushRequest{
        parent: st,
        method: opts.Method,
        url:    u,
        header: http2cloneHeader(opts.Header),
        done:   http2errChanPool.Get().(chan error),
    }
    // 在客户端连接断开或者END_STREAM之前可以发送PUSH，把构造好的PushRequest放到 sc.serveMsgCh channel 里
    select {
        case <-sc.doneServing:
            return http2errClientDisconnected
        case <-st.cw:
            return http2errStreamClosed
        case sc.serveMsgCh <- msg:
    }
}

// 在serve中会 取出 sc.serveMsgCh 中的消息进行对应的操作，当取到 PushRequest 时，就会发送Push消息
func (sc *http2serverConn) serve() {
    // ...
    loopNum := 0
        for {
            loopNum++
            select {
                // ...
                case msg := <-sc.serveMsgCh:
                    switch v := msg.(type) {
                        // ...
                        case *http2startPushRequest:
                            sc.startPush(v)
                        // ...
                    }
            }
        }
}

func (sc *http2serverConn) startPush(msg *http2startPushRequest) {
    // ...
    // 获取Prosise的Stream id，当真正要发送PUSH_PROMISE时才进行获取，并且同时异步启动需要Push的Handler的请求.
    allocatePromisedID := func() (uint32, error) {
        // ...
        sc.maxPushPromiseID += 2
        promisedID := sc.maxPushPromiseID
        // 新建Stream用于push内容的发送
        promised := sc.newStream(promisedID, msg.parent.id, http2stateHalfClosedRemote)
        rw, req, err := sc.newWriterAndRequestNoBody(promised, http2requestParam{
            method:    msg.method,
            scheme:    msg.url.Scheme,
            authority: msg.url.Host,
            path:      msg.url.RequestURI(),
            header:    http2cloneHeader(msg.header), 
        })
        // ...

        // 进行handle请求
        go sc.runHandler(rw, req, sc.handler.ServeHTTP)
        return promisedID, nil
    }
    // 构造好 PUSH_PROMISE, 开始发送
    sc.writeFrame(http2FrameWriteRequest{
        write: &http2writePushPromise{
            streamID:           msg.parent.id,
            method:             msg.method,
            url:                msg.url,
            h:                  msg.header,
            allocatePromisedID: allocatePromisedID,
        },
        stream: msg.parent,
        done:   msg.done,
    })
}

Done.

360云计算

由360云平台团队打造的技术分享公众号，内容涉及 数据库、大数据、微服务、容器、AIOps、IoT 等众多技术领域，通过夯实的技术积累和丰富的一线实战经验，为你带来最有料的技术分享