go语言提倡的并发模型就是：**不要通过共享内存的方式进行通信，而是应该通过通信的方式共享内存**

# 目录

[TOC]

# mind map

![](/images/blog/1/golang/assert/channel.png)

# 结构体

```go
type hchan struct {
	qcount   uint           // total data in the queue
	dataqsiz uint           // size of the circular queue
  // buf数组，这里存储的就是发送的数据
	buf      unsafe.Pointer // points to an array of dataqsiz elements
	elemsize uint16
  // chan是否关闭
	closed   uint32
	elemtype *_type // element type
  // 数组写过队列应该明白就是为了让数组能转起来
	sendx    uint   // send index
	recvx    uint   // receive index
	// 这里维护等待 发送/接收 goroutine
	recvq    waitq  // list of recv waiters
	sendq    waitq  // list of send waiters
	// lock protects all fields in hchan, as well as several
	// fields in sudogs blocked on this channel.
	//
	// Do not change another G's status while holding this lock
	// (in particular, do not ready a G), as this can deadlock
	// with stack shrinking.
	lock mutex // 互斥锁，buf数组并不是线程安全的
}
```
# 调用栈 
下面的内容都是源码分析不想深入的同学，就不要看了。chan有4个动作，相应的**代码位置**：
1. make  -> runtime/chan.makechan
2. send -> runtime/chan.chansend
3. recieve -> runtime/chan.chanrecv
4. closed -> runtime/chan.closechan

## make

```go
func makechan(t *chantype, size int) *hchan {
	elem := t.elem
	// compiler checks this but be safe.
  // 这里是编译器检测
	if elem.size >= 1<<16 {
		throw("makechan: invalid channel element type")
	}
	if hchanSize%maxAlign != 0 || elem.align > maxAlign {
		throw("makechan: bad alignment")
	}
	mem, overflow := math.MulUintptr(elem.size, uintptr(size))
	if overflow || mem > maxAlloc-hchanSize || size < 0 {
		panic(plainError("makechan: size out of range"))
	}
	// Hchan does not contain pointers interesting for GC when elements stored in buf do not contain pointers.
	// buf points into the same allocation, elemtype is persistent.
	// SudoG's are referenced from their owning thread so they can't be collected.
	// TODO(dvyukov,rlh): Rethink when collector can move allocated objects.
	var c *hchan
	switch {
	case mem == 0:// 无缓存channel
		// Queue or element size is zero.
		c = (*hchan)(mallocgc(hchanSize, nil, true))// 无缓存channel就没必要申请buf
		// Race detector uses this location for synchronization.
		c.buf = c.raceaddr()
	case elem.kind&kindNoPointers != 0:// 有缓存channel
		// Elements do not contain pointers.
		// Allocate hchan and buf in one call.
    // 这里提到chan最好不要包含指针传递，并发的话会引起race
		c = (*hchan)(mallocgc(hchanSize+mem, nil, true))// 结构体的大小加上buf的大小
		c.buf = add(unsafe.Pointer(c), hchanSize)
    // 这里画一下内存大小,chan struct中的buf指向就是后面多申请的这一块buf内存
    // |chan struct|buf...|
	default:
		// Elements contain pointers.
		c = new(hchan)
		c.buf = mallocgc(mem, elem, true)
	}
	c.elemsize = uint16(elem.size)
	c.elemtype = elem
	c.dataqsiz = uint(size)
	if debugChan {
		print("makechan: chan=", c, "; elemsize=", elem.size, "; elemalg=", elem.alg, "; dataqsiz=", size, "\n")
	}
	return c
}
```

chan通道实际也是通过共享内存和互斥锁实现

chan最好不要传递指针

## send

无缓冲channel send

```go
// entry point for c <- x from compiled code
//go:nosplit
func chansend1(c *hchan, elem unsafe.Pointer) {
	chansend(c, elem, true, getcallerpc())
}

/*
 * generic single channel send/recv
 * If block is not nil,
 * then the protocol will not
 * sleep but return if it could
 * not complete.
 *
 * sleep can wake up with g.param == nil
 * when a channel involved in the sleep has
 * been closed.  it is easiest to loop and re-run
 * the operation; we'll see that it's now closed.
 */
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
  // 内部检查，避免一些低级的错误发生
	if c == nil {
		if !block {
			return false
		}
		gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
		throw("unreachable")
	}
	if debugChan {
		print("chansend: chan=", c, "\n")
	}
	if raceenabled {
		racereadpc(c.raceaddr(), callerpc, funcPC(chansend))
	}

	// Fast path: check for failed non-blocking operation without acquiring the lock.
	//
  // 没有recv goroutine，就无法send
  // channel buf 满了也是无法再次发送  todo::做着的目的是为了啥
	if !block && c.closed == 0 && ((c.dataqsiz == 0 && c.recvq.first == nil) ||
		(c.dataqsiz > 0 && c.qcount == c.dataqsiz)) {
		return false
	}
	var t0 int64
	if blockprofilerate > 0 {
		t0 = cputicks()
	}
  // buf array共享内存，互斥锁来避免数据竞争
	lock(&c.lock)
	// send on closed channel 这句话应该见到的挺多的了
	if c.closed != 0 {
		unlock(&c.lock)
		panic(plainError("send on closed channel"))
	}
  // 有goroutine等待接收channel，不需要进入buf，直接发送给等待接收的goroutine
	if sg := c.recvq.dequeue(); sg != nil {
		send(c, sg, ep, func() { unlock(&c.lock) }, 3) // 这里是真正发送给接收线程的操作
		return true
	}
  // buf不满
	if c.qcount < c.dataqsiz {
		// chanbuf 获取当前，发送索引的内存地址
		qp := chanbuf(c, c.sendx)
		if raceenabled {
			raceacquire(qp)
			racerelease(qp)
		}
    // 这里是memmove(to, from, elem), 就是将ep(elem)数据写入到发送所有的地址
		typedmemmove(c.elemtype, qp, ep)
    // 接下来的操作就是：如何用数组来模拟一个队列
		c.sendx++
		if c.sendx == c.dataqsiz {
			c.sendx = 0
		}
		c.qcount++
    // 释放锁
		unlock(&c.lock)
		return true
	}

	if !block {
		unlock(&c.lock)
		return false
	}
	// buf满了，需要阻塞发送的goroutine
	// Block on the channel. Some receiver will complete our operation for us.
	gp := getg()
	mysg := acquireSudog()
	mysg.releasetime = 0
	if t0 != 0 {
		mysg.releasetime = -1
	}
	// No stack splits between assigning elem and enqueuing mysg
	// on gp.waiting where copystack can find it.
	mysg.elem = ep
	mysg.waitlink = nil
	mysg.g = gp
	mysg.isSelect = false
	mysg.c = c
	gp.waiting = mysg
	gp.param = nil
  // 这上面就是阻塞并序列化当前goroutine，放到等待发送的队列中，并释放锁
	c.sendq.enqueue(mysg)
  // 当前goroutine的运行状态：Grunning -> Gwaiting
	goparkunlock(&c.lock, waitReasonChanSend, traceEvGoBlockSend, 3)
	// Ensure the value being sent is kept alive until the
	// receiver copies it out. The sudog has a pointer to the
	// stack object, but sudogs aren't considered as roots of the
	// stack tracer.
  // keepalive可以确保ep在此之前不会被gc回收
	KeepAlive(ep)
	// someone woke us up.
	if mysg != gp.waiting {
		throw("G waiting list is corrupted")
	}
	gp.waiting = nil
	if gp.param == nil {
		if c.closed == 0 {
			throw("chansend: spurious wakeup")
		}
		panic(plainError("send on closed channel"))
	}
	gp.param = nil
	if mysg.releasetime > 0 {
		blockevent(mysg.releasetime-t0, 2)
	}
	mysg.c = nil
  // 释放goroutine
	releaseSudog(mysg)
	return true
}
```

## close

