1. 背景与选型

和《基于Redis Cluster的分布式锁实现以互斥方式操作共享资源》一文一样，今天要说的Go队列方案也是有一定项目背景的。

5G消息方兴未艾！前一段时间从事了一段时间5G消息网关的研发，但凡涉及类似消息业务的网关，我们一般都离不开队列这种数据结构的支持。这个5G消息网关项目采用的是Go技术栈开发，那么我们应该如何为它选择一个与业务模型匹配且性能不差的实现呢？

如今一提到消息队列，大家第一个想到的一定是kafka，kafka的确是一款优秀的分布式队列中间件，但对于我们这个系统来说，它有些“重”，部署和运维都有门槛，并且项目组里也没有能很好维护它的专家，毕竟“可控”是技术选择的一个重要因素。除此之外，我们更想在Go技术栈的生态中挑选，但kafka是Java实现的。

Go圈里在性能上能与kafka“掰掰手腕”的成熟选手不多，nats以及其主持持久化的子项目nats-streaming算是其中两个。不过nats的消息送达模型是：At-least-once-delivery，即至少送一次（而没有kafka的精确送一次的送达模型)。一旦消费者性能下降，给nats server返回的应答超时，nats就会做消息的重发处理：即将消息重新加入到队列中。这与我们的业务模型不符，即便nats提供了发送超时的设定，但我们还是无法给出适当的timeout时间。Go圈里的另一个高性能分布式消息队列nsq采用的也是“至少送一次”的消息送达模型，因此也无法满足我们的业务需求。

我们的业务决定了我们需要的队列要支持“多生产者多消费者”模型，Go语言内置的channel也是一个不错的候选。经过多个Go版本的打磨和优化，channel的send和recv操作性能在一定数量goroutine的情况下已经可以满足很多业务场景的需求了。但channel还是不完全满足我们的业务需求。我们的系统要求尽可能将来自客户端的消息接收下来并缓存在队列中。即便下游发送性能变慢，也要将客户消息先收下来，而不是拒收或延迟响应。而channel本质上是一个具有“静态大小”的队列并且Go的channel操作语义会在channel buffer满的情况下阻塞对channel的继续send，这就与我们的场景要求有背离，即便我们使用buffered channel，我们也很难选择一个合适的len值，并且一旦buffer满，它与unbuffered channel行为无异。

这样一来，我们便选择自己实现一个简单的、高性能的满足业务要求的队列，并且最好能像channel那样可以被select监听到数据ready，而不是给消费者带去“心智负担” ：消费者采用轮询的方式查看队列中是否有数据。

2. 设计与实现方案

要设计和实现这样一个队列结构，我们需要解决三个问题：

• 实现队列这个数据结构；
• 实现多goroutine并发访问队列时对消费者和生产者的协调；
• 解决消费者使用select监听队列的问题。

我们逐一来看！

1) 基础队列结构实现来自一个未被Go项目采纳的技术提案

队列是最基础的数据结构，实现一个“先进先出(FIFO)”的练手queue十分容易，但实现一份能加入标准库、资源占用小且性能良好的queue并不容易。Christian Petrin在2018年10月份曾发起一份关于Go标准库加入queue实现的技术提案，提案对基于array和链表的多种queue实现进行详细的比对，并最终给出结论：impl7是最为适宜和有竞争力的标准库queue的候选者。虽然该技术提案目前尚未得到accept，但impl7足可以作为我们的内存队列的基础实现。

2) 为impl7添加并发支持

在性能敏感的领域，我们可以直接使用sync包提供的诸多同步原语来实现goroutine并发安全访问，这里也不例外，一个最简单的让impl7队列实现支持并发的方法就是使用sync.Mutex实现对队列的互斥访问。由于impl7并未作为一个独立的repo存在，我们将其代码copy到我们的实现中(queueimpl7.go)，并将其包名由queueimpl7改名为queue：

// github.com/bigwhite/experiments/blob/master/queue-with-select/safe-queue1/queueimpl7.go

// Package queueimpl7 implements an unbounded, dynamically growing FIFO queue.
// Internally, queue store the values in fixed sized slices that are linked using
// a singly linked list.
// This implementation tests the queue performance when performing lazy creation of
// the internal slice as well as starting with a 1 sized slice, allowing it to grow
// up to 16 by using the builtin append function. Subsequent slices are created with
// 128 fixed size.
package queue

// Keeping below as var so it is possible to run the slice size bench tests with no coding changes.
var (
        // firstSliceSize holds the size of the first slice.
        firstSliceSize = 1

        // maxFirstSliceSize holds the maximum size of the first slice.
        maxFirstSliceSize = 16

        // maxInternalSliceSize holds the maximum size of each internal slice.
        maxInternalSliceSize = 128
)
... ...

下面我们就来为以queueimpl7为底层实现的queue增加并发访问支持：

// github.com/bigwhite/experiments/blob/master/queue-with-select/safe-queue1/safe-queue.go

package queue

import (
    "sync"
)

type SafeQueue struct {
    q *Queueimpl7
    sync.Mutex
}

func NewSafe() *SafeQueue {
    sq := &SafeQueue{
        q: New(),
    }

    return sq
}

func (s *SafeQueue) Len() int {
    s.Lock()
    n := s.q.Len()
    s.Unlock()
    return n
}

func (s *SafeQueue) Push(v interface{}) {
    s.Lock()
    defer s.Unlock()

    s.q.Push(v)
}

func (s *SafeQueue) Pop() (interface{}, bool) {
    s.Lock()
    defer s.Unlock()
    return s.q.Pop()
}

func (s *SafeQueue) Front() (interface{}, bool) {
    s.Lock()
    defer s.Unlock()
    return s.q.Front()
}

我们建立一个新结构体SafeQueue，用于表示支持并发访问的Queue，该结构只是在queueimpl7的Queue的基础上嵌入了sync.Mutex。

