golang中http请求的context传递到异步任务的坑及解决
前言
在golang中,context.Context可以用来用来设置截止日期、同步信号,传递请求相关值的结构体。 与 goroutine 有比较密切的关系。
在web程序中,每个Request都需要开启一个goroutine做一些事情,这些goroutine又可能会开启其他的 goroutine去访问后端资源,比如数据库、RPC服务等,它们需要访问一些共享的资源,比如用户身份信息、认证token、请求截止时间等 这时候可以通过Context,来跟踪这些goroutine,并且通过Context来控制它们, 这就是Go语言为我们提供的Context,中文可以理解为“上下文”。
简单看一下Context结构
type Context interface { Deadline() (deadline time.Time, ok bool) Done() <-chan struct{} Err() error Value(key interface{}) interface{} }
- Deadline方法是获取设置的截止时间的意思,第一个返回值是截止时间,到了这个时间点,Context会自动发起取消请求; 第二个返回值ok==false时表示没有设置截止时间,如果需要取消的话,需要调用取消函数(CancleFunc)进行取消。
- Done方法返回一个只读的chan,类型为struct{},在goroutine中,如果该方法返回的chan可以读取,则意味着parent context已经发起了取消请求, 我们通过Done方法收到这个信号后,就应该做清理操作,然后退出goroutine,释放资源。之后,Err 方法会返回一个错误,告知为什么 Context 被取消。
- Err方法返回取消的错误原因,Context被取消的原因。
- Value方法获取该Context上绑定的值,是一个键值对,通过一个Key才可以获取对应的值,这个值一般是线程安全的。
常用的
// 传递一个父Context作为参数,返回子Context,以及一个取消函数用来取消Context。 func WithCancel(parent Context) (ctx Context, cancel CancelFunc) // 和WithCancel差不多,它会多传递一个截止时间参数,意味着到了这个时间点,会自动取消Context, // 当然我们也可以不等到这个时候,可以提前通过取消函数进行取消。 func WithDeadline(parent Context, deadline time.Time) (Context, CancelFunc) // WithTimeout和WithDeadline基本上一样,这个表示是超时自动取消,是多少时间后自动取消Context的意思 func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc) //WithValue函数和取消Context无关,它是为了生成一个绑定了一个键值对数据的Context, // 绑定的数据可以通过Context.Value方法访问到,这是我们实际用经常要用到的技巧,一般我们想要通过上下文来传递数据时,可以通过这个方法, // 如我们需要tarce追踪系统调用栈的时候。 func WithValue(parent Context, key, val interface{}) Context
HTTP请求的Context传递到异步任务的坑
看下面例子
我们将http的context传递到goroutine 中:
package main import ( "context" "fmt" "net/http" "time" ) func IndexHandler(resp http.ResponseWriter, req *http.Request) { ctx := req.Context() go func(ctx context.Context) { for { select { case <-ctx.Done(): fmt.Println("gorountine off,the err is: ", ctx.Err()) return default: fmt.Println(333) } } }(ctx) time.Sleep(1000) resp.Write([]byte{1}) } func main() { http.HandleFunc("/test1", IndexHandler) http.ListenAndServe("127.0.0.1:8080", nil) }
结果:
从上面结果来看,在http请求返回之后,传入gorountine的context被cancel掉了,如果不巧,你在gorountine中进行一些http调用或者rpc调用传入了这个context,那么对应的请求也将会被cancel掉。
因此,在http请求中异步任务出去时,如果这个异步任务中需要进行一些rpc类请求,那么就不要直接使用或者继承http的context,否则将会被cancel。
纠其原因
http请求再结束后,将会cancel掉这个context,所以异步出去的请求中收到的context是被cancel掉的。
下面来看下源代码:
ListenAndServe–>Server:Server方法中有一个大的for循环,这个for循环中,针对每个请求,都会起一个协程进行处理。
serve方法处理一个连接中的请求,并在一个请求serverHandler{c.server}.ServeHTTP(w, w.req)
结束后cancel掉对应的context:
// Serve a new connection. func (c *conn) serve(ctx context.Context) { c.remoteAddr = c.rwc.RemoteAddr().String() ctx = context.WithValue(ctx, LocalAddrContextKey, c.rwc.LocalAddr()) defer func() { if err := recover(); err != nil && err != ErrAbortHandler { const size = 64 << 10 buf := make([]byte, size) buf = buf[:runtime.Stack(buf, false)] c.server.logf("http: panic serving %v: %v\n%s", c.remoteAddr, err, buf) } if !c.hijacked() { c.close() c.setState(c.rwc, StateClosed, runHooks) } }() if tlsConn, ok := c.rwc.(*tls.Conn); ok { if d := c.server.ReadTimeout; d != 0 { c.rwc.SetReadDeadline(time.Now().Add(d)) } if d := c.server.WriteTimeout; d != 0 { c.rwc.SetWriteDeadline(time.Now().Add(d)) } if err := tlsConn.Handshake(); err != nil { // If the handshake failed due to the client not speaking // TLS, assume they're speaking plaintext HTTP and write a // 400 response on the TLS conn's underlying net.Conn. if re, ok := err.(tls.RecordHeaderError); ok && re.Conn != nil && tlsRecordHeaderLooksLikeHTTP(re.RecordHeader) { io.WriteString(re.Conn, "HTTP/1.0 400 Bad Request\r\n\r\nClient sent an HTTP request to an HTTPS server.\n") re.Conn.Close() return } c.server.logf("http: TLS handshake error from %s: %v", c.rwc.RemoteAddr(), err) return } c.tlsState = new(tls.ConnectionState) *c.tlsState = tlsConn.ConnectionState() if proto := c.tlsState.NegotiatedProtocol; validNextProto(proto) { if fn := c.server.TLSNextProto[proto]; fn != nil { h := initALPNRequest{ctx, tlsConn, serverHandler{c.server}} // Mark freshly created HTTP/2 as active and prevent any server state hooks // from being run on these connections. This prevents closeIdleConns from // closing such connections. See issue https://golang.org/issue/39776. c.setState(c.rwc, StateActive, skipHooks) fn(c.server, tlsConn, h) } return } } // HTTP/1.x from here on. ctx, cancelCtx := context.WithCancel(ctx) c.cancelCtx = cancelCtx defer cancelCtx() c.r = &connReader{conn: c} c.bufr = newBufioReader(c.r) c.bufw = newBufioWriterSize(checkConnErrorWriter{c}, 4<<10) for { // 从连接中读取请求 w, err := c.readRequest(ctx) if c.r.remain != c.server.initialReadLimitSize() { // If we read any bytes off the wire, we're active. c.setState(c.rwc, StateActive, runHooks) } ..... ..... // Expect 100 Continue support req := w.req if req.expectsContinue() { if req.ProtoAtLeast(1, 1) && req.ContentLength != 0 { // Wrap the Body reader with one that replies on the connection req.Body = &expectContinueReader{readCloser: req.Body, resp: w} w.canWriteContinue.setTrue() } } else if req.Header.get("Expect") != "" { w.sendExpectationFailed() return } c.curReq.Store(w) // 启动协程后台读取连接 if requestBodyRemains(req.Body) { registerOnHitEOF(req.Body, w.conn.r.startBackgroundRead) } else { w.conn.r.startBackgroundRead() } // HTTP cannot have multiple simultaneous active requests.[*] // Until the server replies to this request, it can't read another, // so we might as well run the handler in this goroutine. // [*] Not strictly true: HTTP pipelining. We could let them all process // in parallel even if their responses need to be serialized. // But we're not going to implement HTTP pipelining because it // was never deployed in the wild and the answer is HTTP/2. serverHandler{c.server}.ServeHTTP(w, w.req) /** * 重点在这儿,处理完请求后将会调用w.cancelCtx()方法cancel掉context **/ w.cancelCtx() if c.hijacked() { return } w.finishRequest() if !w.shouldReuseConnection() { if w.requestBodyLimitHit || w.closedRequestBodyEarly() { c.closeWriteAndWait() } return } c.setState(c.rwc, StateIdle, runHooks) c.curReq.Store((*response)(nil)) if !w.conn.server.doKeepAlives() { // We're in shutdown mode. We might've replied // to the user without "Connection: close" and // they might think they can send another // request, but such is life with HTTP/1.1. return } if d := c.server.idleTimeout(); d != 0 { c.rwc.SetReadDeadline(time.Now().Add(d)) if _, err := c.bufr.Peek(4); err != nil { return } } c.rwc.SetReadDeadline(time.Time{}) } }
至此,我们知道,http请求在正常结束后将会主动cancel掉context。
此外,在请求异常时候也会主动cancel掉context(cancel目的就是为了快速失败),具体可见w.conn.r.startBackgroundRead()
其中的实现。
在日常开发中,我们知道有时候会存在客户端超时情况,和ctx相关的原因可归纳如下:
- 服务端收到的请求的request context被cancel掉。
- 客户端本身收到context deadline exceeded错误
- 服务端业务业务使用了http的context,但没有用于做rpc等需要建立连接的任务,那么客户端即使收到了context canceled的错误,服务端实际上还是在继续执行业务代码。
- 服务端业务业务使用了http的context,并用于做rpc等需要建立连接的任务,那么客户端收到context canceled错误,并且服务端也会在对应的rpc等建立连接任务处返回context cancled的错误。
最后,如果context cancel掉了,但是业务又在继续执行,有时候并不是我们想要的结果,因为这会占用资源,因此我们可以主动在业务中通过监听context Done的信号来做context canceled的处理,从而可以达到快速失败,节约资源的目的。
总结
以上为个人经验,希望能给大家一个参考,也希望大家多多支持脚本之家。
相关文章
golang中使用proto3协议导致的空值字段不显示的问题处理方案
这篇文章主要介绍了golang中使用proto3协议导致的空值字段不显示的问题处理方案,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧2019-10-10Golang微服务框架Kratos实现Kafka消息队列的方法
消息队列是大型分布式系统不可缺少的中间件,也是高并发系统的基石中间件,所以掌握好消息队列MQ就变得极其重要,在本文当中,您将了解到:什么是消息队列?什么是Kafka?怎样在微服务框架Kratos当中应用Kafka进行业务开发,需要的朋友可以参考下2023-09-09
最新评论