```go
func closechan(c *hchan) {
	if c == nil {
		panic(plainError("close of nil channel"))
	}
	// channel 加锁
	lock(&c.lock)
	if c.closed != 0 {
		unlock(&c.lock)
		panic(plainError("close of closed channel"))
	}
	if raceenabled {
		callerpc := getcallerpc()
		racewritepc(c.raceaddr(), callerpc, funcPC(closechan))
		racerelease(c.raceaddr())
	}
  // 标记channel已经关闭了
	c.closed = 1
	// 这里声明一个链表，收集send/recv队列中的G，方便
	var glist gList

	// release all readers
	for {
		sg := c.recvq.dequeue()
		if sg == nil {
			break
		}
		if sg.elem != nil {
			typedmemclr(c.elemtype, sg.elem)
			sg.elem = nil
		}
		if sg.releasetime != 0 {
			sg.releasetime = cputicks()
		}
		gp := sg.g
		gp.param = nil
		if raceenabled {
			raceacquireg(gp, c.raceaddr())
		}
    // glist收集所有等待接收的goroutine
		glist.push(gp)
	}

	// release all writers (they will panic)
	for {
		sg := c.sendq.dequeue()
		if sg == nil {
			break
		}
		sg.elem = nil
		if sg.releasetime != 0 {
			sg.releasetime = cputicks()
		}
		gp := sg.g
		gp.param = nil
		if raceenabled {
			raceacquireg(gp, c.raceaddr())
		}
    // // glist收集所有等待发送的goroutine
		glist.push(gp)
	}
  // channel解锁
	unlock(&c.lock)

	// Ready all Gs now that we've dropped the channel lock.
	for !glist.empty() {
		gp := glist.pop()
		gp.schedlink = 0
    // glist中的所有goroutine从Gwaiting -> Grunable 
		goready(gp, 3)
	}
}
```

## receive

```go
// chanrecv receives on channel c and writes the received data to ep.
// ep may be nil, in which case received data is ignored.
// If block == false and no elements are available, returns (false, false).
// Otherwise, if c is closed, zeros *ep and returns (true, false).
// Otherwise, fills in *ep with an element and returns (true, true).
// A non-nil ep must point to the heap or the caller's stack.
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
	// 这一块和发送的时候是一样的属于编译器的检查
	if debugChan {
		print("chanrecv: chan=", c, "\n")
	}
	if c == nil {
		if !block {
			return
		}
		gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
		throw("unreachable")
	}
	if !block && (c.dataqsiz == 0 && c.sendq.first == nil ||
		c.dataqsiz > 0 && atomic.Loaduint(&c.qcount) == 0) &&
		atomic.Load(&c.closed) == 0 {
		return
	}

	var t0 int64
	if blockprofilerate > 0 {
		t0 = cputicks()
	}
	// channel加锁
	lock(&c.lock)

	if c.closed != 0 && c.qcount == 0 {
		if raceenabled {
			raceacquire(c.raceaddr())
		}
		unlock(&c.lock)
		if ep != nil {
			typedmemclr(c.elemtype, ep)
		}
		return true, false
	}
	// 给接收者发送element
	if sg := c.sendq.dequeue(); sg != nil {
		// Found a waiting sender. If buffer is size 0, receive value
		// directly from sender. Otherwise, receive from head of queue
		// and add sender's value to the tail of the queue (both map to
		// the same buffer slot because the queue is full).
		recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
		return true, true
	}
	// buf不为空
	if c.qcount > 0 {
		// Receive directly from queue
		qp := chanbuf(c, c.recvx)
		if raceenabled {
			raceacquire(qp)
			racerelease(qp)
		}
		if ep != nil {
      // 将buf c.recvx位置的数据copy到ep内存上
			typedmemmove(c.elemtype, ep, qp)
		}
		typedmemclr(c.elemtype, qp)
    // 等价 c.recvx = (c.recvx + 1) % c.dataqsiz
		c.recvx++ 
		if c.recvx == c.dataqsiz {
			c.recvx = 0
		}
		c.qcount--
    // channel解锁
		unlock(&c.lock)
		return true, true
	}

	if !block {
		unlock(&c.lock)
		return false, false
	}

	// no sender available: block on this channel.
  // 没有接收者，阻塞当前goroutine,压入c.recvq
	gp := getg()
	mysg := acquireSudog()
	mysg.releasetime = 0
	if t0 != 0 {
		mysg.releasetime = -1
	}
	// No stack splits between assigning elem and enqueuing mysg
	// on gp.waiting where copystack can find it.
	mysg.elem = ep
	mysg.waitlink = nil
	gp.waiting = mysg
	mysg.g = gp
	mysg.isSelect = false
	mysg.c = c
	gp.param = nil
	c.recvq.enqueue(mysg)
  // 当前goroutine的状态: Grunning -> Gwaiting
	goparkunlock(&c.lock, waitReasonChanReceive, traceEvGoBlockRecv, 3)
	// someone woke us up
	if mysg != gp.waiting {
		throw("G waiting list is corrupted")
	}
	gp.waiting = nil
	if mysg.releasetime > 0 {
		blockevent(mysg.releasetime-t0, 2)
	}
	closed := gp.param == nil
	gp.param = nil
	mysg.c = nil
  // 释放当前groutine
	releaseSudog(mysg)
	return true, !closed
}
```



神秘的channel。源码面前，了无密码

整理一下如何拼接字符串及几种方式的使用场景

go拼接字符串的方式以及性能比较

[TOC]

go select 是一种仅能用于channel发送和接收消息的专用语句，此语句运行期间是阻塞的。select是go在语言层面提供的I/O多路复用的机制，专门检测多个channel是否准备完毕，可读或可写

# mind map

![](/images/blog/1/golang/assert/select.png)

# 编译器优化

1. case数量为0的话，compiler会优化为gopark阻塞线程

```go
select{} -> runtime.block() -> gopark()

// gc/select/walkselectcases
// optimization: zero-case select
if n == 0 {
  return []*Node{mkcall("block", nil, nil)}// 这里编译器优化为调用runtime.block
}

// runtime.block. 最终调用还是 gopark() 阻塞goroutine
func block() {
	gopark(nil, nil, waitReasonSelectNoCases, traceEvGoStop, 1) // forever
}
```

2. case数量为1的话，compiler会优化为if 

```go
select{								if v<-foo {
case <-foo:			-> 		....
}											}
// 源码在gc/select.walkselectcases
```

3. case数量为1,有default的话compiler会优化为 if else

```go
select{								if v<-foo {
case <-foo:			-> 		....
default:							} else {
}											}
```

# 常见用法

## 阻塞

上面已经提过，编译器会优化case == 0 的G 变为阻塞你状态

```go
select{} -> runtime.block() -> gopark()
```

## 非阻塞

如果ch没有数据的话，就会一直while死循环cpu飙升，每次都会走default

```go
for {
    select {
        case d := <-ch:
        default:
    }
}
```

## 超时控制

```go
c1 := make(chan int)

go func() {
  time.Sleep(4 * time.Second)
  c1 <- 1
}()

select {
  case d, ok :=  <-c1:
  if ok {
    fmt.Println(d)
  }
  case <-time.After(3 * time.Second):
  fmt.Println("timeout...")
}

close(c1)
```

## 随机数

生成随机3位数

```
var num int64
ch := make(chan int64, 3)
for i := 0; i < 3; i++ {
  select {
  case ch <- 0:
  case ch <- 1:
  case ch <- 2:
  case ch <- 3:
  case ch <- 4:
  case ch <- 5:
  case ch <- 6:
  case ch <- 7:
  case ch <- 8:
  case ch <- 9:
  }
  num = num*10 + <-ch
}
fmt.Println(num)
```



## 关闭channel

```go
for {
    select {
        case d := <-ch:
    }
}
```

这样写也会有问题，如果ch被close掉以后，就会一直返回0值。如果select中实际只有一个case，for rang比较安全，上面的大家都实践一下。都是平时见到比较多的坑，如果是多个返回值的话，老老实实的写双返回值。

```go
for {
    select {
        case d, ok := <-ch1:
            if !ok {
                break outer
            }
        case d, ok := <-ch2:
            if !ok {
                break outer
            }
    }
}
```

# 如何随机

go select是用heap sort来实现随机,首先看下什么是堆结构和堆排序

## 堆结构

堆是一种树形结构：

- 堆的每一个子节点都要大于或小于根节点
- 大根堆：每个元素大于其所有子元素
- 小根堆：每个元素小于其所有子元素

## 堆排序

算法思想：堆排序利用堆结构根节点的元素最大的特点，不断取出根节点并维护堆结构来排序

实现步骤：

1. 建堆，将初始化序列建成一个大根堆
2. 去掉堆顶元素，剩余的元素调整为二叉堆
3. 完成最后堆排序

时间复杂度：最坏为O(nlogn)

优势：能在插入操作和删除最大元素操作混合的动态场景中保证对数级别的运行时间，代码实现简单。

## 根据什么来排序的呢？