3) 支持select监听

到这里支持并发的queue虽然实现了，但在使用上还存在一些问题，尤其是对消费者而言，它只能通过轮询的方式来检查队列中是否有消息。而Go并发范式中，select扮演着重要角色，如果能让SafeQueue像普通channel那样能支持select监听，那么消费者在使用时的心智负担将大大降低。于是我们得到了下面第二版的SafeQueue实现：

// github.com/bigwhite/experiments/blob/master/queue-with-select/safe-queue2/safe-queue.go

package queue

import (
    "sync"
    "time"
)

const (
    signalInterval = 200
    signalChanSize = 10
)

type SafeQueue struct {
    q *Queueimpl7
    sync.Mutex
    C chan struct{}
}

func NewSafe() *SafeQueue {
    sq := &SafeQueue{
        q: New(),
        C: make(chan struct{}, signalChanSize),
    }

    go func() {
        ticker := time.NewTicker(time.Millisecond * signalInterval)
        defer ticker.Stop()
        for {
            select {
            case <-ticker.C:
                if sq.q.Len() > 0 {
                    // send signal to indicate there are message waiting to be handled
                    select {
                    case sq.C <- struct{}{}:
                        //signaled
                    default:
                        // not block this goroutine
                    }
                }
            }
        }

    }()

    return sq
}

func (s *SafeQueue) Len() int {
    s.Lock()
    n := s.q.Len()
    s.Unlock()
    return n
}

func (s *SafeQueue) Push(v interface{}) {
    s.Lock()
    defer s.Unlock()

    s.q.Push(v)
}

func (s *SafeQueue) Pop() (interface{}, bool) {
    s.Lock()
    defer s.Unlock()
    return s.q.Pop()
}

func (s *SafeQueue) Front() (interface{}, bool) {
    s.Lock()
    defer s.Unlock()
    return s.q.Front()
}

从上面代码看到，每个SafeQueue的实例会伴随一个goroutine，该goroutine会定期(signalInterval)扫描其所绑定的队列实例中当前消息数，如果大于0，则会向SafeQueue结构中新增的channel发送一条数据，作为一个“事件”。SafeQueue的消费者则可以通过select来监听该channel，待收到“事件”后调用SafeQueue的Pop方法获取队列数据。下面是一个SafeQueue的简单使用示例：

// github.com/bigwhite/experiments/blob/master/queue-with-select/main.go
package main

import (
    "fmt"
    "sync"
    "time"

    queue "github.com/bigwhite/safe-queue/safe-queue2"
)

func main() {
    var q = queue.NewSafe()
    var wg sync.WaitGroup

    wg.Add(2)
    // 生产者
    go func() {
        for i := 0; i < 1000; i++ {
            time.Sleep(time.Second)
            q.Push(i + 1)

        }
        wg.Done()
    }()

    // 消费者
    go func() {
    LOOP:
        for {
            select {
            case <-q.C:
                for {
                    i, ok := q.Pop()
                    if !ok {
                        // no msg available
                        continue LOOP
                    }

                    fmt.Printf("%d\n", i.(int))
                }
            }

        }

    }()

    wg.Wait()
}

从支持SafeQueue的原理可以看到，当有多个消费者时，只有一个消费者能得到“事件”并开始消费。如果队列消息较少，只有一个消费者可以启动消费，这个机制也不会导致“惊群”；当队列中有源源不断的消费产生时，与SafeQueue绑定的goroutine可能会连续发送“事件”，多个消费者都会收到事件并启动消费行为。在这样的实现下，建议消费者在收到“事件”后持续消费，直到Pop的第二个返回值返回false(代表队列为空)，就像上面示例中的那样。

这个SafeQueue的性能“中规中矩”，比buffered channel略好(Go 1.16 darwin下跑的benchmark)：

$go test -bench .
goos: darwin
goarch: amd64
pkg: github.com/bigwhite/safe-queue/safe-queue2
cpu: Intel(R) Core(TM) i5-8257U CPU @ 1.40GHz
BenchmarkParallelQueuePush-8                10687545           110.9 ns/op        32 B/op          1 allocs/op
BenchmarkParallelQueuePop-8                 18185744            55.58 ns/op        0 B/op          0 allocs/op
BenchmarkParallelPushBufferredChan-8        10275184           127.1 ns/op        16 B/op          1 allocs/op
BenchmarkParallelPopBufferedChan-8          10168750           128.8 ns/op        16 B/op          1 allocs/op
BenchmarkParallelPushUnBufferredChan-8       3005150           414.9 ns/op        16 B/op          1 allocs/op
BenchmarkParallelPopUnBufferedChan-8         2987301           402.9 ns/op        16 B/op          1 allocs/op
PASS
ok      github.com/bigwhite/safe-queue/safe-queue2  11.209s

注：BenchmarkParallelQueuePop-8因为是读取空队列，所以没有分配内存，实际情况是会有内存分配的。另外并发goroutine的模拟差异可能导致有结果差异。

3. 扩展与问题

上面实现的SafeQueue是一个纯内存队列，一旦程序停止/重启，未处理的消息都将消失。一个传统的解决方法是采用wal(write ahead log)在推队列之前将消息持久化后写入文件，在消息出队列后将消息状态也写入wal文件中。这样重启程序时，从wal中恢复消息到各个队列即可。我们也可以将wal封装到SafeQueue的实现中，在SafeQueue的Push和Pop时自动操作wal，并对SafeQueue的使用者透明，不过这里有一个前提，那就是队列消息的可序列化（比如使用protobuf)。另外SafeQueue还需提供一个对外的wal消息恢复接口。大家可以考虑一下如何实现这些。

另外在上述的SafeQueue实现中，我们在给SafeQueue增加select监听时引入两个const：

