一、网络编程入门
在tcp/ip模型的应用层和传输层中，基于传输层抽象了一系列接口，而此接口即用于我们的实际开发中为socket(套接字)。
socket也叫套接字, 是为了方便程序员进行网络开发而被设计出来的编程接口. socket在OSI七层模型中是属于传输层(TCP,UDP协议)之上 的一个抽象接口.Go语言的标准库net包里对socket封装了一些列实现网络通信的api
1.1 net 包
net 包是网络相关的核心包。net 里面包含了 http、rpc 等关键包。 在 net 里面，最重要的两个调用：
Listen(network, addr string)：监听某个端口，等待客户端连接Dial(network, addr string)：拨号，其实也就是连上某个服务端
1.2 通信基本流程
基本分成两个大阶段。创建连接阶段:
服务端开始监听一个端口客户端拨通服务端，两者协商创建连接（TCP）
通信阶段：
客户端不断发送请求服务端读取请求服务端处理请求服务端写回响应
1.3 net.Listen
Listen 是监听一个端口，准备读取数据。它还有几个类似接口，可以直接使用：
ListenTCPListenUDPListenIPListenUDP
这几个方法都是返回 Listener 的具体类，如TCPListener。一般用 Listen 就可以，除非你要依赖于具体的网络协议特性。网络通信用 TCP 还是用 UDP 是一个影响巨大的事情，一般确认了就不会改。
func Serve(addr string) error {
listener, err := net.Listen("tcp", addr)
if err != nil {
return err
}
for {
conn, err := listener.Accept()
if err != nil {
return err
}
go func() {
handleConn(conn)
}()
}
}
复制代码
代码模板就这样。在handleConn里面读取数据-做点操作-写回响应。
1.4 处理连接
处理连接基本上就是在一个 for 循环内：
先读数据：读数据要根据上层协议来决定怎么读。例如，简单的 RPC 协议一般是分成两段读，先读头部，根据头部得知 Body 有多长，再把剩下的数据读出来。处理数据回写响应：即便处理数据出错，也要返回一个错误给客户端，不然客户端不知道你处理出错了。
func handleConn(conn net.Conn) {
for {
// 读数据
bs := make([]byte, 8)
_, err := conn.Read(bs)
if err == io.EOF || err == io.ErrUnexpectedEOF ||
err == net.ErrClosed {
// 一般关闭的错误比较懒得管
// 也可以把关闭错误输出到日志
_ = conn.Close()
return
}
if err != nil {
continue
}
res := handleMsg(bs)
_, err = conn.Write(res)
if err == io.EOF || err == io.ErrUnexpectedEOF ||
err == net.ErrClosed {
_ = conn.Close()
return
}
}
}
复制代码
1.5 错误处理
在读写的时候，都可能遇到错误，一般来说代表连接已经关掉的是这三个：
EOF、ErrUnexpectedEOF 和 ErrClosed
但是，我建议只要是出错了就直接关闭，这样对客户端和服务端代码都简单。
func handleConnV1(conn net.Conn) {
for {
// 读数据
bs := make([]byte, 8)
_, err := conn.Read(bs)
if err != nil {
// 一般关闭的错误比较懒得管
// 也可以把关闭错误输出到日志
_ = conn.Close()
return
}
res := handleMsg(bs)
_, err = conn.Write(res)
if err != nil {
_ = conn.Close()
return
}
}
}
复制代码
1.6 net.Dial
net.Dial 是指创建一个连接，连上远端的服务器。它也是有几个类似的方法：
DialIPDialTCPDialUDPDialUnixDialTimeout
只有 DialTimeout 稍微特殊一点，它多了一个超时参数。类似于 Listen，这里建议大家直接使用DialTimeout，因为设置超时可以避免一直阻塞。
func Connect(addr string) error {
conn, err := net.DialTimeout("tcp", addr, 3*time.Second)
if err != nil {
return err
}
defer func() {
_ = conn.Close()
}()
for {
// 发送请求
_, err := conn.Write([]byte("hello"))
if err != nil {
return err
}
res := make([]byte, 8)
// 接收响应
_, err = conn.Read(res)
if err != nil {
return err
}
fmt.Println(string(res))
time.Sleep(time.Second)
}
}
复制代码
这个模板和服务端处理请求的模板也很像。
1.7 goroutine 问题
前面的模板，是在创建了连接之后，就交给另外一个 goroutine 去处理，除了这个位置，还有两个位置：
在读取了请求之后，交给别的 goroutine处理，当前的 goroutine 继续读请求写响应的时候，交给别的 goroutine 去写
由上至下：
TCP 通信效率提高系统复杂度提高
因为 goroutine 非常轻量，所以即便是在模式一下，对于小型应用来说，性能也可以满足。
1.8 创建简单的 TCP 服务器
在前面的代码里面，创建的接收数据的字节数组都是固定长度的，那么问题在于，在真实的环境下，长度应该是不确定的。比如说发送字符串“Hello”和发送字符串“Hello，world”，这长度就不太一样了，怎么办？
用前8字节来描述数据的长度，再根据描述的数据长度来读取真实发送的数据(和tcp粘包问题的一般解决思路是一样的)
// 假定我们永远用 8 个字节来存放数据长度
const lenBytes = 8
type Server struct {
addr string
}
func (s *Server) StartAndServe() error {
listener, err := net.Listen("tcp", s.addr)
if err != nil {
return err
}
for {
conn, err := listener.Accept()
if err != nil {
return err
}
go func() {
// 直接在这里处理
handleConn(conn)
}()
}
}
func (s *Server) handleConn(conn net.Conn) error {
for {
// 读数据长度
bs := make([]byte, lenBytes)
_, err := conn.Read(bs)
if err != nil {
return err
}
reqBs := make([]byte, binary.BigEndian.Uint64(bs))
// 根据数据长度读取数据
_, err = conn.Read(reqBs)
if err != nil {
return err
}
res := string(reqBs) + ", from response"
// 总长度
bs = make([]byte, lenBytes, len(res)+lenBytes)
// 写入消息长度
binary.BigEndian.PutUint64(bs, uint64(len(res)))
bs = append(bs, res...)
_, err = conn.Write(bs)
if err != nil {
return err
}
}
}
复制代码
1.9 面试要点
网络的基础知识，包含 TCP 和 UDP 的基础知识。
三次握手和四次挥手如何利用 Go 写一个简单的 TCP 服务器。直接面 net 里面的 API 是很少见的，但是如果有编程题环节，那么可能会让你直接写一个简单的 TCP 服务器。记住 goroutine 和连接的关系，可以在不同的环节使用不同的 goroutine，以充分利用TCP 的全双工通信。
二、连接池
在前面的示例代码里面，客户端创建的连接都是一次性使用。然而，创建一个连接是非常昂贵的：
要发起系统调用TCP 要完成三次握手高并发的情况，可能耗尽文件描述符
连接池就是事先创建好一定数量的连接从而复用这些创建好的连接，避免频繁的创建连接消耗资源。
1.1 开源实例
(1) silenceper/pool
Github 地址：github.com/silenceper/…
InitialCap: 这种参数是在初始化的时候直接创建好的连接数量。过小，启动的时候可能大部分请求都需要创建连接；过大，则浪费。MaxIdle：最大空闲连接数，过大浪费，过小无法 应付突发流量MaxCap：最大连接数
一般连接池处理流程
阻塞的地方可以有超时控制，例如最多阻塞1s从空闲处取出来的连接，可能需要进一步 检查这个连接有没有超时（就是很久没用了）
Put 会先看有没有阻塞的 goroutine（线程），有就直接转交如果空闲队列满了，又没有人需要连接，那么需要关闭这个连接
silenceper/pool Get 方法
// channelPool 存放连接信息
type channelPool struct {
mu sync.RWMutex
conns chan *idleConn // 空闲连接池
factory func() (interface{}, error)
close func(interface{}) error
ping func(interface{}) error
idleTimeout, waitTimeOut time.Duration
maxActive int
openingConns int
connReqs []chan connReq // 被阻塞的Get请求(连接请求)
}
复制代码
// Release 释放连接池中所有连接
func (c *channelPool) Release() {
c.mu.Lock()
conns := c.conns // 修改空闲连接池的引用
c.conns = nil // 将空闲连接池 conns 改为nil
c.factory = nil
c.ping = nil
closeFun := c.close // 修改关闭方法的引用
c.close = nil
c.mu.Unlock()
if conns == nil {
return
}
// 先关闭 channel
close(conns)
// 从 channel 中取出空闲连接然后关闭
for wrapConn := range conns {
//log.Printf("Type %v\n",reflect.TypeOf(wrapConn.conn))
closeFun(wrapConn.conn)
}
}
复制代码
// Get 从pool中取一个连接
func (c *channelPool) Get() (interface{}, error) {
conns := c.getConns() // 防止并发修改，主要是防止 Release方法
if conns == nil {
return nil, ErrClosed
}
for {
select {
case wrapConn := <-conns: // 拿到空闲连接
if wrapConn == nil {
return nil, ErrClosed
}
//判断连接是否超时，超时则丢弃
if timeout := c.idleTimeout; timeout > 0 {
if wrapConn.t.Add(timeout).Before(time.Now()) {
//丢弃并关闭该连接
c.Close(wrapConn.conn)
continue
}
}
//判断连接是否失效，失效则丢弃，如果用户没有设定 ping 方法，就不检查
if c.ping != nil {
if err := c.Ping(wrapConn.conn); err != nil {
c.Close(wrapConn.conn)
continue
}
}
return wrapConn.conn, nil
default: // 没有拿到空闲连接
c.mu.Lock()
log.Debugf("openConn %v %v", c.openingConns, c.maxActive)
// 连接池满了
if c.openingConns >= c.maxActive {
req := make(chan connReq, 1)
c.connReqs = append(c.connReqs, req)
c.mu.Unlock()、
// 阻塞在这里，直到有人放回连接请求
ret, ok := <-req
if !ok {
return nil, ErrMaxActiveConnReached
}
// 这里只检查了连接是否超时
if timeout := c.idleTimeout; timeout > 0 {
if ret.idleConn.t.Add(timeout).Before(time.Now()) {
// 如果超时丢弃并关闭该连接
c.Close(ret.idleConn.conn)
continue
}
}
// 其实也可以考虑继续检查连通性
return ret.idleConn.conn, nil
}
if c.factory == nil {
c.mu.Unlock()
return nil, ErrClosed
}
// 虽然没有空闲连接，但是连接池还没满，直接创建新的连接
conn, err := c.factory()
if err != nil {
c.mu.Unlock()
return nil, err
}
c.openingConns++
c.mu.Unlock()
return conn, nil
}
}
}
复制代码
silenceper/pool Put 方法
// Put 将连接放回pool中
func (c *channelPool) Put(conn interface{}) error {
if conn == nil {
return errors.New("connection is nil. rejecting")
}
c.mu.Lock()
if c.conns == nil {
c.mu.Unlock()
return c.Close(conn)
}
// 有阻塞的Get请求(连接请求), 把连接丢过去先到先得
if l := len(c.connReqs); l > 0 {
req := c.connReqs[0]
copy(c.connReqs, c.connReqs[1:])
c.connReqs = c.connReqs[:l-1]
// 这里将 连接请求 加入到请求阻塞队列中
req <- connReq{
idleConn: &idleConn{conn: conn, t: time.Now()},
}
c.mu.Unlock()
return nil
} else {
select {
// 空闲连接池没满
case c.conns <- &idleConn{conn: conn, t: time.Now()}:
c.mu.Unlock()
return nil
default:
c.mu.Unlock()
// 空闲连接池已满，直接关闭该连接
return c.Close(conn)
}
}
}
复制代码
总结：
Get 要考虑：
有空闲连接，直接返回否则，没超过最大连接数，直接创建新的否则，阻塞调用方Put 要考虑：
有 Get 请求被阻塞，把连接丢过去否则，没超过最大空闲连接数，放到空闲列表否则，直接关闭
1.2 连接池运作图解
(1) 起步
刚开始啥都没有，直接创建一个新的。
(2) 超过上限
假如说我们不断请求连接，直到超过了十个连接。请求被阻塞。
(3) 放回去，有阻塞请求
假如说这时候有人用完了连接，就放回来了。唤醒一个请求，然后将连接交过去。
(4) 放回去空闲连接队列
如果这个时候没有阻塞请求，并且此时空闲连接队列还没有满，那么就放回去空闲连接队列。
(5) 空闲连接队列满了
空闲队列都满了，只能关掉这个连接了。
(6) 从空闲连接队列 GET
空闲队列有可用连接，直接拿。
1.3 sql.DB 中连接池管理
它也基本遵循前面总结的：
利用 channel 来管理空闲连接利用一个队列来阻塞请求
sql.DB 有很多细节，这里我们只是看它怎么管连接的。
因为本身 DB 比较复杂，所以在 putConn 的时候要做很多校验，维持好整体状态：
处理 ErrBadConn 的情况确保 dc 并没有任何人在使用处理超时
这个方法步骤和 silenceper/pool 的 Put 流程几乎一致。
总结：过期时间处理
在 sql.DB 和连接池里面都看到了一个过期时间的处理。在开发中，还有类似的场景，例如说本地缓存的过期时间。可能的方案都是：
每一个连接都有一个 goroutine 盯着，过期了就直接 close 掉一个 goroutine 定期检查所有的连接，把过期的关掉不管，要用之前就检查一下过期了没
空闲连接我们都是放在 channel 里，怎么定期检查？
package net
import (
"net"
"sync"
"sync/atomic"
"time"
)
type Option func(p *SimplePool)
type SimplePool struct {
idleChan chan conn
waitChan chan *conReq
factory func() (net.Conn, error)
idleTimeout time.Duration
maxCnt int32
// 连接数
cnt int32
l sync.Mutex
}
func NewSimplePool(factory func()(net.Conn, error), opts...Option) *SimplePool {
res := &SimplePool {
idleChan: make(chan conn, 16),
waitChan: make(chan *conReq, 128),
factory: factory,
maxCnt: 128,
}
for _, opt := range opts {
opt(res)
}
return res
}
func (p *SimplePool) Get() (net.Conn, error) {
for {
select {
case c := <-p.idleChan:
// 超时，直接关闭.
// 有没有觉得奇怪，就是明明我们就是需要一个连接，但是我们还关闭了
if c.lastActive.Add(p.idleTimeout).Before(time.Now()) {
atomic.AddInt32(&p.cnt, -1)
_ = c.c.Close()
continue
}
return c.c, nil
default:
cnt := atomic.AddInt32(&p.cnt, 1)
if cnt <= p.maxCnt {
return p.factory()
}
atomic.AddInt32(&p.cnt, -1)
req := &conReq{
con: make(chan conn, 1),
}
// 可能阻塞在这两句，对应不同的情况。
// 所以实际上 waitChan 根本不需要设计很大的容量
// 另外，这里需不需要加锁？
p.waitChan <- req
c := <- req.con
return c.c, nil
}
}
}
func (p *SimplePool) Put(c net.Conn) {
// 为什么我只在这个部分加锁，其余部分都不加？
p.l.Lock()
if len(p.waitChan) > 0 {
req := <- p.waitChan
p.l.Unlock()
req.con <- conn{c: c, lastActive: time.Now()}
return
}
p.l.Unlock()
select {
case p.idleChan <- conn{c: c, lastActive: time.Now()}:
default:
defer func() {
atomic.AddInt32(&p.maxCnt, -1)
}()
_ = c.Close()
}
}
// WithMaxIdleCnt 自定义最大空闲连接数量
func WithMaxIdleCnt(maxIdleCnt int32) Option {
return func(p *SimplePool) {
p.idleChan = make(chan conn, maxIdleCnt)
}
}
// WithMaxCnt 自定义最大连接数量
func WithMaxCnt(maxCnt int32) Option {
return func(p *SimplePool) {
p.maxCnt = maxCnt
}
}
type conn struct {
c net.Conn
lastActive time.Time
}
type conReq struct {
con chan conn
}
1.4 面试要点
几个参数的含义：初始连接，最大空闲连接，最大连接数连接池的运作原理：拿连接会发生什么，放回去又会发生什么sql.DB 解决过期连接的懒惰策略，可以类比其它如本地缓存的