```go
// 打乱顺序
for i := 1; i < ncases; i++ {
  // fastrandn 随机数
  j := fastrandn(uint32(i + 1))
  pollorder[i] = pollorder[j]
  pollorder[j] = uint16(i)
}
// sort the cases by Hchan address to get the locking order.
// 根据channel的内存地址来获取上锁的顺序
// simple heap sort, to guarantee n log n time and constant stack footprint.
// 堆排序：O(nlogn)时间保证和堆栈占用的保证
for i := 0; i < ncases; i++ {
    j := i
    // Start with the pollorder to permute cases on the same channel.
    c := scases[pollorder[i]].c
    for j > 0 && scases[lockorder[(j-1)/2]].c.sortkey() < c.sortkey() {
        k := (j - 1) / 2
        lockorder[j] = lockorder[k]
        j = k
    }
    lockorder[j] = pollorder[i]
}
for i := ncases - 1; i >= 0; i-- {
    o := lockorder[i]
    c := scases[o].c
    lockorder[i] = lockorder[0]
    j := 0
    for {
        k := j*2 + 1
        if k >= i {
            break
        }
        if k+1 < i && scases[lockorder[k]].c.sortkey() < scases[lockorder[k+1]].c.sortkey() {
            k++
        }
        if c.sortkey() < scases[lockorder[k]].c.sortkey() {
            lockorder[j] = lockorder[k]
            j = k
            continue
        }
        break
    }
    lockorder[j] = o
}
```

# 调用栈

1. `func Select(cases []SelectCase) (chosen int, recv Value, recvOK bool)`  -> reflect/value.Select
2. `func reflect_rselect(cases []runtimeSelect) (int, bool)`  -> reflect/select.reflect_rselect
3. `func selectgo(cas0 *scase, order0 *uint16, ncases int) (int, bool)` -> reflect/select.selectgo

3个函数有3个结构体(SelectCase/runtimeSelect/scase)，都是针对case抽象出来的结构体，判断每个case的类型，接收/发送/默认。主要功能是在selectgo中，代码过长，只看下主要代码：

```go
// cas0 points to an array of type [ncases]scase, and order0 points to
// an array of type [2*ncases]uint16. Both reside on the goroutine's
// stack (regardless of any escaping in selectgo).
func selectgo(cas0 *scase, order0 *uint16, ncases int) (int, bool) {
	...
  // 指向case0数组首地址
	cas1 := (*[1 << 16]scase)(unsafe.Pointer(cas0))
  // 长度是case1的两倍
	order1 := (*[1 << 17]uint16)(unsafe.Pointer(order0))
  // slice里面有两个冒号什么意思呢？ a[x:y:z] 切片长度: y-x(ncases-0 = ncases) 切片容量:z-x (ncases-0 = ncases)
  scases := cas1[:ncases:ncases] //scases len: ncases cap: ncases
  // 轮询顺序
	pollorder := order1[:ncases:ncases]
  // cannel 加速顺序
	lockorder := order1[ncases:][:ncases:ncases]
  // 下面操作就是剔除channel == nil的case
	for i := range scases {
		cas := &scases[i]
		if cas.c == nil && cas.kind != caseDefault {
			*cas = scase{}
		}
	}
	var t0 int64
	if blockprofilerate > 0 {
		t0 = cputicks()
		for i := 0; i < ncases; i++ {
			scases[i].releasetime = -1
		}
	}
	// 这里的代码牵扯到select是如何随机的，下面统一讲	
	...

	// 对所有的channel加锁
	sellock(scases, lockorder)

	var (
		gp     *g
		sg     *sudog
		c      *hchan
		k      *scase
		sglist *sudog
		sgnext *sudog
		qp     unsafe.Pointer
		nextp  **sudog
	)

loop:
	// pass 1 - look for something already waiting
	var dfli int
	var dfl *scase
	var casi int
	var cas *scase
	var recvOK bool
  // 这里开始遍历case数组
	for i := 0; i < ncases; i++ {
		casi = int(pollorder[i])
		cas = &scases[casi] // case结构体
		c = cas.c // case对应的channel
		switch cas.kind {
		case caseNil: // channel为nil
			continue
		case caseRecv:// channel为接收者
			sg = c.sendq.dequeue()
			if sg != nil {// channel.sendq 发送队列存在runnable G
				goto recv
			}
			if c.qcount > 0 {// channel.qcount channel元素数量大于0
				goto bufrecv
			}
			if c.closed != 0 {// channel已经关闭了
				goto rclose
			}
		case caseSend:// channel为发送者
      ...
			if c.closed != 0 {
				goto sclose
			}
			sg = c.recvq.dequeue()
			if sg != nil {// channel的接收者队列 存在 runnable G
				goto send
			}
			if c.qcount < c.dataqsiz {// channel队列，没有满
				goto bufsend
			}
		case caseDefault:// default case
			dfli = casi
			dfl = cas
		}
	}

	if dfl != nil {
		selunlock(scases, lockorder)
		casi = dfli
		cas = dfl
		goto retc
	}

	// pass 2 - enqueue on all chans
  // 当前G 入队 && runnable -> waitting
	gp = getg()
	if gp.waiting != nil {
		throw("gp.waiting != nil")
	}
	nextp = &gp.waiting
	for _, casei := range lockorder {
		... // 这些就是序列化当前G
		switch cas.kind {
		case caseRecv:
      // 入队：channel接收者队列
			c.recvq.enqueue(sg)
		case caseSend:
      // 入队：channel发送者队列
			c.sendq.enqueue(sg)
		}
	}
	// wait for someone to wake us up
	gp.param = nil
	gopark(selparkcommit, nil, waitReasonSelect, traceEvGoBlockSelect, 1)
	sellock(scases, lockorder)
	gp.selectDone = 0
	sg = (*sudog)(gp.param)
	gp.param = nil

	// pass 3 - dequeue from unsuccessful chans
	// otherwise they stack up on quiet channels
	// record the successful case, if any.
	// We singly-linked up the SudoGs in lock order.
  // 被唤醒 - 这种情况不存在default
	casi = -1
	cas = nil
	sglist = gp.waiting
	// Clear all elem before unlinking from gp.waiting.
	for sg1 := gp.waiting; sg1 != nil; sg1 = sg1.waitlink {
		sg1.isSelect = false
		sg1.elem = nil
		sg1.c = nil
	}
	gp.waiting = nil

	for _, casei := range lockorder {
		k = &scases[casei]
		if k.kind == caseNil {
			continue
		}
		if sglist.releasetime > 0 {
			k.releasetime = sglist.releasetime
		}
		if sg == sglist {
			// sg has already been dequeued by the G that woke us up.
			casi = int(casei)
			cas = k
		} else {
			c = k.c
			if k.kind == caseSend {
				c.sendq.dequeueSudoG(sglist)
			} else {
				c.recvq.dequeueSudoG(sglist)
			}
		}
		sgnext = sglist.waitlink
		sglist.waitlink = nil
		releaseSudog(sglist)
		sglist = sgnext
	}
	// 当前G被唤醒，如果没有对应的case，重新进入loop循环
	if cas == nil {
		// We can wake up with gp.param == nil (so cas == nil)
		// when a channel involved in the select has been closed.
		// It is easiest to loop and re-run the operation;
		// we'll see that it's now closed.
		// Maybe some day we can signal the close explicitly,
		// but we'd have to distinguish close-on-reader from close-on-writer.
		// It's easiest not to duplicate the code and just recheck above.
		// We know that something closed, and things never un-close,
		// so we won't block again.
		goto loop
	}

	c = cas.c

	if debugSelect {
		print("wait-return: cas0=", cas0, " c=", c, " cas=", cas, " kind=", cas.kind, "\n")
	}

	if cas.kind == caseRecv {
		recvOK = true
	}

	if raceenabled {
		if cas.kind == caseRecv && cas.elem != nil {
			raceWriteObjectPC(c.elemtype, cas.elem, cas.pc, chanrecvpc)
		} else if cas.kind == caseSend {
			raceReadObjectPC(c.elemtype, cas.elem, cas.pc, chansendpc)
		}
	}
	if msanenabled {
		if cas.kind == caseRecv && cas.elem != nil {
			msanwrite(cas.elem, c.elemtype.size)
		} else if cas.kind == caseSend {
			msanread(cas.elem, c.elemtype.size)
		}
	}

	selunlock(scases, lockorder)
	goto retc
...
```

go select 是一种仅能用于channel发送和接收消息的专用语句，此语句运行期间是阻塞的。select是go在语言层面提供的I/O多路复用的机制，专门检测多个channel是否准备完毕，可读或可写

go select 源码分析,用法


[TOC]