const (
    signalInterval = 200
    signalChanSize = 10
)

对于SafeQueue的使用者而言，这两个默认值可能不满足需求，那么我们可以将SafeQueue的New方法做一些改造，采用“功能选项(functional option)”的模式为用户提供设置这两个值的可选接口，这个“作业”也留给大家了^_^。

本文所有示例代码可以在这里下载 – https://github.com/bigwhite/experiments/tree/master/queue-with-select。

“Gopher部落”知识星球正式转正（从试运营星球变成了正式星球）！“gopher部落”旨在打造一个精品Go学习和进阶社群！高品质首发Go技术文章，“三天”首发阅读权，每年两期Go语言发展现状分析，>每天提前1小时阅读到新鲜的Gopher日报，网课、技术专栏、图书内容前瞻，六小时内必答保证等满足你关于Go语言生态的所有需>求！部落目前虽小，但持续力很强。在2021年上半年，部落将策划两个专题系列分享，并且是部落独享哦：

• Go技术书籍的书摘和读书体会系列
• Go与eBPF系列

欢迎大家加入！

Go技术专栏“改善Go语⾔编程质量的50个有效实践”正在慕课网火热热销中！本专栏主要满足>广大gopher关于Go语言进阶的需求，围绕如何写出地道且高质量Go代码给出50条有效实践建议，上线后收到一致好评！欢迎大家订
阅！

我的网课“Kubernetes实战：高可用集群搭建、配置、运维与应用”在慕课网热卖>中，欢迎小伙伴们订阅学习！

我爱发短信：企业级短信平台定制开发专家 https://tonybai.com/。smspush : 可部署在企业内部的定制化短信平台，三网覆盖，不惧大并发接入，可定制扩展； 短信内容你来定，不再受约束, 接口丰富，支持长短信，签名可选。2020年4月8日，中国三大电信运营商联合发布《5G消息白皮书》，51短信平台也会全新升级到“51商用消息平台”，全面支持5G RCS消息。

著名云主机服务厂商DigitalOcean发布最新的主机计划，入门级Droplet配置升级为：1 core CPU、1G内存、25G高速SSD，价格5$/月。有使用DigitalOcean需求的朋友，可以打开这个链接地址：https://m.do.co/c/bff6eed92687 开启你的DO主机之路。

Gopher Daily(Gopher每日新闻)归档仓库 – https://github.com/bigwhite/gopherdaily

我的联系方式：

• 微博：https://weibo.com/bigwhite20xx
• 微信公众号：iamtonybai
• 博客：tonybai.com
• github: https://github.com/bigwhite
• “Gopher部落”知识星球：https://public.zsxq.com/groups/51284458844544

微信赞赏：

商务合作方式：撰稿、出书、培训、在线课程、合伙创业、咨询、广告合作。

Golang的主要 设计目标之一就是面向大规模后端服务程序，网络通信这块是服务端 程序必不可少也是至关重要的一部分。在日常应用中，我们也可以看到Go中的net以及其subdirectories下的包均是“高频+刚需”，而TCP socket则是网络编程的主流，即便您没有直接使用到net中有关TCP Socket方面的接口，但net/http总是用到了吧，http底层依旧是用tcp socket实现的。

网络编程方面，我们最常用的就是tcp socket编程了，在posix标准出来后，socket在各大主流OS平台上都得到了很好的支持。关于tcp programming，最好的资料莫过于W. Richard Stevens 的网络编程圣经《UNIX网络 编程 卷1：套接字联网API》 了，书中关于tcp socket接口的各种使用、行为模式、异常处理讲解的十分细致。Go是自带runtime的跨平台编程语言，Go中暴露给语言使用者的tcp socket api是建立OS原生tcp socket接口之上的。由于Go runtime调度的需要，golang tcp socket接口在行为特点与异常处理方面与OS原生接口有着一些差别。这篇博文的目标就是整理出关于Go tcp socket在各个场景下的使用方法、行为特点以及注意事项。

一、模型

从tcp socket诞生后，网络编程架构模型也几经演化，大致是：“每进程一个连接” –> “每线程一个连接” –> “Non-Block + I/O多路复用(linux epoll/windows iocp/freebsd darwin kqueue/solaris Event Port)”。伴随着模型的演化，服务程序愈加强大，可以支持更多的连接，获得更好的处理性能。

目前主流web server一般均采用的都是”Non-Block + I/O多路复用”（有的也结合了多线程、多进程）。不过I/O多路复用也给使用者带来了不小的复杂度，以至于后续出现了许多高性能的I/O多路复用框架， 比如libevent、libev、libuv等，以帮助开发者简化开发复杂性，降低心智负担。不过Go的设计者似乎认为I/O多路复用的这种通过回调机制割裂控制流 的方式依旧复杂，且有悖于“一般逻辑”设计，为此Go语言将该“复杂性”隐藏在Runtime中了：Go开发者无需关注socket是否是 non-block的，也无需亲自注册文件描述符的回调，只需在每个连接对应的goroutine中以“block I/O”的方式对待socket处理即可，这可以说大大降低了开发人员的心智负担。一个典型的Go server端程序大致如下：

//go-tcpsock/server.go
func handleConn(c net.Conn) {
    defer c.Close()
    for {
        // read from the connection
        // ... ...
        // write to the connection
        //... ...
    }
}

func main() {
    l, err := net.Listen("tcp", ":8888")
    if err != nil {
        fmt.Println("listen error:", err)
        return
    }

    for {
        c, err := l.Accept()
        if err != nil {
            fmt.Println("accept error:", err)
            break
        }
        // start a new goroutine to handle
        // the new connection.
        go handleConn(c)
    }
}

