Fasthttp 是什么？
官方介绍：fasthttp是为一些高性能边缘情况而设计的。除非您的服务器/客户端需要处理数千个每秒的小到中等请求并需要一致的低毫秒响应时间，否则fasthttp可能不适合您。对于大多数情况，net/http更好，因为它更易于使用并且可以处理更多情况。对于大多数情况，您甚至不会注意到性能差异。
目前，VertaMedia 在生产环境中成功地使用fasthttp，每个物理服务器可以提供高达200K rps的服务，并支持超过1.5M个并发keep-alive连接。
Fasthttp 为什么比 net/http 快？
- 使用连接池，不再像 net/http 一样每accept一个请求就分配一个新的goroutine处理请求（在请求量大的时候会有性能问题）
- 对象池化
- 使用 []byte 缓冲池，减少内存分配
- 避免 []byte 和 string 之间的转化，[]byte 和 string 和转化会产生内存拷贝，fasthttp 实现了无需内存拷贝的转化方法，具体见代码 b2s 和 s2b
源码解析
1
2
3
4
5
6
7
8
9
func main() {
if err := fasthttp.ListenAndServe(":8088", requestHandler); err != nil {
log.Fatalf("Error in ListenAndServe: %s", err)
}
}
func requestHandler(ctx *fasthttp.RequestCtx) {
fmt.Fprintf(ctx, "Hello, world!\n\n")
}
ListenAndServe 工作流程
- 监听端口，获取 listen 连接，这个和 net/http 一致
- 循环监听用户请求
- 获取到连接之后首先会去 ready 队列里获取 workerChan，获取不到就会去对象池获取
- 将监听的连接传入到 workerChan 的 channel 中
- 有一个协程执行 workerFunc 循环读取 workerChan 的 channel，获取到连接对象后会对请求进行处理
ListenAndServer 方法
1
2
3
4
5
6
7
8
// Accepted connections are configured to enable TCP keep-alives.
func (s *Server) ListenAndServe(addr string) error {
ln, err := net.Listen("tcp4", addr)
if err != nil {
return err
}
return s.Serve(ln)
}
获取 listen 监听对象，调用 server 进行处理
Server 方法
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
// 为了便于分析只保留了核心代码
// Serve serves incoming connections from the given listener.
//
// Serve blocks until the given listener returns permanent error.
func (s *Server) Serve(ln net.Listener) error {
var lastOverflowErrorTime time.Time
var lastPerIPErrorTime time.Time
var c net.Conn
var err error
// 获取最大并发数，默认是 256 * 1024个
maxWorkersCount := s.getConcurrency()
...
// 初始化 workerPool
wp := &workerPool{
WorkerFunc: s.serveConn, // 设置连接处理函数
MaxWorkersCount: maxWorkersCount, // 设置最大并发数
LogAllErrors: s.LogAllErrors,
MaxIdleWorkerDuration: s.MaxIdleWorkerDuration, // 设置单个连接对象的最大空闲时间
Logger: s.logger(),
connState: s.setState,
}
// 启动 workpool ，里面 有一个 clean 方法，默认每10s清理一下超过最大空闲时间的链接
wp.Start()
// Count our waiting to accept a connection as an open connection.
// This way we can't get into any weird state where just after accepting
// a connection Shutdown is called which reads open as 0 because it isn't
// incremented yet.
atomic.AddInt32(&s.open, 1)
defer atomic.AddInt32(&s.open, -1)
for {
// accept 请求，这里和 net/http 一致
if c, err = acceptConn(s, ln, &lastPerIPErrorTime); err != nil {
wp.Stop()
if err == io.EOF {
return nil
}
return err
}
s.setState(c, StateNew)
atomic.AddInt32(&s.open, 1)
// 处理连接
if !wp.Serve(c) {
...
}
c = nil
}
}
首先初始化 workpool ，设置最大并发数，连接处理函数，空闲连接清理任务。然后是获取用户请求，最后交由
Serve 方法对请求进行处理，如果超过最大并发数则返回503。
wp.serve 方法
1
2
3
4
5
6
7
8
9
10
func (wp *workerPool) Serve(c net.Conn) bool {
// 获取连接
ch := wp.getCh()
if ch == nil {
return false
}
// 把请求对象传递给channel，连接处理函数对连接进行处理
ch.ch <- c
return true
}
wp.getCh() 方法
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
func (wp *workerPool) getCh() *workerChan {
var ch *workerChan
createWorker := false
wp.lock.Lock()
// 获取空闲连接
ready := wp.ready
n := len(ready) - 1
if n < 0 {
if wp.workersCount < wp.MaxWorkersCount {
createWorker = true
wp.workersCount++
}
} else {
ch = ready[n]
ready[n] = nil
wp.ready = ready[:n]
}
wp.lock.Unlock()
if ch == nil {
if !createWorker {
return nil
}
// 空闲连接为空则从对象池获取
vch := wp.workerChanPool.Get()
ch = vch.(*workerChan)
// 启动协程调用处理函数，这个函数用于对连接对象进行处理
// 处理完成后把对象放回对象池
go func() {
wp.workerFunc(ch)
wp.workerChanPool.Put(vch)
}()
}
return ch
}
getCh 先从空闲队列获取 workerChan，获取不到则从对象池获取。并且开启一个协程调用 workerFunc 方法，该方法 阻塞等待连接对象，并进行后续请求的处理。
wp.workerFunc 方法
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
func (wp *workerPool) workerFunc(ch *workerChan) {
var c net.Conn
var err error
// 读取 ch
for c = range ch.ch {
// 这个 nil 是 workerPool 在异步调用 clean 方法检查该 workerChan 空闲时间超长了就会往 channel 中传 // 入一个 nil
if c == nil {
break
}
// 调用初始化时候注册的处理函数
if err = wp.WorkerFunc(c); err != nil && err != errHijacked {
errStr := err.Error()
if wp.LogAllErrors || !(strings.Contains(errStr, "broken pipe") ||
strings.Contains(errStr, "reset by peer") ||
strings.Contains(errStr, "request headers: small read buffer") ||
strings.Contains(errStr, "unexpected EOF") ||
strings.Contains(errStr, "i/o timeout") ||
errors.Is(err, ErrBadTrailer)) {
wp.Logger.Printf("error when serving connection %q<->%q: %v", c.LocalAddr(), c.RemoteAddr(), err)
}
}
if err == errHijacked {
wp.connState(c, StateHijacked)
} else {
_ = c.Close()
wp.connState(c, StateClosed)
}
c = nil
// 请求完成workerChan放回空闲队列
if !wp.release(ch) {
break
}
}
wp.lock.Lock()
wp.workersCount--
wp.lock.Unlock()
}
func (wp *workerPool) release(ch *workerChan) bool {
ch.lastUseTime = time.Now()
wp.lock.Lock()
if wp.mustStop {
wp.lock.Unlock()
return false
}
wp.ready = append(wp.ready, ch)
wp.lock.Unlock()
return true
}
等待连接对象到来调用 WorkerFunc 对连接进行处理，处理完成后释放 workerChan 对象。
wp.WorkerFunc(c) 对应注册的 serveConn 方法
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
func (s *Server) serveConn(c net.Conn) (err error) {
defer s.serveConnCleanup()
atomic.AddUint32(&s.concurrency, 1)
var proto string
if proto, err = s.getNextProto(c); err != nil {
return
}
if handler, ok := s.nextProtos[proto]; ok {
// Remove read or write deadlines that might have previously been set.
// The next handler is responsible for setting its own deadlines.
if s.ReadTimeout > 0 || s.WriteTimeout > 0 {
if err := c.SetDeadline(zeroTime); err != nil {
panic(fmt.Sprintf("BUG: error in SetDeadline(zeroTime): %v", err))
}
}
return handler(c)
}
serverName := s.getServerName()
connRequestNum := uint64(0)
connID := nextConnID()
connTime := time.Now()
maxRequestBodySize := s.MaxRequestBodySize
if maxRequestBodySize <= 0 {
maxRequestBodySize = DefaultMaxRequestBodySize
}
writeTimeout := s.WriteTimeout
previousWriteTimeout := time.Duration(0)
// 从对象池获取 ctx
// fasthttp 通过使用对象池减少内存的创建与回收从而提高程序性能
ctx := s.acquireCtx(c)
ctx.connTime = connTime
isTLS := ctx.IsTLS()
var (
br *bufio.Reader
bw *bufio.Writer
timeoutResponse *Response
hijackHandler HijackHandler
hijackNoResponse bool
connectionClose bool
continueReadingRequest = true
)
for {
connRequestNum++
// If this is a keep-alive connection set the idle timeout.
if connRequestNum > 1 {
if d := s.idleTimeout(); d > 0 {
if err := c.SetReadDeadline(time.Now().Add(d)); err != nil {
break
}
}
}
if !s.ReduceMemoryUsage || br != nil {
if br == nil {
br = acquireReader(ctx)
}
// If this is a keep-alive connection we want to try and read the first bytes
// within the idle time.
if connRequestNum > 1 {
var b []byte
b, err = br.Peek(1)
if len(b) == 0 {
// If reading from a keep-alive connection returns nothing it means
// the connection was closed (either timeout or from the other side).
if err != io.EOF {
err = ErrNothingRead{err}
}
}
}
} else {
// If this is a keep-alive connection acquireByteReader will try to peek
// a couple of bytes already so the idle timeout will already be used.
br, err = acquireByteReader(&ctx)
}
ctx.Request.isTLS = isTLS
ctx.Response.Header.noDefaultContentType = s.NoDefaultContentType
ctx.Response.Header.noDefaultDate = s.NoDefaultDate
// Secure header error logs configuration
ctx.Request.Header.secureErrorLogMessage = s.SecureErrorLogMessage
ctx.Response.Header.secureErrorLogMessage = s.SecureErrorLogMessage
ctx.Request.secureErrorLogMessage = s.SecureErrorLogMessage
ctx.Response.secureErrorLogMessage = s.SecureErrorLogMessage
if err == nil {
s.setState(c, StateActive)
if s.ReadTimeout > 0 {
if err := c.SetReadDeadline(time.Now().Add(s.ReadTimeout)); err != nil {
break
}
} else if s.IdleTimeout > 0 && connRequestNum > 1 {
// If this was an idle connection and the server has an IdleTimeout but
// no ReadTimeout then we should remove the ReadTimeout.
if err := c.SetReadDeadline(zeroTime); err != nil {
break
}
}
if s.DisableHeaderNamesNormalizing {
ctx.Request.Header.DisableNormalizing()
ctx.Response.Header.DisableNormalizing()
}
// Reading Headers.
//
// If we have pipeline response in the outgoing buffer,
// we only want to try and read the next headers once.
// If we have to wait for the next request we flush the
// outgoing buffer first so it doesn't have to wait.
if bw != nil && bw.Buffered() > 0 {
err = ctx.Request.Header.readLoop(br, false)
if err == errNeedMore {
err = bw.Flush()
if err != nil {
break
}
err = ctx.Request.Header.Read(br)
}
} else {
err = ctx.Request.Header.Read(br)
}
if err == nil {
if onHdrRecv := s.HeaderReceived; onHdrRecv != nil {
reqConf := onHdrRecv(&ctx.Request.Header)
if reqConf.ReadTimeout > 0 {
deadline := time.Now().Add(reqConf.ReadTimeout)
if err := c.SetReadDeadline(deadline); err != nil {
panic(fmt.Sprintf("BUG: error in SetReadDeadline(%v): %v", deadline, err))
}
}
if reqConf.MaxRequestBodySize > 0 {
maxRequestBodySize = reqConf.MaxRequestBodySize
} else if s.MaxRequestBodySize > 0 {
maxRequestBodySize = s.MaxRequestBodySize
} else {
maxRequestBodySize = DefaultMaxRequestBodySize
}
if reqConf.WriteTimeout > 0 {
writeTimeout = reqConf.WriteTimeout
} else {
writeTimeout = s.WriteTimeout
}
}
// read body
if s.StreamRequestBody {
err = ctx.Request.readBodyStream(br, maxRequestBodySize, s.GetOnly, !s.DisablePreParseMultipartForm)
} else {
err = ctx.Request.readLimitBody(br, maxRequestBodySize, s.GetOnly, !s.DisablePreParseMultipartForm)
}
}
if (s.ReduceMemoryUsage && br.Buffered() == 0) || err != nil {
releaseReader(s, br)
br = nil
}
}
if err != nil {
if err == io.EOF {
err = nil
} else if nr, ok := err.(ErrNothingRead); ok {
if connRequestNum > 1 {
// This is not the first request and we haven't read a single byte
// of a new request yet. This means it's just a keep-alive connection
// closing down either because the remote closed it or because
// or a read timeout on our side. Either way just close the connection
// and don't return any error response.
err = nil
} else {
err = nr.error
}
}
if err != nil {
bw = s.writeErrorResponse(bw, ctx, serverName, err)
}
break
}
// 'Expect: 100-continue' request handling.
// See https://www.w3.org/Protocols/rfc2616/rfc2616-sec8.html#sec8.2.3 for details.
if ctx.Request.MayContinue() {
// Allow the ability to deny reading the incoming request body
if s.ContinueHandler != nil {
if continueReadingRequest = s.ContinueHandler(&ctx.Request.Header); !continueReadingRequest {
if br != nil {
br.Reset(ctx.c)
}
ctx.SetStatusCode(StatusExpectationFailed)
}
}
if continueReadingRequest {
if bw == nil {
bw = acquireWriter(ctx)
}
// Send 'HTTP/1.1 100 Continue' response.
_, err = bw.Write(strResponseContinue)
if err != nil {
break
}
err = bw.Flush()
if err != nil {
break
}
if s.ReduceMemoryUsage {
releaseWriter(s, bw)
bw = nil
}
// Read request body.
if br == nil {
br = acquireReader(ctx)
}
if s.StreamRequestBody {
err = ctx.Request.ContinueReadBodyStream(br, maxRequestBodySize, !s.DisablePreParseMultipartForm)
} else {
err = ctx.Request.ContinueReadBody(br, maxRequestBodySize, !s.DisablePreParseMultipartForm)
}
if (s.ReduceMemoryUsage && br.Buffered() == 0) || err != nil {
releaseReader(s, br)
br = nil
}
if err != nil {
bw = s.writeErrorResponse(bw, ctx, serverName, err)
break
}
}
}
// store req.ConnectionClose so even if it was changed inside of handler
connectionClose = s.DisableKeepalive || ctx.Request.Header.ConnectionClose()
if serverName != "" {
ctx.Response.Header.SetServer(serverName)
}
ctx.connID = connID
ctx.connRequestNum = connRequestNum
ctx.time = time.Now()
// If a client denies a request the handler should not be called
if continueReadingRequest {
s.Handler(ctx)
}
timeoutResponse = ctx.timeoutResponse
if timeoutResponse != nil {
// Acquire a new ctx because the old one will still be in use by the timeout out handler.
ctx = s.acquireCtx(c)
timeoutResponse.CopyTo(&ctx.Response)
}
if ctx.IsHead() {
ctx.Response.SkipBody = true
}
hijackHandler = ctx.hijackHandler
ctx.hijackHandler = nil
hijackNoResponse = ctx.hijackNoResponse && hijackHandler != nil
ctx.hijackNoResponse = false
if writeTimeout > 0 {
if err := c.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil {
panic(fmt.Sprintf("BUG: error in SetWriteDeadline(%v): %v", writeTimeout, err))
}
previousWriteTimeout = writeTimeout
} else if previousWriteTimeout > 0 {
// We don't want a write timeout but we previously set one, remove it.
if err := c.SetWriteDeadline(zeroTime); err != nil {
panic(fmt.Sprintf("BUG: error in SetWriteDeadline(zeroTime): %v", err))
}
previousWriteTimeout = 0
}
connectionClose = connectionClose ||
(s.MaxRequestsPerConn > 0 && connRequestNum >= uint64(s.MaxRequestsPerConn)) ||
ctx.Response.Header.ConnectionClose() ||
(s.CloseOnShutdown && atomic.LoadInt32(&s.stop) == 1)
if connectionClose {
ctx.Response.Header.SetConnectionClose()
} else if !ctx.Request.Header.IsHTTP11() {
// Set 'Connection: keep-alive' response header for HTTP/1.0 request.
// There is no need in setting this header for http/1.1, since in http/1.1
// connections are keep-alive by default.
ctx.Response.Header.setNonSpecial(strConnection, strKeepAlive)
}
if serverName != "" && len(ctx.Response.Header.Server()) == 0 {
ctx.Response.Header.SetServer(serverName)
}
if !hijackNoResponse {
if bw == nil {
bw = acquireWriter(ctx)
}
if err = writeResponse(ctx, bw); err != nil {
break
}
// Only flush the writer if we don't have another request in the pipeline.
// This is a big of an ugly optimization for https://www.techempower.com/benchmarks/
// This benchmark will send 16 pipelined requests. It is faster to pack as many responses
// in a TCP packet and send it back at once than waiting for a flush every request.
// In real world circumstances this behaviour could be argued as being wrong.
if br == nil || br.Buffered() == 0 || connectionClose {
err = bw.Flush()
if err != nil {
break
}
}
if connectionClose {
break
}
if s.ReduceMemoryUsage && hijackHandler == nil {
releaseWriter(s, bw)
bw = nil
}
}
if hijackHandler != nil {
var hjr io.Reader = c
if br != nil {
hjr = br
br = nil
}
if bw != nil {
err = bw.Flush()
if err != nil {
break
}
releaseWriter(s, bw)
bw = nil
}
err = c.SetDeadline(zeroTime)
if err != nil {
break
}
go hijackConnHandler(ctx, hjr, c, s, hijackHandler)
err = errHijacked
break
}
if ctx.Request.bodyStream != nil {
if rs, ok := ctx.Request.bodyStream.(*requestStream); ok {
releaseRequestStream(rs)
}
ctx.Request.bodyStream = nil
}
s.setState(c, StateIdle)
ctx.userValues.Reset()
ctx.Request.Reset()
ctx.Response.Reset()
if atomic.LoadInt32(&s.stop) == 1 {
err = nil
break
}
}
if br != nil {
releaseReader(s, br)
}
if bw != nil {
releaseWriter(s, bw)
}
if hijackHandler == nil {
s.releaseCtx(ctx)
}
return
}
该函数主要对用户请求进行处理，并通过将 ctx，reader，writer 对象池化减少内存的分配与回收，从而提高程序的性能。