# 什么是rpc，rpc和http的区别，为什么要用rpc

访问量加大以后，为了解决用户访问卡顿的问题，我们就上集群架设nginx部署多个服务，由nginx负载均衡把请求转发到其他服务上。过一段时间发现发现数据库扛不住了，应用服务完好，这时候我们就上数据库读写分离，再架设几台数据库服务器，做主从，分库分表。服务访问量越来越高，项目功能也越来越多，整个项目也愈发庞大。修改一个类就需要全盘上传，切换nginx重启，分布流程越来越长，也越来越复杂。我们开始把模块拆分，用户系统、订单系统、评论系统...，这样就达到用户模块代码修改的时候只需要修改用户信息服务。但是还是需要切换顶层的nginx，把需要重启的服务的流量切到可用服务上。这个时候rpc登场了，所有服务启动的时候注册到一个注册机里面，然后顶层处理在接收到nginx请求后，去注册机里找一个可用的服务并调用接口。

用户和订单肯定是分不开的，分离成为两个系统不同的主机不同的进程以后如何进行交互，可以通过http或者rpc，http的优点是：简单、通用、开发方便。相较http rpc的优点是：长连接（不必像http 3次握手）、rpc一般都有注册中心，有丰富的监控管理；

一般来说，RPC服务主要是针对大型企业的，而HTTP服务主要是针对小企业的，因为RPC效率更高，而HTTP服务开发迭代会更快。

# 前期准备