用户层眼中看到的goroutine中的“block socket”，实际上是通过Go runtime中的netpoller通过Non-block socket + I/O多路复用机制“模拟”出来的，真实的underlying socket实际上是non-block的，只是runtime拦截了底层socket系统调用的错误码，并通过netpoller和goroutine 调度让goroutine“阻塞”在用户层得到的Socket fd上。比如：当用户层针对某个socket fd发起read操作时，如果该socket fd中尚无数据，那么runtime会将该socket fd加入到netpoller中监听，同时对应的goroutine被挂起，直到runtime收到socket fd 数据ready的通知，runtime才会重新唤醒等待在该socket fd上准备read的那个Goroutine。而这个过程从Goroutine的视角来看，就像是read操作一直block在那个socket fd上似的。具体实现细节在后续场景中会有补充描述。

二、TCP连接的建立

众所周知，TCP Socket的连接的建立需要经历客户端和服务端的三次握手的过程。连接建立过程中，服务端是一个标准的Listen + Accept的结构(可参考上面的代码)，而在客户端Go语言使用net.Dial或DialTimeout进行连接建立：

阻塞Dial：

conn, err := net.Dial("tcp", "google.com:80")
if err != nil {
    //handle error
}
// read or write on conn

或是带上超时机制的Dial：

conn, err := net.DialTimeout("tcp", ":8080", 2 * time.Second)
if err != nil {
    //handle error
}
// read or write on conn

对于客户端而言，连接的建立会遇到如下几种情形：

1、网络不可达或对方服务未启动

如果传给Dial的Addr是可以立即判断出网络不可达，或者Addr中端口对应的服务没有启动，端口未被监听，Dial会几乎立即返回错误，比如：

//go-tcpsock/conn_establish/client1.go
... ...
func main() {
    log.Println("begin dial...")
    conn, err := net.Dial("tcp", ":8888")
    if err != nil {
        log.Println("dial error:", err)
        return
    }
    defer conn.Close()
    log.Println("dial ok")
}

如果本机8888端口未有服务程序监听，那么执行上面程序，Dial会很快返回错误：

$go run client1.go
2015/11/16 14:37:41 begin dial...
2015/11/16 14:37:41 dial error: dial tcp :8888: getsockopt: connection refused

2、对方服务的listen backlog满