## 安装 Protocol Buffers v3 工具
从 [https://github.com/google/protobuf/releases](https://github.com/google/protobuf/releases) 下载工具
1. 解压
2. bin/protoc 添加环境变量

## 安装 protoc 插件 protoc-gen-go
```shell
go get -u github.com/golang/protobuf/protoc-gen-go
```

# c & s 基本服务与请求

[【源代码】](https://github.com/douyacun/foo_go/tree/master/cmd/rpc/cs)

protoc 是 [Protocol Buffers](https://developers.google.com/protocol-buffers/docs/overview) 的一个工具，负责生成协议的序列化、反序列化相应语言的代码，看个例子:

首先定义proto协议文件：
```protobuf
// 这里声明协议版本，默认是proto2, 不向下兼容
syntax = "proto3";

// 这里指定 go 包名
package greeter;

// 定义要序列化的信息的结构
message HelloRequest {
    // string 字段类型， 也可以自定义字段类型
    string name = 1; // 注意这里1是唯一的， 1-15 占用1个字节， 16 - 2047 占用2个字节
    // 预留字段
    reserved "foo", "bar";
}
// 同一个proto文件中可以定义多个message type
message HelloReply {
    string message = 1;
}
// 定义接口
service Greeter {
    rpc SayHello (HelloRequest) returns (HelloReply);
}
```
具体的字段类型个语言对照可以查看[https://developers.google.com/protocol-buffers/docs/proto3#scalar](https://developers.google.com/protocol-buffers/docs/proto3#scalar)


编译grpc生成序列化和反序列化代码
```shell
protoc --go_out=plugins=grpc:. hello.proto
```

gRPC插件会为服务端和客户端生成不同的接口：
```go
//client api
type GreeterClient interface {
	SayHello(ctx context.Context, in *HelloRequest, opts ...grpc.CallOption) (*HelloReply, error)
}

// server API
type GreeterServer interface {
	SayHello(context.Context, *HelloRequest) (*HelloReply, error)
}
```

## server

1. 定义结构体，实现服务端接口
2. 注册rpc 服务
3. 启用tcp服务，接受一个链接启用一个goroutine，完成http/2握手，接收gRPC请求，调用注册函数

```go
const Port = ":4000"
// 定义结构体
type server struct{}
// 实现 service greeter 定义的接口
func (p *server) SayHello(ctx context.Context, req *greeter.HelloRequest) (*greeter.HelloReply, error) {
	reply := &greeter.HelloReply{
		Message:              "hello " + req.GetName(),
	}
	return reply, nil
}
func main() {
  // 注册rpc server
	greeter.RegisterGreeterServer(s, new(server))
	lis, err := net.Listen("tcp", Port)
	if err != nil {
		log.Fatal(err)
	}
	if err := grpc.NewServer().Serve(lis); err != nil {
		log.Fatalf("failed to serve: %v", err)
	}
}
```

## client

1. 连接服务器，`grpc.WithInsecure` 不需要权限验证
2. 初始化客户端，gRPC自动生成的客户端
3. 调用api

```go
func main() {
	conn, err := grpc.Dial("localhost:4000", grpc.WithInsecure())
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	client := greeter.NewGreeterClient(conn)
	if err != nil {
		log.Fatal(err)
	}
	reply, err := client.SayHello(context.Background(), &greeter.HelloRequest{Name: "world"})
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(reply.GetMessage())
}
```

# stream 简单聊天室

[【源代码】](https://github.com/douyacun/foo_go/tree/master/cmd/rpc/stream)

定义protobuf

```protobuf
// 定义接口
service Greeter {
		...
    // 声明关键字stream，这里可以只有参数使用或者返回值使用stream
    rpc RouteChat (stream HelloRequest) returns (stream HelloReply);
}
```

## server

```go
type Greeter_RouteChatServer interface {
	Send(*HelloReply) error
	Recv() (*HelloRequest, error)
	grpc.ServerStream
}
```

greeter.Greeter_RouteChatServer 接口,有Send/Recv两个方法相当于write/read，当前stream要当作单个connection，接下来的操作就是对stream的读写操作了

```go
func (p *server) RouteChat(stream greeter.Greeter_RouteChatServer) error {
	for {
    // Recv 返回前面定义 HelloRequest 结构体
		request, err := stream.Recv()
		if err != nil {
      // EOF 表示客户端结束了连接
			if err == io.EOF {
				return nil
			}
			return err
		}
		err = stream.Send(&greeter.HelloReply{Message: "hello " + request.GetName()})
		if err != nil {
			return err
		}
	}
}

```

## client

同服务端建立长连接，间隔1秒发送一次请求

```go
func main() {
	conn, err := grpc.Dial("localhost:4000", grpc.WithInsecure())
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	client := greeter.NewGreeterClient(conn)
	stream, err := client.RouteChat(context.Background())
	if err != nil {
		log.Fatal(stream)
	}
	go func() {
		for{
			if err := stream.Send(&greeter.HelloRequest{Name: "world"}); err != nil {
				log.Fatal(err)
			}
			time.Sleep(time.Second)
		}
	}()
	for {
		reply, err := stream.Recv()
		if err != nil {
			if err == io.EOF {
				break
			}
			log.Fatal(err)
		}
		fmt.Println(reply.GetMessage())
	}
}
```

# 发布订阅模式

[【源码阅读】](https://github.com/douyacun/foo_go/tree/master/cmd/rpc/pubsub)

找个优秀的包，godoc搜索 pubsub https://godoc.org/?q=pubsub 看到docker的pubsub引用最多，而且和第二名差距甚大，决定使用docker提供的pubsub

本地使用发布订阅：

```go
p := pubsub.NewPublisher(100*time.Millisecond, 10)
c := p.Subscribe()

p.Publish("hi")

msg := <-c
fmt.Println(msg)
```

使用grpc 实现 发布订阅模式， 创建protobuf文件

```protobuf
syntax = "proto3";

package douyacun;

message Topic {
    string name = 1;
}
message PubsubMessage {
    bytes data = 1;
}
message PublishRequest {
    Topic topic = 1;
    PubsubMessage messages = 2;
}
message PublishResponse {
    uint64 message_id = 1;
}
service Publisher {
    rpc Publish (PublishRequest) returns (PublishResponse);
    rpc Subscribe (Topic) returns (stream PubsubMessage);
}
```

## server

```go
type server struct {
	Pub *pubsub.Publisher
}

func (s *server) Publish(c context.Context, pub *douyacun.PublishRequest) (*douyacun.PublishResponse, error) {
	s.Pub.Publish(pub)
	return &douyacun.PublishResponse{MessageId: 1}, nil
}

func (s *server) Subscribe(req *douyacun.Topic, stream douyacun.Publisher_SubscribeServer) error {
	ch := s.Pub.SubscribeTopic(func(v interface{}) bool {
    // 订阅全部消息
		if req.GetName() == "" {
			return true
		}
		if it, ok := v.(*douyacun.PublishRequest); ok {
			if it.Topic.GetName() == req.GetName() {
				return true
			}
		}
		return false
	})
	for v := range ch {
		if pub, ok := v.(*douyacun.PublishRequest); ok {
			if err := stream.Send(pub.GetMessages()); err != nil {
				return err
			}
		}
	}
	return nil
}
```

## client

### 订阅

这样就可以多启动几个订阅进程

```go
cc, err := grpc.Dial(":12345", grpc.WithInsecure())
if err != nil {
  log.Fatalf("grpc.Dial 错误：%v", err)
}
client := douyacun.NewPublisherClient(cc)
stream, err := client.Subscribe(context.Background(), &douyacun.Topic{Name: "golang"})
if err != nil {
  log.Fatalf("client.Subscribe 错误: %v", err)
}
for {
  msg, err := stream.Recv()
  if err != nil {
    if err == io.EOF {
      return
    }
    log.Fatal(err)
  }
  fmt.Printf("新消息：%s\n", msg.String())
}
```

### 发布

```go
cc, err := grpc.Dial(":12345", grpc.WithInsecure())
if err != nil {
  log.Fatal(err)
}
client := douyacun.NewPublisherClient(cc)
for {
  resp, err := client.Publish(context.Background(), &douyacun.PublishRequest{
    Topic: &douyacun.Topic{
      Name: "golang",
    },
    Messages: &douyacun.PubsubMessage{
      Data: []byte("welcome!"),
    },
  })
  if err != nil {
    log.Fatal(err)
  }
  fmt.Println(resp.MessageId)
  time.Sleep(time.Second)
}
```



什么是rpc，rpc和http的区别，为什么要用rpc

gRPC入门

[TOC]

版本:

- go 1.13.4
- tcmalloc  gperftools 2.7

我是学习go源码时看到go内存分配使用tcmalloc,早在看redis源码时就了解到tcmalloc了，redis支持zmalloc、tcmalloc、jemalloc几种内存分配，之前只是看了redis自带的zmalloc,没深入了解一下tcmalloc。看来存在必有其道理，不定啥时候就用到

- https://github.com/gperftools/gperftools
- [TCMalloc : Thread-Caching Malloc](http://goog-perftools.sourceforge.net/doc/tcmalloc.html)

学习内存分配，就要先了解一下物理内存的相关概念和描述：

- [linux物理内存描述](http://www.ilinuxkernel.com/files/Linux_Physical_Memory_Description.pdf)
- [Linux物理内存页面分配](http://www.ilinuxkernel.com/files/Linux_Physical_Memory_Page_Allocation.pdf)
- [linux内存地址映射](http://www.ilinuxkernel.com/files/Linux_Memory_Address_Mapping.pdf)

看完这些以后总结了几个问题：

- 什么是逻辑地址、线性地址、物理地址，linux是如何进行地址映射的？

- linux是管理内存的(NUMA), 节点(node)、区域(zone)、page(页面)？

- 为什么一个16G的内存，内核需要160M维护page_table？

- 什么是缺页异常？（给顶的逻辑地址不合法，逻辑页面没有对应的物理页面，MMU将会产生中断，向核心发出信号）


# 基础知识

## 堆栈

![](/images/blog/1/golang/assert/堆栈.png)

进程如何分配内存：

1. static alloction: 声明全局变量时属于静态分配，该空间只会分配一次，并且永远不会释放。
2. automatic allocation: 声明函数参数/局部变量 的复合语句会自动栈空间，复合语句推出时会自动释放变量
3. dynamic allocation: malloc/free

# TCMalloc

1. **TCMalloc要比操作系统提供glibc 2.3 malloc（ptmalloc2）内存分配快**。内存的分配速度和合适的大小对于开发者来说非常重要，如果满足不了开发者的需求可能更倾向于自己实现内存分配规则
2. **TCMalloc减少了多线程之间的锁竞争**。对于小对象分配，几乎是0竞争，对于大对象，TCMalloc使用避免上下文切换的自旋锁，ptmalloc2使用per thread arenas(每个线程都会有自己的锁，避免线程之间的竞争)，问题是内存无法从arenas移动到另外一个arenas。例如：线程一申请一块300M内存，任务完成后。线程二在相同的地址空间启动，如果两个线程绑定了不同arena,线程一申请的300M内存就无法复用，必须重新申请，导致内存爆炸。
3. **TCMalloc另一个优势是对琐碎内存的高效利用**。
> 自旋锁：多线程同步的一种锁，线程反复检查锁变量是否可用。由于线程在这一过程中保持执行，因此是一种忙等待。一旦获取了自旋锁，线程会一直保持该锁，直至显式释放自旋锁。自旋锁避免了进程上下文的调度开销，因此对于线程只会阻塞很短时间的场合是有效的 - [摘自维基百科](https://zh.wikipedia.org/wiki/自旋锁)

## 概述

基本思想: 小内存通过线程层次缓存分配(无锁，速度快)，分配失败后请求上层补一批，前面分配了过多的内存则回收

线程:

ThreadCache: 每个线程私有一份，尺寸小于256K的小内存申请均由ThreadCache进行分配，线程之间不需要竞争，非常高效。节省了加锁释放锁的时间。

全局: 

PageHeap: 中央堆处理器，被所有线程共享（分配需要锁定），负责和操作系统申请、释放内存，大尺寸内存申请直接通过PageHeap分配，

CentralCache: 作为PageHeap和ThreadCache的中间人，负责将PageHeap的内存切分为小块，恰当时机分配给ThreadCache，回收ThreadCache的内存并部分返还给PageHeap

span: 一块连续的内存页。
object: object是由span切分成的小块，object预设了一些规则，如8byte、16byte、32bbyte..., 同一个span切出来的object都是相同的规格，object不大于256k，超大内存直接分配span使用\

ThreadCache、CentralCache管理object, PageHeap管理的是span

## PageHeap 大内存分配

大对象（>32k）对齐内存页面4k(4的倍数)， 由central page heap处理，central page heap也是一个数组

k个pages分配过程：

1. 查看k个页面的空闲列表（free_list）。
2. 如果free_list为空，看下一个空闲列表是否有，如此循环
3. 最终，查询到最后一个空闲列表
4. 如果匹配失败，会从操作系统申请
5. 如果有大于k个page的空闲列表，会分配k给当前申请，剩下的pages会重新插入到相应的空闲列表中。

```c++
// 根据起始page的位置和长度确定span的范围
struct Span {
  PageID        start;          // Starting page number
  Length        length;         // Number of pages in span
  Span*         next;           // 注意，这里span结构一个双向链表
  Span*         prev;           // 
  union {
    void* objects;              // Span会在CentralFreeList中拆分成由object组成的free list
    char span_iter_space[sizeof(SpanSet::iterator)];
  };
  unsigned int  refcount : 16;  // span的object被引用次数，refcount = 0时表示此span没有被使用
  unsigned int  sizeclass : 8;  // span被切分的object属于哪个级别的sizeClass
  unsigned int  location : 2;   // Span在的位置IN_USE？normal？returned？
  unsigned int  sample : 1;     // Sampled jobject?
  bool          has_span_iter : 1; 
  // What freelist the span is on: IN_USE if on none, or normal or returned
  enum { IN_USE, ON_NORMAL_FREELIST, ON_RETURNED_FREELIST };
};
```
- IN_USE：正在使用 ON_NORMAL_FREELIST：未被使用 ON_RETURNED_FREELIST：用户已经归还

> refcount、sizeclass、objects数据CentralFreeList管理的内存

PageHeap维护了
- page和span的关系
- 空闲span的伙伴关系

PageHeap结构体

```c++
class PERFTOOLS_DLL_DECL PageHeap {
  PageMap pagemap_;
  struct SpanList {
    Span        normal;
    Span        returned;
  };
  SpanSet large_normal_;
  SpanSet large_returned_;

  SpanList free_[kMaxPages];
};
```

- PageMap pagemap_ 维护id和Page的映射关系
- 大于kMaxPages页面统一存放在large_normal_中，不在按照页面数目区分

## CentralCache
CentralCache是由size_class个CentralFreeList组成的数组由Static(static_vars.h)类管理
```c++
class CentralFreeList {
  SpinLock lock_;
  // We keep linked lists of empty and non-empty spans.
  size_t   size_class_;     // size-class span切分尺寸，每个span只能按同一种size-class切分
  Span     empty_;          // 未分配Span
  Span     nonempty_;       // 已分配Span
}
```
CentralFreeList作为中间人，从PageHeap中取出部分span并按照预定大小将其拆分为大小固定的object供ThreadCache共享。CentralFreelist是全局的，除了构造函数，其余操作都需加锁。

## ThreadCache 小内存分配

**小对象分配**

1. 将其大小映射到相应的大小object
2. 在本地thread_cached中相应的free_list查找
3. 如果free_list不为空，从列表中删除第一个对象并返回它

**thread_cache free_list为空**

1. 从此大小类的central cache的free_list中获取一堆object
2. 将获取到的object放到thread_cache的free_list中
3. 将新获取的object其中一个返回

**central_cache free_list为空**

1. 从central page allocator中申请一批page
2. 拆分为该大小级别（size_class）的一组object
3. 新对象(object)放置在中央缓存（central cache）的空闲列表（free_list）中

每个thread独立维护了各自的离散式空闲列表

```c++
// https://github.com/gperftools/gperftools/blob/master/src/thread_cache.h
// https://github.com/gperftools/gperftools/blob/master/src/thread_cache.cc
class ThreadCache {
  class FreeList {
    void*    list_;       // 链表第一个节点
    uint32_t length_;      // 当前长度
    int32_t size_;
  }
  ThreadCache* next_;
  ThreadCache* prev_;
  // kClassSizesMax 96
  FreeList      list_[96];    // size_class 数组，可用链表
  int32         size_;        // 可用内存大小
  int32         max_size_;    // size_ > max_size_ --> Scavenge()
  pthread_t     tid_;         // 属于哪个线程
}
```
ThreadCache list_变量即为size class的实现，在实现free list时没有使用next来指针指向下一个位置，而是直接使用 void* list_, 将下一个object的地址直接存储在上个前8个字节中，可以模拟单向链表，object分配个应用程序后可以直接覆盖前8个字节，节省了一个指针的空间。

![object](/images/blog/1/golang/assert/tcmalloc-object.png)

## size-class
```c++
  // 1. Sizes <= 1024 have an alignment >= 8. ceil(size/8)
  // 2. Sizes > 1024 have an alignment >= 128. ceil(size/128).

  //   Size       Expression                      Index
  //   -------------------------------------------------------
  //   0          (0 + 7) / 8                     0
  //   1          (1 + 7) / 8                     1
  //   ...
  //   1024       (1024 + 7) / 8                  128
  //   1025       (1025 + 127 + (120<<7)) / 128   129
  //   ...
  //   32768      (32768 + 127 + (120<<7)) / 128  376
```



# go 内存分配

go在程序启动时会分配一块虚拟内存地址是连续的内存, 结构如下

![](/images/blog/1/golang/assert/go-内存布局.png)

## 布局

### spans

是page和其管理对象的反查表，在回收时，从哪来回哪去。利用该表可访问地址相邻大块内存使用状态，以便将多个相邻内存合并成更大的内存，减少碎片，更好适应分配需求。

### bitemap

和gc有关，先不做过多深入，研究gc时候会深入理解

栈(stack)、全局变量和静态变量（.data/.bss）占用1bit，0: gc不需要关心是否回收，1: 需要分析是否需要回收

堆(heap) 需要占用2bit

### arena

用户对象分配区域，就是通常说的堆(heap)，go从heap中申请的内存就是在这个区域。

### 逃逸分析

什么时候从heap分配对象，go会自动确定哪些对象应该放在栈上，哪些应该放在堆上。简单的说：当一个对象的内容可能在生成该对象的函数结束后被访问，那么这个对象就会被分配在堆上。

**返回对象指针**

```go
func foo() *string {
	str := "Hello world"
	return &str
}
func main() {
	fmt.Println(foo())
}
```

```shell
$ git:(master) ✗ go run -gcflags '-m -l' bar.go
# command-line-arguments
./bar.go:7:9: &str escapes to heap
./bar.go:6:2: moved to heap: str
./bar.go:11:17: foo() escapes to heap
./bar.go:11:13: main ... argument does not escape
0xc0000101e0
```

**传递了对象的指针到其他函数**

```go
func greeter(name *string) {
	 *name = "hello " + *name
}
func main() {
	name := "douyacun"
	greeter(&name)
	fmt.Println(name)
}
```

```shell
➜  bar git:(master) ✗ go run -gcflags '-m -l' bar.go
# command-line-arguments
./bar.go:6:20: "hello " + *name escapes to heap
./bar.go:5:14: greeter name does not escape
./bar.go:12:13: name escapes to heap
./bar.go:11:10: main &name does not escape
./bar.go:12:13: main ... argument does not escape
hello douyacun
```

**在闭包中使用了对象并且需要修改对象**

```go
func main() {
	name := "douyacun"
	greeter := func() {
		name = "hello " + name
	}
	greeter()
	fmt.Println(name)
}
```

```shell
➜  bar git:(master) ✗ go run -gcflags '-m -l' bar.go
# command-line-arguments
./bar.go:11:13: name escapes to heap
./bar.go:8:19: "hello " + name escapes to heap
./bar.go:7:13: main func literal does not escape
./bar.go:11:13: main ... argument does not escape
hello douyacun
```

**使用new**

```go
func main() {
	name := new([]string)
	fmt.Println(name)
}
```

```shell
➜  bar git:(master) ✗ go run -gcflags '-m -l' bar.go
# command-line-arguments
./bar.go:7:13: name escapes to heap
./bar.go:6:13: new([]string) escapes to heap
./bar.go:7:13: main ... argument does not escape
&[]
```

## 小对象分配过程

这里翻译了一下源代码的注释，描述的很清楚`runtime/mheap.go`

**mcache分配**

1. 将尺寸四舍五入到small size classes的一种
2. 在 **P** 的 mcache中查看相应的mspan
3. 扫描mspan的bitmap查找空闲坑位
4. 如果有空闲坑位，分配这个坑位，这个过程不需要获取锁

> P是协程中的用于运行go代码的虚拟资源,同一时间只能有一个线程访问同一个P,所以p的资源不需要锁

**mcentral分配**

1. mcache没有空闲坑位
2. 从有坑位的mcentral列表中获取一个新的mspan
3. 获得这个mspan需要锁定mcentral，mcentral是在线程的共享空间中，会有竞争

**mheap分配**

1. mcentral没有坑位
2. 从mheap申请一批page，切分为mspan
3. 这个过程需要加锁
4. 如果mheap没有坑位，从操作系统申请

小对象分3个阶段获取可用的span，然后从span中分配对象

![](/images/blog/1/golang/assert/go-内存分配-三阶段分配.png)



## 大对象分配

分配大内存直接通过mhe

ap获取，绕过mcache/mcentral, 大对象分配时会四舍五入(8kb)，在包含k个页面的空闲列表中寻找第K个条目，如果他是空的，向上查找，直到找到，如果失败，从操作系统查看早


go内存分配原理

innodb关键特性包括：
- 插入缓冲 (insert buffer)
- 两次写(double write)
- 自适应hash索引(adaptive hash index)
- 异步I/O (async i/o)
- 刷新临界页 (flush neighbor page)

# 插入缓冲
## insert buffer

每张表都有一个主键索引(没有明确指明，会默认生成一个不可见的主键索引)，主键索引(聚集索引)的插入顺序是非常快的，但是不可能每张表只有一个聚集索引，更多情况下一张表会有很多辅助索引，对于主键索引的插入式顺序存放的，但是对于辅助索引的叶子节点的插入就不是顺序的了。

innodb 对于非聚集索引的插入和更新操作，不是每一次直接插入到索引页，而是先判断索引页是否在缓冲池中,若有直接插入，否则，先放入到insert buffer中。再以一定频率合并insert buffer和索引子节点。通常能将多个插入操作合并到一个操作中，因为在一个索引页中。使用insert buffer需要同时满足两个条件：
- 索引是辅助索引
- 索引不是唯一索引

`show engine innodb status\G;`可以查看insert buffer的使用情况

- seg_size 代表insert buffer大小为 n*16k
- free list len代表空闲列表的长度
- inserts代表插入记录数
- merged recs代表了合并后的插入记录数
- merges 代表合并的次数，实际读取页的次数

> 缓冲池: 是为了通过内存速度来弥补磁盘速度较慢对于性能的影响，包括insert buffer、data pag、index page、lock info、adaptive hash index

**IBUF_POOL_SIZE_PER_MAX_SIZE**可以调整insert_buffer_pool可以占用缓冲池的大小

## change buffer
change buffer可以对DML操作(insert,delete,update)进行缓冲操作,适用对象仍然是非唯一的辅助索引， 对一条记录update的步骤：
- 将记录标记为已删除
- 真正将记录删除

**innodb_change_buffer_max_size**控制change buffer最大使用内存的数量，25表示最多使用1/4缓冲池空间

# double write
插入缓冲给innodb存储引擎带来的是性能上的提升，double write提高了innodb存储引擎的数据页的可靠性

那么double write做了什么？\
数据库宕机时，可能对数据页只写入部分，这种情况称为部分写失效。可以通过redo log来恢复，在写入失效时，先通过页的副本还原该页，在进行重做，这就是double write

double write有一段2m的内存空间，在磁盘上还有2m*的共享表空间，数据先从内存空间写入到共享空间，立即调用fsync同步，此时是顺序写入速度快，在完成写入后，再将共享空间的数据写入到各个表空间文件中。此时写入是离散的。double write写入时会合并写入操作，消耗要小于两次io的消耗

```shell
mysql> show global status like 'innodb_dblwr%\G'
```

- innodb_dblwr_pages_writen: inndob一共写了多少页
- innodb_dblwr_writes: 实际写入次数

`skip_innodb_doublewrite`可以禁止使用doublewrite功能

# 自适应hash索引
hash索引的查找复杂度O(1),B+树的查找取决于B+树的高度，生产环境一般是3～4,需要3～4次

innodb会监控对表上各索引的查询，如果二级索引被频繁访问，成为热点数据，建立hash索引可以带来，速度的提升

hash索引有一个要求，查询模式必须是一样的，必须是等值查询

**`innodb_adaptive_hash_index`**: 是否开启自适应hash索引

# 异步I/O - AIO
AIO优势:
1. 多个I/O异步执行
2. 合并多个I/O为同步操作

mysql read、脏页刷新，磁盘写入都是通过AIO来实现的

# 刷新邻接页
原理是：innodb刷新脏页时，会检测该页所在区的所有页，如果有脏页AIO就会将多个I/O操作合并为一个操作。
**innodb_flush_neighbors** 可以控制是否启用，传统机械键盘建议启用，固态硬盘建议关闭

innodb说这些特性，相信比说支持事务要更有说服力

innodb关键特性

mysql中常见的日志文件：
- 错误日志(error log)
- 二进制日志(binlog)
- 慢查询日志(slow log)
- 查询日志(log)

# 错误日志
查看错误日志的位置
```shell
mysql> show variables like 'log_error'\G;
```

# 慢查询日志

开启慢查询日志`log_slow_queries`
```shell
mysql> show variables like 'log_slowqueries'\G;
```
慢查询时间`long_query_time`
```shell
mysql> show variables like 'long_query_time'\G
```

sql没有用到索引也会记录到慢查询日志
```shell
mysql> show variables like 'long_throttle_queries_not_using_indexes'\G;
```

mysqldumpslow查看执行时间最长的10条sql
```shell
$ mysqldumpslow -s -al -n 10 mysqlslow.log
```

mysql支持将慢查询日志写入到slow_log表，**log_output**指定慢查询输出格式，默认为file，设置为table会写入到slow_log表

```shell
mysql> show create table mysql.slow_log;
```

# rds执行日志记录sls
阿里云关系数据库RDS（MySQL版本）的SQL执行日志已经与日志服务打通，提供准实时的SQL执行分析功能，并提供内置可深钻的报表、支持自定义审计规则、异常报警、报表订阅与生态对接能力等。
- 据库的登录、退出。
- DDL（Data Definition Language）：对数据库结构定义的SQL，例如CREATE、ALTER DROP、TRUNCATE、COMMENT等。
- DML（Data Manipulation Language）：SQL操作语句，包括SELECT、INSERT、UPDATE、DELETE等。
- 其他SQL执行，包括任何其他通过SQL执行的控制，例如回滚、控制等。
- 失败的SQL执行也会被记录。
- 对SQL执行的延迟、执行结果、影响的行数等信息的提取。

https://help.aliyun.com/document_detail/118266.html

# 二进制日志
- max_binlog_size: 二进制文件最大值，超过该值产生新的文件，后缀加1
- binlog_cache_size: innodb所有未提交的事物都会被记录缓存中去，事物提交时，在写二进制日志中。当一个session开启一个事物时，会自动初始化一个大小为binlog_cache_size大小的缓存，当事物的记录大于binlog_cache_size的时候，会把缓冲的日志写入一个临时文件
    - binlog_cache_use 缓冲写二进制日志次数 
    - binlog_cache_disk_use 临时文件写二进制日志次数

查看binlog文件的内容：
```shell
mysql> mysqlbinlog --start-position=203 douyacun.00007
```

# 套接字文件
本地mysql可以使用套接字文件来连接
```shell
mysql> show variables like 'socket';
```

# pid文件
进程pid，可以用pid文件来重启进程
```shell
mysql> show variables like "";
```



错误日志,慢查询日志,查询日志,二进制日志

mysql日志文件


# fsync,  fdatasync 同步内核的页缓存到磁盘
```c
#include <unistd.h>

int fsync(int fd);
int fdatasync(int fd);
```

- fsync： 文件描述符(fd)在内核中所有的[脏页](/linux/页缓存.md#dirty_page)同步到的磁盘，调用会被阻塞直到磁盘写入完成，能确保被修改过的块立即写入到磁盘
- fdatasync：类似于fysnc，文件属性(st_atime,st_mtime等)不会同步到磁盘，目的是为了减少磁盘交互

# 与fdatasync相比fsync的性能问题#性能
fysnc 除了同步文件的修改内容(脏页)，还会同步文件的描述信息(metadata、包括size、访问时间st_atime、st_mtime等)，因为文件数据和metadata通常存在磁盘不同的地方，所以fsync至少需要两次I/O操作

多余的一次IO操作，有多么昂贵呢？根据Wikipedia的数据，当前硬盘驱动的平均寻道时间（Average seek time）大约是3~15ms，7200RPM硬盘的平均旋转延迟（Average rotational latency）大约为4ms，因此一次IO操作的耗时大约为10ms左右。

文件的尺寸(st_size)如果发生变化，是需要立即同步的否则OS一旦崩溃，即使文件的数据部分已同步，由于metadata没有同步，依然读不到修改的内容，最后的饿访问时间和修改时间是不需要每次同步的

读redis源码-AOF和磁盘I/O造成的延迟

sync_fsync_fdatasync_linux

# 参数
- -c 统计每一系统调用的所执行的时间,次数和出错的次数等.
- -d 输出strace关于标准错误的调试信息.
- -f 跟踪由fork调用所产生的子进程.
- -ff 如果提供-o filename,则所有进程的跟踪结果输出到相应的filename.pid中,pid是各进程的进程号.
- -F 尝试跟踪vfork调用.在-f时,vfork不被跟踪.
- -h 输出简要的帮助信息.
- -i 输出系统调用的入口指针.
- -q 禁止输出关于脱离的消息.
- -r 打印出相对时间关于,,每一个系统调用.
- -t 在输出中的每一行前加上时间信息.
- -tt 在输出中的每一行前加上时间信息,微秒级.
- -ttt 微秒级输出,以秒了表示时间.
- -T 显示每一调用所耗的时间.
- -v 输出所有的系统调用.一些调用关于环境变量,状态,输入输出等调用由于使用频繁,默认不输出.
- -V 输出strace的版本信息.
- -x 以十六进制形式输出非标准字符串
- -xx 所有字符串以十六进制形式输出.
- -a column 设置返回值的输出位置.默认 为40.
- -e expr 指定一个表达式,用来控制如何跟踪.格式：[qualifier=][!]value1[,value2]...
qualifier只能是 trace,abbrev,verbose,raw,signal,read,write其中之一.value是用来限定的符号或数字.默认的 qualifier是 trace.感叹号是否定符号.例如:-eopen等价于 -e trace=open,表示只跟踪open调用.而-etrace!=open 表示跟踪除了open以外的其他调用.有两个特殊的符号 all 和 none. 注意有些shell使用!来执行历史记录里的命令,所以要使用\\.
- -e trace=set 只跟踪指定的系统 调用.例如:-e trace=open,close,rean,write表示只跟踪这四个系统调用.默认的为set=all.
- -e trace=file 只跟踪有关文件操作的系统调用.
- -e trace=process 只跟踪有关进程控制的系统调用.
- -e trace=network 跟踪与网络有关的所有系统调用.
- -e strace=signal 跟踪所有与系统信号有关的 系统调用
- -e trace=ipc 跟踪所有与进程通讯有关的系统调用
- -e abbrev=set 设定strace输出的系统调用的结果集.-v 等与 abbrev=none.默认为abbrev=all.
- -e raw=set 将指定的系统调用的参数以十六进制显示.
- -e signal=set 指定跟踪的系统信号.默认为all.如 signal=!SIGIO(或者signal=!io),表示不跟踪SIGIO信号.
- -e read=set 输出从指定文件中读出 的数据.例如: -e read=3,5
- -e write=set 输出写入到指定文件中的数据.
- -o filename 将strace的输出写入文件filename
- -p pid 跟踪指定的进程pid.
- -s strsize 指定输出的字符串的最大长度.默认为32.文件名一直全部输出.
- -u username 以username的UID和GID执行被跟踪的命令

# 查看php-fpm master进程的系统调用
```shell
[root@cpc-dev01 work]# strace -p 175664
strace: Process 175664 attached
epoll_wait(11, [], 1, 322)              = 0
epoll_wait(11, [], 1, 333)              = 0
epoll_wait(11, [], 1, 94)               = 0
getsockopt(10, SOL_TCP, TCP_INFO, "\n\0\0\0\0\0\0\0@B\17\0\0\0\0\0\30\2\0\0\0\0\0\0\0\0\0\0\200\0\0\0"..., [104]) = 0
epoll_wait(11, [], 1, 239)              = 0
epoll_wait(11, [], 1, 333)              = 0
epoll_wait(11, [], 1, 333)              = 0
epoll_wait(11, [], 1, 94)               = 0
getsockopt(10, SOL_TCP, TCP_INFO, "\n\0\0\0\0\0\0\0@B\17\0\0\0\0\0\30\2\0\0\0\0\0\0\0\0\0\0\200\0\0\0"..., [104]) = 0
epoll_wait(11, [], 1, 239)              = 0
epoll_wait(11, [], 1, 333)              = 0
epoll_wait(11, [], 1, 269)              = 0
epoll_wait(11, [], 1, 64)               = 0
epoll_wait(11, [], 1, 94)               = 0
getsockopt(10, SOL_TCP, TCP_INFO, "\n\0\0\0\0\0\0\0@B\17\0\0\0\0\0\30\2\0\0\0\0\0\0\0\0\0\0\200\0\0\0"..., [104]) = 0
```

# 统计一下mysql系统调用
```
[root@00h ~]# strace -p 2940 -c
strace: Process 2940 attached
^Cstrace: Process 2940 detached
% time     seconds  usecs/call     calls    errors syscall
------ ----------- ----------- --------- --------- ----------------
 75.28    0.001836        1836         1           poll
 18.00    0.000439         439         1           restart_syscall
  6.40    0.000156          39         4           futex
  0.33    0.000008           4         2           accept
------ ----------- ----------- --------- --------- ----------------
100.00    0.002439                     8           total
```

可以使用strace对应用的系统调用和信号传递的跟踪结果来对应用进行分析

strace,用法,使用,参数



# redis 持久化策略
- master关闭持久化
- slave 开启rdb，必要时开启aof和rdb

# aof和rdb配置
- `save` rdb出发记录条件
    - `save 900 1`     # 服务器在900秒内，修改了1次
    - `save 300 10`     # 服务器在300秒内，修改了10次
    - `save 60 10000`   # 服务器在60秒内，修改了10000次 
    - `save ""` 关闭rdb持久化
- `rdbchecksum yes` 压缩rdb文件
- `rdbcompression yes` 压缩rdb文件
- `dbfilename dump.rdb` rdb文件

- `appendonly yes` 启用aof增量持久化
- `appendfilename "appendonly.aof"` aof文件
- `appendfsync` `fsync`脏页同步策略
    - `always` 每次修改
    - `everysec` 每秒记录一次
    - `no` 取决于系统
- `no-appendfsync-on-rewrite yes` # 如果是开启状态的，并且有bgsave或bgrewriteaof正在进行的话不调用系统函数`fsync()`(同步脏页到磁盘中)
```c
/* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are
    * children doing I/O in the background. 
    *
    * 如果 no-appendfsync-on-rewrite 选项为开启状态，
    * 并且有 BGSAVE 或者 BGREWRITEAOF 正在进行的话，
    * 那么不执行 fsync 
    */
if (server.aof_no_fsync_on_rewrite &&
    (server.aof_child_pid != -1 || server.rdb_child_pid != -1))
        return;
```

# aof 实现原理
aof重写并不读取分析现有aof文件，是通过读取当前数据库状态来实现，然后用一条命令代替之前记录这个键值对的多条命令,aof重写函数，因为这个函数会进行大量的写入操作，这个线程会被长时间阻塞，no-appendfsync-on-rewrite 是为了避免在此期间与aof重写进程占用文件描述符

- fork一个子进程处理
- 父进程将新输入的写命令追加到server.aof_rewrite_buffer

aof.c/rewriteAppendOnlyFileBackground
```c
/* This is how rewriting of the append only file in background works:
 * 1) The user calls BGREWRITEAOF
 * 2) Redis calls this function, that forks():
 *    2a) the child rewrite the append only file in a temp file.
 *    2b) the parent accumulates differences in server.aof_rewrite_buf.
 * 3) When the child finished '2a' exists.
 * 4) The parent will trap the exit code, if it's OK, will append the
 *    data accumulated into server.aof_rewrite_buf into the temp file, and
 *    finally will rename(2) the temp file in the actual file name.
 *    The the new file is reopened as the new append only file. Profit!
 */
int rewriteAppendOnlyFileBackground(void) {
    pid_t childpid;
    long long start;

    // 已经有进程在进行 AOF 重写了
    if (server.aof_child_pid != -1) return REDIS_ERR;

    // 记录 fork 开始前的时间，计算 fork 耗时用
    start = ustime();

    if ((childpid = fork()) == 0) {
        char tmpfile[256];

        /* Child */

        // 关闭网络连接 fd
        closeListeningSockets(0);

        // 为进程设置名字，方便记认
        redisSetProcTitle("redis-aof-rewrite");

        // 创建临时文件，并进行 AOF 重写
        snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
        if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) {
            size_t private_dirty = zmalloc_get_private_dirty();

            if (private_dirty) {
                redisLog(REDIS_NOTICE,
                    "AOF rewrite: %zu MB of memory used by copy-on-write",
                    private_dirty/(1024*1024));
            }
            // 发送重写成功信号
            exitFromChild(0);
        } else {
            // 发送重写失败信号
            exitFromChild(1);
        }
    } else {
        /* Parent */
        // 记录执行 fork 所消耗的时间
        server.stat_fork_time = ustime()-start;

        if (childpid == -1) {
            redisLog(REDIS_WARNING,
                "Can't rewrite append only file in background: fork: %s",
                strerror(errno));
            return REDIS_ERR;
        }

        redisLog(REDIS_NOTICE,
            "Background append only file rewriting started by pid %d",childpid);

        // 记录 AOF 重写的信息
        server.aof_rewrite_scheduled = 0;
        server.aof_rewrite_time_start = time(NULL);
        server.aof_child_pid = childpid;

        // 关闭字典自动 rehash
        updateDictResizePolicy();

        /* We set appendseldb to -1 in order to force the next call to the
         * feedAppendOnlyFile() to issue a SELECT command, so the differences
         * accumulated by the parent into server.aof_rewrite_buf will start
         * with a SELECT statement and it will be safe to merge. 
         *
         * 将 aof_selected_db 设为 -1 ，
         * 强制让 feedAppendOnlyFile() 下次执行时引发一个 SELECT 命令，
         * 从而确保之后新添加的命令会设置到正确的数据库中
         */
        server.aof_selected_db = -1;
        replicationScriptCacheFlush();
        return REDIS_OK;
    }
    return REDIS_OK; /* unreached */
}
```

# 残酷的事实
1. redis 是单线程的数据库，导致单核cpu的最大处理能力就是redis单实例处理能力的天花板了, 举个例子:

新功能上线又有点不放心，于是做了个开关放在 Redis，所有应用可以很方便的共享。通过读取 Redis 中的开关 key 来判断是否启用某个功能，对每个请求做判断。这里的问题是什么？

这个key只能放在一个实例上，而所有的流量都要去这个实例去get，导致该分片实例亚历山大，他的极限在我们的实例上不过4w QPS

- 根据数据性质把 Redis 集群分类；我的经验是分三类：cache、buffer 和 db
   - cache：临时缓存数据，加分片扩容容易，一般无持久化需要。
   - buffer：用作缓冲区，平滑后端数据库的写操作，根据数据重要性可能有持久化需求。
   - db：替代数据库的用法，有持久化需求。

[redis实践](http://www.ittidy.com/it/479.html)