还有一种场景就是对方服务器很忙，瞬间有大量client端连接尝试向server建立，server端的listen backlog队列满，server accept不及时((即便不accept，那么在backlog数量范畴里面，connect都会是成功的，因为new conn已经加入到server side的listen queue中了，accept只是从queue中取出一个conn而已)，这将导致client端Dial阻塞。我们还是通过例子感受Dial的行为特点：

服务端代码：

//go-tcpsock/conn_establish/server2.go
... ...
func main() {
    l, err := net.Listen("tcp", ":8888")
    if err != nil {
        log.Println("error listen:", err)
        return
    }
    defer l.Close()
    log.Println("listen ok")

    var i int
    for {
        time.Sleep(time.Second * 10)
        if _, err := l.Accept(); err != nil {
            log.Println("accept error:", err)
            break
        }
        i++
        log.Printf("%d: accept a new connection\n", i)
    }
}

客户端代码：

//go-tcpsock/conn_establish/client2.go
... ...
func establishConn(i int) net.Conn {
    conn, err := net.Dial("tcp", ":8888")
    if err != nil {
        log.Printf("%d: dial error: %s", i, err)
        return nil
    }
    log.Println(i, ":connect to server ok")
    return conn
}

func main() {
    var sl []net.Conn
    for i := 1; i < 1000; i++ {
        conn := establishConn(i)
        if conn != nil {
            sl = append(sl, conn)
        }
    }

    time.Sleep(time.Second * 10000)
}

从程序可以看出，服务端在listen成功后，每隔10s钟accept一次。客户端则是串行的尝试建立连接。这两个程序在Darwin下的执行 结果：

$go run server2.go
2015/11/16 21:55:41 listen ok
2015/11/16 21:55:51 1: accept a new connection
2015/11/16 21:56:01 2: accept a new connection
... ...

$go run client2.go
2015/11/16 21:55:44 1 :connect to server ok
2015/11/16 21:55:44 2 :connect to server ok
2015/11/16 21:55:44 3 :connect to server ok
... ...

2015/11/16 21:55:44 126 :connect to server ok
2015/11/16 21:55:44 127 :connect to server ok
2015/11/16 21:55:44 128 :connect to server ok

2015/11/16 21:55:52 129 :connect to server ok
2015/11/16 21:56:03 130 :connect to server ok
2015/11/16 21:56:14 131 :connect to server ok
... ...

可以看出Client初始时成功地一次性建立了128个连接，然后后续每阻塞近10s才能成功建立一条连接。也就是说在server端 backlog满时(未及时accept)，客户端将阻塞在Dial上，直到server端进行一次accept。至于为什么是128，这与darwin 下的默认设置有关：

$sysctl -a|grep kern.ipc.somaxconn
kern.ipc.somaxconn: 128

如果我在ubuntu 14.04上运行上述server程序，我们的client端初始可以成功建立499条连接。

如果server一直不accept，client端会一直阻塞么？我们去掉accept后的结果是：在Darwin下，client端会阻塞大 约1分多钟才会返回timeout：

2015/11/16 22:03:31 128 :connect to server ok
2015/11/16 22:04:48 129: dial error: dial tcp :8888: getsockopt: operation timed out

而如果server运行在ubuntu 14.04上，client似乎一直阻塞，我等了10多分钟依旧没有返回。 阻塞与否看来与server端的网络实现和设置有关。

3、网络延迟较大，Dial阻塞并超时

如果网络延迟较大，TCP握手过程将更加艰难坎坷（各种丢包），时间消耗的自然也会更长。Dial这时会阻塞，如果长时间依旧无法建立连接，则Dial也会返回“ getsockopt: operation timed out”错误。

在连接建立阶段，多数情况下，Dial是可以满足需求的，即便阻塞一小会儿。但对于某些程序而言，需要有严格的连接时间限定，如果一定时间内没能成功建立连接，程序可能会需要执行一段“异常”处理逻辑，为此我们就需要DialTimeout了。下面的例子将Dial的最长阻塞时间限制在2s内，超出这个时长，Dial将返回timeout error：

//go-tcpsock/conn_establish/client3.go
... ...
func main() {
    log.Println("begin dial...")
    conn, err := net.DialTimeout("tcp", "104.236.176.96:80", 2*time.Second)
    if err != nil {
        log.Println("dial error:", err)
        return
    }
    defer conn.Close()
    log.Println("dial ok")
}

执行结果如下（需要模拟一个延迟较大的网络环境）：

$go run client3.go
2015/11/17 09:28:34 begin dial...
2015/11/17 09:28:36 dial error: dial tcp 104.236.176.96:80: i/o timeout

三、Socket读写

连接建立起来后，我们就要在conn上进行读写，以完成业务逻辑。前面说过Go runtime隐藏了I/O多路复用的复杂性。语言使用者只需采用goroutine+Block I/O的模式即可满足大部分场景需求。Dial成功后，方法返回一个net.Conn接口类型变量值，这个接口变量的动态类型为一个*TCPConn：

//$GOROOT/src/net/tcpsock_posix.go
type TCPConn struct {
    conn
}

TCPConn内嵌了一个unexported类型：conn，因此TCPConn”继承”了conn的Read和Write方法，后续通过Dial返回值调用的Write和Read方法均是net.conn的方法：

//$GOROOT/src/net/net.go
type conn struct {
    fd *netFD
}

func (c *conn) ok() bool { return c != nil && c.fd != nil }

// Implementation of the Conn interface.

// Read implements the Conn Read method.
func (c *conn) Read(b []byte) (int, error) {
    if !c.ok() {
        return 0, syscall.EINVAL
    }
    n, err := c.fd.Read(b)
    if err != nil && err != io.EOF {
        err = &OpError{Op: "read", Net: c.fd.net, Source: c.fd.laddr, Addr: c.fd.raddr, Err: err}
    }
    return n, err
}

// Write implements the Conn Write method.
func (c *conn) Write(b []byte) (int, error) {
    if !c.ok() {
        return 0, syscall.EINVAL
    }
    n, err := c.fd.Write(b)
    if err != nil {
        err = &OpError{Op: "write", Net: c.fd.net, Source: c.fd.laddr, Addr: c.fd.raddr, Err: err}
    }
    return n, err
}

下面我们先来通过几个场景来总结一下conn.Read的行为特点。

1、Socket中无数据

连接建立后，如果对方未发送数据到socket，接收方(Server)会阻塞在Read操作上，这和前面提到的“模型”原理是一致的。执行该Read操作的goroutine也会被挂起。runtime会监视该socket，直到其有数据才会重新
调度该socket对应的Goroutine完成read。由于篇幅原因，这里就不列代码了，例子对应的代码文件：go-tcpsock/read_write下的client1.go和server1.go。

2、Socket中有部分数据

如果socket中有部分数据，且长度小于一次Read操作所期望读出的数据长度，那么Read将会成功读出这部分数据并返回，而不是等待所有期望数据全部读取后再返回。

Client端：

//go-tcpsock/read_write/client2.go
... ...
func main() {
    if len(os.Args) <= 1 {
        fmt.Println("usage: go run client2.go YOUR_CONTENT")
        return
    }
    log.Println("begin dial...")
    conn, err := net.Dial("tcp", ":8888")
    if err != nil {
        log.Println("dial error:", err)
        return
    }
    defer conn.Close()
    log.Println("dial ok")

    time.Sleep(time.Second * 2)
    data := os.Args[1]
    conn.Write([]byte(data))

    time.Sleep(time.Second * 10000)
}

Server端：

//go-tcpsock/read_write/server2.go
... ...
func handleConn(c net.Conn) {
    defer c.Close()
    for {
        // read from the connection
        var buf = make([]byte, 10)
        log.Println("start to read from conn")
        n, err := c.Read(buf)
        if err != nil {
            log.Println("conn read error:", err)
            return
        }
        log.Printf("read %d bytes, content is %s\n", n, string(buf[:n]))
    }
}
... ...

我们通过client2.go发送”hi”到Server端：
运行结果:

$go run client2.go hi
2015/11/17 13:30:53 begin dial...
2015/11/17 13:30:53 dial ok

$go run server2.go
2015/11/17 13:33:45 accept a new connection
2015/11/17 13:33:45 start to read from conn
2015/11/17 13:33:47 read 2 bytes, content is hi
...

Client向socket中写入两个字节数据(“hi”)，Server端创建一个len = 10的slice，等待Read将读取的数据放入slice；Server随后读取到那两个字节：”hi”。Read成功返回，n =2 ，err = nil。

3、Socket中有足够数据

如果socket中有数据，且长度大于等于一次Read操作所期望读出的数据长度，那么Read将会成功读出这部分数据并返回。这个情景是最符合我们对Read的期待的了：Read将用Socket中的数据将我们传入的slice填满后返回：n = 10, err = nil。

我们通过client2.go向Server2发送如下内容：abcdefghij12345，执行结果如下：

$go run client2.go abcdefghij12345
2015/11/17 13:38:00 begin dial...
2015/11/17 13:38:00 dial ok

$go run server2.go
2015/11/17 13:38:00 accept a new connection
2015/11/17 13:38:00 start to read from conn
2015/11/17 13:38:02 read 10 bytes, content is abcdefghij
2015/11/17 13:38:02 start to read from conn
2015/11/17 13:38:02 read 5 bytes, content is 12345

client端发送的内容长度为15个字节，Server端Read buffer的长度为10，因此Server Read第一次返回时只会读取10个字节；Socket中还剩余5个字节数据，Server再次Read时会把剩余数据读出（如：情形2）。

4、Socket关闭

如果client端主动关闭了socket，那么Server的Read将会读到什么呢？这里分为“有数据关闭”和“无数据关闭”。

“有数据关闭”是指在client关闭时，socket中还有server端未读取的数据，我们在go-tcpsock/read_write/client3.go和server3.go中模拟这种情况：

$go run client3.go hello
2015/11/17 13:50:57 begin dial...
2015/11/17 13:50:57 dial ok

$go run server3.go
2015/11/17 13:50:57 accept a new connection
2015/11/17 13:51:07 start to read from conn
2015/11/17 13:51:07 read 5 bytes, content is hello
2015/11/17 13:51:17 start to read from conn
2015/11/17 13:51:17 conn read error: EOF

从输出结果来看，当client端close socket退出后，server3依旧没有开始Read，10s后第一次Read成功读出了5个字节的数据，当第二次Read时，由于client端 socket关闭，Read返回EOF error。

通过上面这个例子，我们也可以猜测出“无数据关闭”情形下的结果，那就是Read直接返回EOF error。

5、读取操作超时

有些场合对Read的阻塞时间有严格限制，在这种情况下，Read的行为到底是什么样的呢？在返回超时错误时，是否也同时Read了一部分数据了呢？这个实验比较难于模拟，下面的测试结果也未必能反映出所有可能结果。我们编写了client4.go和server4.go来模拟这一情形。

//go-tcpsock/read_write/client4.go
... ...
func main() {
    log.Println("begin dial...")
    conn, err := net.Dial("tcp", ":8888")
    if err != nil {
        log.Println("dial error:", err)
        return
    }
    defer conn.Close()
    log.Println("dial ok")

    data := make([]byte, 65536)
    conn.Write(data)

    time.Sleep(time.Second * 10000)
}

//go-tcpsock/read_write/server4.go
... ...
func handleConn(c net.Conn) {
    defer c.Close()
    for {
        // read from the connection
        time.Sleep(10 * time.Second)
        var buf = make([]byte, 65536)
        log.Println("start to read from conn")
        c.SetReadDeadline(time.Now().Add(time.Microsecond * 10))
        n, err := c.Read(buf)
        if err != nil {
            log.Printf("conn read %d bytes,  error: %s", n, err)
            if nerr, ok := err.(net.Error); ok && nerr.Timeout() {
                continue
            }
            return
        }
        log.Printf("read %d bytes, content is %s\n", n, string(buf[:n]))
    }
}

在Server端我们通过Conn的SetReadDeadline方法设置了10微秒的读超时时间，Server的执行结果如下：

$go run server4.go

2015/11/17 14:21:17 accept a new connection
2015/11/17 14:21:27 start to read from conn
2015/11/17 14:21:27 conn read 0 bytes,  error: read tcp 127.0.0.1:8888->127.0.0.1:60970: i/o timeout
2015/11/17 14:21:37 start to read from conn
2015/11/17 14:21:37 read 65536 bytes, content is

虽然每次都是10微秒超时，但结果不同，第一次Read超时，读出数据长度为0；第二次读取所有数据成功，没有超时。反复执行了多次，没能出现“读出部分数据且返回超时错误”的情况。

和读相比，Write遇到的情形一样不少，我们也逐一看一下。

1、成功写

前面例子着重于Read，client端在Write时并未判断Write的返回值。所谓“成功写”指的就是Write调用返回的n与预期要写入的数据长度相等，且error = nil。这是我们在调用Write时遇到的最常见的情形，这里不再举例了。

2、写阻塞

TCP连接通信两端的OS都会为该连接保留数据缓冲，一端调用Write后，实际上数据是写入到OS的协议栈的数据缓冲的。TCP是全双工通信，因此每个方向都有独立的数据缓冲。当发送方将对方的接收缓冲区以及自身的发送缓冲区写满后，Write就会阻塞。我们来看一个例子：client5.go和server.go。

//go-tcpsock/read_write/client5.go
... ...
func main() {
    log.Println("begin dial...")
    conn, err := net.Dial("tcp", ":8888")
    if err != nil {
        log.Println("dial error:", err)
        return
    }
    defer conn.Close()
    log.Println("dial ok")

    data := make([]byte, 65536)
    var total int
    for {
        n, err := conn.Write(data)
        if err != nil {
            total += n
            log.Printf("write %d bytes, error:%s\n", n, err)
            break
        }
        total += n
        log.Printf("write %d bytes this time, %d bytes in total\n", n, total)
    }

    log.Printf("write %d bytes in total\n", total)
    time.Sleep(time.Second * 10000)
}

//go-tcpsock/read_write/server5.go
... ...
func handleConn(c net.Conn) {
    defer c.Close()
    time.Sleep(time.Second * 10)
    for {
        // read from the connection
        time.Sleep(5 * time.Second)
        var buf = make([]byte, 60000)
        log.Println("start to read from conn")
        n, err := c.Read(buf)
        if err != nil {
            log.Printf("conn read %d bytes,  error: %s", n, err)
            if nerr, ok := err.(net.Error); ok && nerr.Timeout() {
                continue
            }
        }

        log.Printf("read %d bytes, content is %s\n", n, string(buf[:n]))
    }
}
... ...

Server5在前10s中并不Read数据，因此当client5一直尝试写入时，写到一定量后就会发生阻塞：

$go run client5.go

2015/11/17 14:57:33 begin dial...
2015/11/17 14:57:33 dial ok
2015/11/17 14:57:33 write 65536 bytes this time, 65536 bytes in total
2015/11/17 14:57:33 write 65536 bytes this time, 131072 bytes in total
2015/11/17 14:57:33 write 65536 bytes this time, 196608 bytes in total
2015/11/17 14:57:33 write 65536 bytes this time, 262144 bytes in total
2015/11/17 14:57:33 write 65536 bytes this time, 327680 bytes in total
2015/11/17 14:57:33 write 65536 bytes this time, 393216 bytes in total
2015/11/17 14:57:33 write 65536 bytes this time, 458752 bytes in total
2015/11/17 14:57:33 write 65536 bytes this time, 524288 bytes in total
2015/11/17 14:57:33 write 65536 bytes this time, 589824 bytes in total
2015/11/17 14:57:33 write 65536 bytes this time, 655360 bytes in total

在Darwin上，这个size大约在679468bytes。后续当server5每隔5s进行Read时，OS socket缓冲区腾出了空间，client5就又可以写入了：

$go run server5.go
2015/11/17 15:07:01 accept a new connection
2015/11/17 15:07:16 start to read from conn
2015/11/17 15:07:16 read 60000 bytes, content is
2015/11/17 15:07:21 start to read from conn
2015/11/17 15:07:21 read 60000 bytes, content is
2015/11/17 15:07:26 start to read from conn
2015/11/17 15:07:26 read 60000 bytes, content is
....

client端：

2015/11/17 15:07:01 write 65536 bytes this time, 720896 bytes in total
2015/11/17 15:07:06 write 65536 bytes this time, 786432 bytes in total
2015/11/17 15:07:16 write 65536 bytes this time, 851968 bytes in total
2015/11/17 15:07:16 write 65536 bytes this time, 917504 bytes in total
2015/11/17 15:07:27 write 65536 bytes this time, 983040 bytes in total
2015/11/17 15:07:27 write 65536 bytes this time, 1048576 bytes in total
.... ...

3、写入部分数据

Write操作存在写入部分数据的情况，比如上面例子中，当client端输出日志停留在“write 65536 bytes this time, 655360 bytes in total”时，我们杀掉server5，这时我们会看到client5输出以下日志：

...
2015/11/17 15:19:14 write 65536 bytes this time, 655360 bytes in total
2015/11/17 15:19:16 write 24108 bytes, error:write tcp 127.0.0.1:62245->127.0.0.1:8888: write: broken pipe
2015/11/17 15:19:16 write 679468 bytes in total

显然Write并非在655360这个地方阻塞的，而是后续又写入24108后发生了阻塞，server端socket关闭后，我们看到Wrote返回er != nil且n = 24108，程序需要对这部分写入的24108字节做特定处理。

4、写入超时

如果非要给Write增加一个期限，那我们可以调用SetWriteDeadline方法。我们copy一份client5.go，形成client6.go，在client6.go的Write之前增加一行timeout设置代码：

conn.SetWriteDeadline(time.Now().Add(time.Microsecond * 10))

启动server6.go，启动client6.go，我们可以看到写入超时的情况下，Write的返回结果：

$go run client6.go
2015/11/17 15:26:34 begin dial...
2015/11/17 15:26:34 dial ok
2015/11/17 15:26:34 write 65536 bytes this time, 65536 bytes in total
... ...
2015/11/17 15:26:34 write 65536 bytes this time, 655360 bytes in total
2015/11/17 15:26:34 write 24108 bytes, error:write tcp 127.0.0.1:62325->127.0.0.1:8888: i/o timeout
2015/11/17 15:26:34 write 679468 bytes in total

可以看到在写入超时时，依旧存在部分数据写入的情况。

综上例子，虽然Go给我们提供了阻塞I/O的便利，但在调用Read和Write时依旧要综合需要方法返回的n和err的结果，以做出正确处理。net.conn实现了io.Reader和io.Writer接口，因此可以试用一些wrapper包进行socket读写，比如bufio包下面的Writer和Reader、io/ioutil下的函数等。

Goroutine safe

基于goroutine的网络架构模型，存在在不同goroutine间共享conn的情况，那么conn的读写是否是goroutine safe的呢？在深入这个问题之前，我们先从应用意义上来看read操作和write操作的goroutine-safe必要性。

对于read操作而言，由于TCP是面向字节流，conn.Read无法正确区分数据的业务边界，因此多个goroutine对同一个conn进行read的意义不大，goroutine读到不完整的业务包反倒是增加了业务处理的难度。对与Write操作而言，倒是有多个goroutine并发写的情况。不过conn读写是否goroutine-safe的测试不是很好做，我们先深入一下runtime代码，先从理论上给这个问题定个性：

net.conn只是*netFD的wrapper结构，最终Write和Read都会落在其中的fd上：

type conn struct {
    fd *netFD
}

netFD在不同平台上有着不同的实现，我们以net/fd_unix.go中的netFD为例：

// Network file descriptor.
type netFD struct {
    // locking/lifetime of sysfd + serialize access to Read and Write methods
    fdmu fdMutex

    // immutable until Close
    sysfd       int
    family      int
    sotype      int
    isConnected bool
    net         string
    laddr       Addr
    raddr       Addr

    // wait server
    pd pollDesc
}

我们看到netFD中包含了一个runtime实现的fdMutex类型字段，从注释上来看，该fdMutex用来串行化对该netFD对应的sysfd的Write和Read操作。从这个注释上来看，所有对conn的Read和Write操作都是有fdMutex互斥的，从netFD的Read和Write方法的实现也证实了这一点：

func (fd *netFD) Read(p []byte) (n int, err error) {
    if err := fd.readLock(); err != nil {
        return 0, err
    }
    defer fd.readUnlock()
    if err := fd.pd.PrepareRead(); err != nil {
        return 0, err
    }
    for {
        n, err = syscall.Read(fd.sysfd, p)
        if err != nil {
            n = 0
            if err == syscall.EAGAIN {
                if err = fd.pd.WaitRead(); err == nil {
                    continue
                }
            }
        }
        err = fd.eofError(n, err)
        break
    }
    if _, ok := err.(syscall.Errno); ok {
        err = os.NewSyscallError("read", err)
    }
    return
}

func (fd *netFD) Write(p []byte) (nn int, err error) {
    if err := fd.writeLock(); err != nil {
        return 0, err
    }
    defer fd.writeUnlock()
    if err := fd.pd.PrepareWrite(); err != nil {
        return 0, err
    }
    for {
        var n int
        n, err = syscall.Write(fd.sysfd, p[nn:])
        if n > 0 {
            nn += n
        }
        if nn == len(p) {
            break
        }
        if err == syscall.EAGAIN {
            if err = fd.pd.WaitWrite(); err == nil {
                continue
            }
        }
        if err != nil {
            break
        }
        if n == 0 {
            err = io.ErrUnexpectedEOF
            break
        }
    }
    if _, ok := err.(syscall.Errno); ok {
        err = os.NewSyscallError("write", err)
    }
    return nn, err
}

每次Write操作都是受lock保护，直到此次数据全部write完。因此在应用层面，要想保证多个goroutine在一个conn上write操作的Safe，需要一次write完整写入一个“业务包”；一旦将业务包的写入拆分为多次write，那就无法保证某个Goroutine的某“业务包”数据在conn发送的连续性。

同时也可以看出即便是Read操作，也是lock保护的。多个Goroutine对同一conn的并发读不会出现读出内容重叠的情况，但内容断点是依 runtime调度来随机确定的。存在一个业务包数据，1/3内容被goroutine-1读走，另外2/3被另外一个goroutine-2读 走的情况。比如一个完整包：world，当goroutine的read slice size < 5时，存在可能：一个goroutine读到 “worl”,另外一个goroutine读出”d”。

四、Socket属性

原生Socket API提供了丰富的sockopt设置接口，但Golang有自己的网络架构模型，golang提供的socket options接口也是基于上述模型的必要的属性设置。包括

• SetKeepAlive
• SetKeepAlivePeriod
• SetLinger
• SetNoDelay （默认no delay）
• SetWriteBuffer
• SetReadBuffer

不过上面的Method是TCPConn的，而不是Conn的，要使用上面的Method的，需要type assertion：

tcpConn, ok := c.(*TCPConn)
if !ok {
    //error handle
}

tcpConn.SetNoDelay(true)

对于listener socket, golang默认采用了 SO_REUSEADDR，这样当你重启 listener程序时，不会因为address in use的错误而启动失败。而listen backlog的默认值是通过获取系统的设置值得到的。不同系统不同：mac 128, linux 512等。

五、关闭连接

和前面的方法相比，关闭连接算是最简单的操作了。由于socket是全双工的，client和server端在己方已关闭的socket和对方关闭的socket上操作的结果有不同。看下面例子：

//go-tcpsock/conn_close/client1.go
... ...
func main() {
    log.Println("begin dial...")
    conn, err := net.Dial("tcp", ":8888")
    if err != nil {
        log.Println("dial error:", err)
        return
    }
    conn.Close()
    log.Println("close ok")

    var buf = make([]byte, 32)
    n, err := conn.Read(buf)
    if err != nil {
        log.Println("read error:", err)
    } else {
        log.Printf("read % bytes, content is %s\n", n, string(buf[:n]))
    }

    n, err = conn.Write(buf)
    if err != nil {
        log.Println("write error:", err)
    } else {
        log.Printf("write % bytes, content is %s\n", n, string(buf[:n]))
    }

    time.Sleep(time.Second * 1000)
}

//go-tcpsock/conn_close/server1.go
... ...
func handleConn(c net.Conn) {
    defer c.Close()

    // read from the connection
    var buf = make([]byte, 10)
    log.Println("start to read from conn")
    n, err := c.Read(buf)
    if err != nil {
        log.Println("conn read error:", err)
    } else {
        log.Printf("read %d bytes, content is %s\n", n, string(buf[:n]))
    }

    n, err = c.Write(buf)
    if err != nil {
        log.Println("conn write error:", err)
    } else {
        log.Printf("write %d bytes, content is %s\n", n, string(buf[:n]))
    }
}
... ...

上述例子的执行结果如下：

$go run server1.go
2015/11/17 17:00:51 accept a new connection
2015/11/17 17:00:51 start to read from conn
2015/11/17 17:00:51 conn read error: EOF
2015/11/17 17:00:51 write 10 bytes, content is

$go run client1.go
2015/11/17 17:00:51 begin dial...
2015/11/17 17:00:51 close ok
2015/11/17 17:00:51 read error: read tcp 127.0.0.1:64195->127.0.0.1:8888: use of closed network connection
2015/11/17 17:00:51 write error: write tcp 127.0.0.1:64195->127.0.0.1:8888: use of closed network connection

从client1的结果来看，在己方已经关闭的socket上再进行read和write操作，会得到”use of closed network connection” error；
从server1的执行结果来看，在对方关闭的socket上执行read操作会得到EOF error，但write操作会成功，因为数据会成功写入己方的内核socket缓冲区中，即便最终发不到对方socket缓冲区了，因为己方socket并未关闭。因此当发现对方socket关闭后，己方应该正确合理处理自己的socket，再继续write已经无任何意义了。

六、小结

本文比较基础，但却很重要，毕竟golang是面向大规模服务后端的，对通信环节的细节的深入理解会大有裨益。另外Go的goroutine+阻塞通信的网络通信模型降低了开发者心智负担，简化了通信的复杂性，这点尤为重要。

本文代码实验环境：go 1.5.1 on Darwin amd64以及部分在ubuntu 14.04 amd64。

本文demo代码在这里可以找到。