近期计划攻坚一个新的专题系列——如何基于 golang 从零到一实现 redis.

这里选择的学习素材是 hdt3213 大佬于 github 上开源的 godis 项目. 在大佬的实现中，充分利用了 golang 的特性，将 redis 存储层由单线程模型转为并发模型，其中在实现数据的 expire 机制时，采用的是单机时间轮模型进行过期数据的删除操作.

godis 项目开源地址：http://github.com/HDT3213/godis

godis 时间轮代码：http://github.com/HDT3213/godis/blob/master/lib/timewheel/timewheel.go

实际上，在我之前分享个人项目——分布式定时器 xtimer 中也有到了时间轮算法，这部分内容具有一定共性. 借此机会，本期单独对时间轮算法的原理和实践内容进行梳理，并基于 golang 从零到一开源实现出一个单机版和 redis 分布式版的时间轮，供大家一起交流探讨.

个人开源的时间轮项目地址为：http://github.com/xiaoxuxiansheng/timewheel

本期内容的目录大纲如下：

1 时间轮原理

时间轮是定时任务调度系统中常用到的一种经典的算法模型. 有关时间轮算法的详细概念介绍，可以参见论文:《Hashed and Hierarchical Time Wheels: Data Structures for the Efficient Implementation of a Time Facility》http://www.cs.columbia.edu/~nahum/w6998/papers/sosp87-timing-wheels.pdf

接下来我也会从个人角度出发，谈谈我对于时间轮算法的一些浅显理解.

1.1 时间轮概念

聊时间轮之前，先聊聊时间的概念.

首先时间是一维、单向的，我们可以用一条一维的时间轴将其具象化. 我们对时间轴进行刻度拆分，每个刻度对应一个时间范围，那么刻度拆分得越细，则表示的时间范围越精确.

然而，我们知道时间是没有尽头的，因此这条一维时间轴的长度是无穷大的. 倘若我们想要建立一个数据结构去表达这条由一系列刻度聚合形成的时间轴，这个数据结构所占用的空间也是无穷无尽的.

那么，我们应该如何优化这个问题呢？此时，大家不妨低头看一眼自己的手表. 手表或时钟这类日常生活中用来关联表达时间的工具，采用的是首尾衔接的环状结构来替代无穷长度的一维时间轴，每当钟表划过一圈，刻度从新回到零值，但是已有的时间进度会传承往下.

本期所聊的时间轮算法采用的思路正是与之类似，下面我们梳理一下核心流程：

• 建立一个环状数据结构
• 每个刻度对应一个时间范围
• 创建定时任务时，根据距今的相对时长，推算出需要向后推移的刻度值
• 倘若来到环形数组的结尾，则重新从起点开始计算，但是记录时把执行轮次数加1
• 一个刻度可能存在多笔定时任务，所以每个刻度需要挂载一个定时任务链表

接下来，我们建立时间轮的扫描机制，就如同钟表中的指针一般，按照固定的时间节奏，沿着环形数组周而复始地持续向下扫描. 每当来到一个刻度时，则取出链表中轮次为 0 的定时任务进行执行. 这就是时间轮算法的核心思路.

1.2 多级时间轮

接下来聊一聊时间轮中的等级制度与多级时间轮的概念.

首先捋一捋，时间轮每个周期轮次中，使用的数据结构容量与所表达的时间范围之间的关系.

我们把时间轮中的每个刻度记为一个 slot，每个 slot 表示的时间范围记为 t.

假设时间轮中总共包含 2m 个 slot，求问如何组织我们的时间轮数据结构，能够使得时间轮每个轮次对应表达的时间范围尽可能的长. （一个轮次对应的时间范围越长，在时间流逝过程中轮次的迭代速度就越慢，于是每个 slot 对应的定时任务链表长度就越短，执行定时任务时的检索效率就越高.）

这里最简单的方式就是进行采用一维纵向排列的方式，那么能够表达的时间范围就是 2m * t，某个刻度对应的时间值就记为 {slot_i}.

另一种思路是，我们在时间轮中建立一种等级秩序.

比如我们将 2m 个 slot 拆成两个等级——level1 和 level2. 最终我们通过 {level1_slot}_{level2_slot} 的方式进行时间的表达.

我们给 level2 分配 m 个 slot，其中每个 slot 对应的时间范围同样为 t. 而 level1 同样也分配 m 个 slot，但是此时其中每个 slot 对应的时间范围应该为 m * t，因为在 level1 中的 slot 确定时，level2 中还有 m 种 slot 的组合方式.

如此一来，这种组织方式下，时间轮单个轮次所能表达的时间范围就是 m * m * t.

这里探讨的核心不是具体某级时间轮的时间范围结果，而是抛出了一种多级时间轮的思路，从一到二是质变，从二到三、从三到四就仅仅是量变的问题，可以继续复刻相同的思路.

回过头来看，我们会发现日常使用的时间表达式正是采用了这样一种多级时间轮的等级制度，比如当前的时刻为：2023-09-23 15:50:00. 这本质上是一种通过 {year}-{month}-{date}-{hour}-{minute}-{second} 组成的 6 级时间轮等级结构.

后续在本文第 3 章探讨如何基于 redis zset 实现时间轮的话题中，我们会进一步利用这种多级时间轮的思路，将每个任务首先基于前 5 级 {year}-{month}-{date}-{hour}-{minute} 的等级表达式进行分钟级时间片的纵向拆分，最终在 1 分钟范围内进行定时任务的有序组织，保证在每次插入、删除和检索任务时，处理的数据量级能够维持在分钟级的数量，最大化地提高时间轮结果的处理性能.

2 单机版实现

聊完原理部分，下面我们一起进入实战环节.

在本章中，我们会使用 golang 标准库的定时器工具 time ticker 结合环状数组的设计思路，实现一个单机版的单级时间轮.

2.1 核心类

2.1.1 时间轮

在对时间轮的类定义中，核心字段如下图所示：

在几个核心字段中：

• slots——类似于时钟的表盘
• curSlot——类似于时钟的指针
• ticker 是使用 golang 标准库的定时器工具，类似于驱动指针运转的齿轮

在创建时间轮实例时，会通过一个异步的常驻 goroutine 执行定时任务的检索、添加、删除等操作，并通过几个 channel 进行 goroutine 的执行逻辑和生命周期的控制：

• stopc：用于停止 goroutine
• addTaskCh：用于接收创建定时器指令
• removeTaskCh：用于接收删除定时任务的指令

// 单机版时间轮
type TimeWheel struct {
    // 单例工具，保证时间轮停止操作只能执行一次
    sync.Once
    // 时间轮运行时间间隔
    interval     time.Duration
    // 时间轮定时器
    ticker       *time.Ticker
    // 停止时间轮的 channel
    stopc        chan struct{}
    // 新增定时任务的入口 channel  
    addTaskCh    chan *taskElement
    // 删除定时任务的入口 channel
    removeTaskCh chan string
    // 通过 list 组成的环状数组. 通过遍历环状数组的方式实现时间轮
    // 定时任务数量较大，每个 slot 槽内可能存在多个定时任务，因此通过 list 进行组装
    slots        []*list.List
    // 当前遍历到的环状数组的索引
    curSlot      int
    // 定时任务 key 到任务节点的映射，便于在 list 中删除任务节点
    keyToETask   map[string]*list.Element
}

此处有几个技术细节需要提及：

首先：所谓环状数组指的是逻辑意义上的. 在实际的实现过程中，会通过一个定长数组结合循环遍历的方式，来实现这个逻辑意义上的“环状”性质.

其次：数组每一轮能表达的时间范围是固定的. 每当在添加添加一个定时任务时，需要根据其延迟的相对时长推算出其所处的 slot 位置，其中可能跨遍历轮次的情况，这时候需要额外通过定时任务中的 cycle 字段来记录这一信息，避免定时任务被提前执行.

最后：时间轮中一个 slot 可能需要挂载多笔定时任务，因此针对每个 slot，需要采用 golang 标准库 container/list 中实现的双向链表进行定时任务数据的存储.

2.1.2 定时任务

下面是对一笔定时任务的类定义：

• key：每个定时任务的全局唯一标识键
• task：包含了定时任务执行逻辑的闭包函数
• pos：定时任务在环形数组所处的位置，即数组的索引 index
• cycle：定时任务的延迟轮次. 时间轮的 curSlot 指针每完成一整轮的数组遍历，所有定时任务的 cycle 指数都需要减 1. 当定时任务 cycle 指数为 0 时，代表该任务在当前遍历轮次执行.

// 封装了一笔定时任务的明细信息
type taskElement struct {
    // 内聚了定时任务执行逻辑的闭包函数
    task  func()
    // 定时任务挂载在环状数组中的索引位置
    pos   int
    // 定时任务的延迟轮次. 指的是 curSlot 指针还要扫描过环状数组多少轮，才满足执行该任务的条件
    cycle int
    // 定时任务的唯一标识键
    key   string
}

2.2 构造器

在创建时间轮的构造器函数中，需要传入两个入参：

• slotNum：由使用方指定 slot 的个数，默认为 10
• interval：由使用方指定每个 slot 对应的时间范围，默认为 1 秒

初始化时间轮实例的过程中，会完成定时器 ticker 以及各个 channel 的初始化，并针对数组 中的各个 slot 进行初始化，每个 slot 位置都需要填充一个 list.

每个时间轮实例都会异步调用 run 方法，启动一个常驻 goroutine 用于接收和处理定时任务.

// 创建单机版时间轮 slotNum——时间轮环状数组长度  interval——扫描时间间隔
func NewTimeWheel(slotNum int, interval time.Duration) *TimeWheel {
    // 环状数组长度默认为 10
    if slotNum <= 0 {
        slotNum = 10
    }
    // 扫描时间间隔默认为 1 秒
    if interval <= 0 {
        interval = time.Second
    }


    // 初始化时间轮实例
    t := TimeWheel{
        interval:     interval,
        ticker:       time.NewTicker(interval),
        stopc:        make(chan struct{}),
        keyToETask:   make(map[string]*list.Element),
        slots:        make([]*list.List, 0, slotNum),
        addTaskCh:    make(chan *taskElement),
        removeTaskCh: make(chan string),
    }
    for i := 0; i < slotNum; i++ {
        t.slots = append(t.slots, list.New())
    }
    
    // 异步启动时间轮常驻 goroutine
    go t.run()
    return &t
}

2.3 启动与停止

时间轮运行的核心逻辑位于 timeWheel.run 方法中，该方法会通过 for 循环结合 select 多路复用的方式运行，属于 golang 中非常常见的异步编程风格.

goroutine 运行过程中需要从以下四类 channel 中接收不同的信号，并进行逻辑的分发处理：

• stopc：停止时间轮，使得当前 goroutine 退出
• ticker：接收到 ticker 的信号说明时间由往前推进了一个 interval，则需要批量检索并执行当前 slot 中的定时任务. 并推进指针 curSlot 往前偏移
• addTaskCh：接收创建定时任务的指令
• removeTaskCh：接收删除定时任务的指令

此处值得一提的是，后续不论是创建、删除还是检索定时任务，都是通过这个常驻 goroutine 完成的，因此在访问一些临界资源的时候，不需要加锁，因为不存在并发访问的情况

// 运行时间轮
func (t *TimeWheel) run() {
    defer func() {
        if err := recover(); err != nil {
            // ...
        }
    }()
 
    // 通过 for + select 的代码结构运行一个常驻 goroutine 是常规操作
    for {
        select {
        // 停止时间轮
        case <-t.stopc:
            return
        // 接收到定时信号
        case <-t.ticker.C:
            // 批量执行定时任务
            t.tick()
        // 接收创建定时任务的信号
        case task := <-t.addTaskCh:
            t.addTask(task)
        // 接收到删除定时任务的信号
        case removeKey := <-t.removeTaskCh:
            t.removeTask(removeKey)
        }
    }
}

时间轮提供了一个 Stop 方法，用于手动停止时间轮，回收对应的 goroutine 和 ticker 资源.

停止时间轮的操作是通过关闭 stopc channel 完成的，由于 channel 不允许被反复关闭，因此这里通过 sync.Once 保证该逻辑只被调用一次.

// 停止时间轮
func (t *TimeWheel) Stop() {
    // 通过单例工具，保证 channel 只能被关闭一次，避免 panic
    t.Do(func() {
        // 定制定时器 ticker
        t.ticker.Stop()
        // 关闭定时器运行的 stopc
        close(t.stopc)
    })
}

2.4 创建任务

创建一笔定时任务的核心步骤如下：

• 使用方往 addTaskCh 中投递定时任务，由常驻 goroutine 接收定时任务
• 根据执行时间，推算出定时任务所处的 slot 位置以及需要延迟的轮次 cycle
• 将定时任务包装成一个 list node，追加到对应 slot 位置的 list 尾部
• 以定时任务唯一键为 key，list node 为 value，在 keyToETask map 中建立映射关系，方便后续删除任务时使用

// 添加定时任务到时间轮中
func (t *TimeWheel) AddTask(key string, task func(), executeAt time.Time) {
    // 根据执行时间推算得到定时任务从属的 slot 位置，以及需要延迟的轮次
    pos, cycle := t.getPosAndCircle(executeAt)
    // 将定时任务通过 channel 进行投递
    t.addTaskCh <- &taskElement{
        pos:   pos,
        cycle: cycle,
        task:  task,
        key:   key,
    }
}

// 根据执行时间推算得到定时任务从属的 slot 位置，以及需要延迟的轮次
func (t *TimeWheel) getPosAndCircle(executeAt time.Time) (int, int) {
    delay := int(time.Until(executeAt))
    // 定时任务的延迟轮次
    cycle := delay / (len(t.slots) * int(t.interval))
    // 定时任务从属的环状数组 index
    pos := (t.curSlot + delay/int(t.interval)) % len(t.slots)
    return pos, cycle
}

// 常驻 goroutine 接收到创建定时任务后的处理逻辑
func (t *TimeWheel) addTask(task *taskElement) {
    // 获取到定时任务从属的环状数组 index 以及对应的 list
    list := t.slots[task.pos]
    // 倘若定时任务 key 之前已存在，则需要先删除定时任务
    if _, ok := t.keyToETask[task.key]; ok {
        t.removeTask(task.key)
    }
    // 将定时任务追加到 list 尾部
    eTask := list.PushBack(task)
    // 建立定时任务 key 到将定时任务所处的节点
    t.keyToETask[task.key] = eTask
}

2.5 删除任务

删除一笔定时任务的核心步骤如下：

• 使用方往 removeTaskCh 中投递删除任务的 key，由常驻 goroutine 接收处理
• 从 keyToETask map 中，找到该任务对应的 list node
• 从 keyToETask map 中移除该组 kv 对
• 从对应 slot 的 list 中移除该 list node

// 删除定时任务，投递信号
func (t *TimeWheel) RemoveTask(key string) {
    t.removeTaskCh <- key
}

// 时间轮常驻 goroutine 接收到删除任务信号后，执行的删除任务逻辑
func (t *TimeWheel) removeTask(key string) {
    eTask, ok := t.keyToETask[key]
    if !ok {
        return
    }
    // 将定时任务节点从映射 map 中移除
    delete(t.keyToETask, key)
    // 获取到定时任务节点后，将其从 list 中移除
    task, _ := eTask.Value.(*taskElement)
    _ = t.slots[task.pos].Remove(eTask)
}

2.6 执行定时任务

最后来捋一下最核心的链路——检索并批量执行定时任务的流程.

首先，每当接收到 ticker 信号时，会根据当前的 curSlot 指针，获取到对应 slot 位置挂载的定时任务 list，调用 execute 方法执行其中的定时任务. 最后通过 circularIncr 方法推进 curSlot 指针向前移动.

// 常驻 goroutine 每次接收到定时信号后用于执行定时任务的逻辑
func (t *TimeWheel) tick() {
    // 根据 curSlot 获取到当前所处的环状数组索引位置，取出对应的 list
    list := t.slots[t.curSlot]
     // 在方法返回前，推进 curSlot 指针的位置，进行环状遍历
    defer t.circularIncr()
    // 批量处理满足执行条件的定时任务
    t.execute(list)
}

在 execute 方法中，会对 list 中的定时任务进行遍历：

• 对于 cycle > 0 的定时任务，说明当前还未达到执行条件，需要将其 cycle 值减 1，留待后续轮次再处理
• 对于 cycle = 0 的定时任务，开启一个 goroutine ，执行其中的闭包函数 task，并将其从 list 和 map 中移除

// 执行定时任务，每次处理一个 list
func (t *TimeWheel) execute(l *list.List) {
    // 遍历 list
    for e := l.Front(); e != nil; {
        // 获取到每个节点对应的定时任务信息
        taskElement, _ := e.Value.(*taskElement)
        // 倘若任务还存在延迟轮次，则只对 cycle 计数器进行扣减，本轮不作任务的执行
        if taskElement.cycle > 0 {
            taskElement.cycle--
            e = e.Next()
            continue
        }


        // 当前节点对应定时任务已达成执行条件，开启一个 goroutine 负责执行任务
        go func() {
            defer func() {
                if err := recover(); err != nil {
                    // ...
                }
            }()
            taskElement.task()
        }()


        // 任务已执行，需要把对应的任务节点从 list 中删除
        next := e.Next()
        l.Remove(e)
        // 把任务 key 从映射 map 中删除
        delete(t.keyToETask, taskElement.key)
        e = next
    }
}

在 circularIncr 方法中，呼应了环状数组的逻辑处理方式：

// 每次 tick 后需要推进 curSlot 指针的位置，slots 在逻辑意义上是环状数组，所以在到达尾部时需要从新回到头部 
func (t *TimeWheel) circularIncr() {
    t.curSlot = (t.curSlot + 1) % len(t.slots)
}

3 分布式版实现

本章我们讨论一下，如何基于 redis 实现分布式版本的时间轮，以贴合实际生产环境对分布式定时任务调度系统的诉求.

redis 版时间轮的实现思路是使用 redis 中的有序集合 sorted set（简称 zset） 进行定时任务的存储管理，其中以每个定时任务执行时间对应的时间戳作为 zset 中的 score，完成定时任务的有序排列组合.

zset 数据结构的 redis 官方文档链接：https://redis.io/docs/data-types/sorted-sets/

这里有两个的技术细节需要提前和大家同步：

• 分钟级时间分片：为了避免产生 redis 大 key 问题，此处采用本文 1.2 小节中提到的多级时间轮等级制度，以分钟的维度进行时间片的纵向划分，每个分钟级时间片对应一个独立的 zset 有序表，保证每次执行任务时处理的数据规模仅为分钟的量级
• 惰性删除机制：为了简化删除定时任务的流程. 在使用方指定删除定时任务时，我们不直接从 zset 中删除数据，而是额外记录一个已删除任务的 set. 后续在检索定时任务时，通过使用 set 进行定时任务的过滤，实现定时任务的惰性删除.

3.1 核心类

3.1.1 redis 时间轮

在 redis 版时间轮中有两个核心类，第一个是关于时间轮的类定义：

• redisClient：定时任务的存储是基于 redis zset 实现的，因此需要内置一个 redis 客户端，这部分在 3.2 小节展开；
• httpClient：定时任务执行时，是通过请求使用方预留回调地址的方式实现的，因此需要内置一个 http 客户端
• channel × 2：ticker 和 stopc 对应为 golang 标准库定时器以及停止 goroutine 的控制器

// 基于 redis 实现的分布式版时间轮
type RTimeWheel struct {
    // 内置的单例工具，用于保证 stopc 只被关闭一次
    sync.Once
    // redis 客户端
    redisClient *redis.Client
    // http 客户端. 在执行定时任务时需要使用到.
    httpClient  *thttp.Client
    // 用于停止时间轮的控制器 channel
    stopc       chan struct{
    // 触发定时扫描任务的定时器 
    ticker      *time.Ticker
}

3.1.2 定时任务

定时任务的类型定义如下，其中包括定时任务的唯一键 key，以及执行定时任务回调时需要使用到的 http 协议参数.

// 使用方提交的每一笔定时任务
type RTaskElement struct {
    // 定时任务全局唯一 key
    Key         string            `json:"key"`
    // 定时任务执行时，回调的 http url
    CallbackURL string            `json:"callback_url"`
    // 回调时使用的 http 方法
    Method      string            `json:"method"`
    // 回调时传递的请求参数
    Req         interface{}       `json:"req"`
    // 回调时使用的 http 请求头
    Header      map[string]string `json:"header"`
}

3.2 redis lua 使用事项

本项目使用的 redis 客户端是我个人基于 golang-redis 客户端 sdk——redigo 进一步封装实现的，redigo 的开源地址为： https://github.com/gomodule/redigo

此处主要在 redisClient 中封装了一个 Eval 方法，便于使用 redis lua 脚本执行复合指令.

// Eval：执行 lua 脚本.
// src——lua 脚本内容 keyCount——key 的数量 keysAndArgs——由 key 和 args 组成的列表
func (c *Client) Eval(ctx context.Context, src string, keyCount int, keysAndArgs []interface{}) (interface{}, error) {
    args := make([]interface{}, 2+len(keysAndArgs))
    args[0] = src
    args[1] = keyCount
    copy(args[2:], keysAndArgs)


    // 从 redis 链接池中获取一个连接
    conn, err := c.pool.GetContext(ctx)
    if err != nil {
        return -1, err
    }
    // 使用完成后将连接放回池子
    defer conn.Close()
    // 执行 lua 脚本
    return conn.Do("EVAL", args...)
}

lua 脚本是 redis 的高级功能，能够保证针在单个 redis 节点内执行的一系列指令具备原子性，中途不会被其他操作者打断.

redis lua 功能介绍：https://redis.io/docs/interact/programmability/eval-intro/

lua 语法教程：https://www.runoob.com/lua/lua-tutorial.html

此处之所以需要使用 lua 脚本，是因为在实现时间轮的过程中，存在一系列本身不具备原子性但在业务流程中不可拆解的复合操作，需要由 lua 脚本赋予其原子性质.

在使用 lua 时，尤其需要注意的点是，只有操作的数据属于单个 redis 节点时，才能保证其原子性. 然而在生产环境中，redis 通常采用纵向分治的集群模式，这使得 key 不同的数据可能被分发在不同的 redis 节点上，此时 lua 脚本的性质就无法保证.

在使用 lua 脚本时，倘若一系列复合操作都是针对于同一个 key，那么数据必然位于同一个节点，没有任何疑议. 倘若我们在 lua 中涉及到对多个 key 的操作，那么这些 key 对应的数据就可能从属于不同的 redis 节点，此时 lua 脚本存在失效的风险.

针对这个问题，本项目采取的是定制的分区策略，来保证指定的 key 一定被分发到相同的 redis 节点上. 此处使用的方式是通过 "{}" 进行 hash_tag 的标识，所有拥有相同 hash_tag 的 key 都一定会被分发到相同的节点上.

该分区策略可参见 redis 官方文档：https://redis.io/commands/cluster-keyslot/

指令示例如下：

> CLUSTER KEYSLOT somekey
(integer) 11058
> CLUSTER KEYSLOT foo{hash_tag}
(integer) 2515
> CLUSTER KEYSLOT bar{hash_tag}
(integer) 2515

3.3 构造器

在构造时间轮实例时，使用方需要注入 redis 客户端以及 http 客户端.

在初始化流程中，ticker 为 golang 标准库实现的定时器，定时器的执行时间间隔固定为 1 s. 此外会异步运行 run 方法，启动一个常驻 goroutine，生命周期会通过 stopc channel 进行控制.

// 构造 redis 实现的分布式时间轮
func NewRTimeWheel(redisClient *redis.Client, httpClient *thttp.Client) *RTimeWheel {
    // 创建时间轮实例
    r := RTimeWheel{
        // 创建定时器，每隔 1 s 执行一次
        ticker:      time.NewTicker(time.Second),
        redisClient: redisClient,
        httpClient:  httpClient,
        stopc:       make(chan struct{}),
    }


    // 异步启动时间轮
    go r.run()
    return &r
}

3.4 启动与停止

时间轮常驻 goroutine 运行流程同样通过 for + select 的形式运行：

• 接收到 stopc 信号时，goroutine 退出，时间轮停止运行
• 接收到 ticker 信号时，开启一个异步 goroutine 用于执行当前批次的定时任务

// 运行时间轮
func (r *RTimeWheel) run() {
    // 通过 for + select 的代码结构运行一个常驻 goroutine 是常规操作
    for {
        select {
        // 接收到终止信号，则退出 goroutine
        case <-r.stopc:
            return
        // 每次接收到来自定时器的信号，则批量扫描并执行定时任务
        case <-r.ticker.C:
            // 每次 tick 获取任务
            go r.executeTasks()
        }
    }
}

停止时间轮的 Stop 方法通过关闭 stopc 保证常驻 goroutine 能够及时退出.

// 停止时间轮
func (r *RTimeWheel) Stop() {
    // 基于单例工具，保证 stopc 只能被关闭一次
    r.Do(func() {
        // 关闭 stopc，使得常驻 goroutine 停止运行
        close(r.stopc)
        // 终止定时器 ticker
        r.ticker.Stop()
    })
}

3.5 创建任务

在创建定时任务时，每笔定时任务需要根据其执行的时间找到从属的分钟时间片.

定时任务真正的存储逻辑定义在一段 lua 脚本中，通过 redis 客户端的 Eval 方法执行.

// 添加定时任务
func (r *RTimeWheel) AddTask(ctx context.Context, key string, task *RTaskElement, executeAt time.Time) error {
    // 前置对定时任务的参数进行校验
    if err := r.addTaskPrecheck(task); err != nil {
        return err
    }


    task.Key = key
    // 将定时任务序列化成字节数组
    taskBody, _ := json.Marshal(task)
    // 通过执行 lua 脚本，实现将定时任务添加 redis zset 中. 本质上底层使用的是 zadd 指令.
    _, err := r.redisClient.Eval(ctx, LuaAddTasks, 2, []interface{}{
        // 分钟级 zset 时间片
        r.getMinuteSlice(executeAt),
        // 标识任务删除的集合
        r.getDeleteSetKey(executeAt),
        // 以执行时刻的秒级时间戳作为 zset 中的 score
        executeAt.Unix(),
        // 任务明细
        string(taskBody),
        // 任务 key，用于存放在删除集合中
        key,
    })
    return err
}

下面展示的是获取分钟级定时任务有序表 minuteSlice 以及已删除任务集合 deleteSet 的细节.

此处呼应了 3.3 小节，通过以分钟级表达式作为 {hash_tag} 的方式，确保 minuteSlice 和 deleteSet 一定会分发到相同的 redis 节点之上，进一步保证 lua 脚本的原子性能够生效.

• 获取定时任务有序表 key 的方法：

func (r *RTimeWheel) getMinuteSlice(executeAt time.Time) string {
    return fmt.Sprintf("xiaoxu_timewheel_task_{%s}", util.GetTimeMinuteStr(executeAt))
}

• 获取删除任务集合 key 的方法：

func (r *RTimeWheel) getDeleteSetKey(executeAt time.Time) string {
    return fmt.Sprintf("xiaoxu_timewheel_delset_{%s}", util.GetTimeMinuteStr(executeAt))
}

下面展示一下创建定时任务流程中 lua 脚本的执行逻辑：

const (
    // 添加任务时，如果存在删除 key 的标识，则将其删除
    // 添加任务时，根据时间（所属的 min）决定数据从属于哪个分片{}
    LuaAddTasks = `
       -- 获取的首个 key 为 zset 的 key
       local zsetKey = KEYS[1]
       -- 获取的第二个 key 为标识已删除任务 set 的 key
       local deleteSetKey = KEYS[2]
       -- 获取的第一个 arg 为定时任务在 zset 中的 score
       local score = ARGV[1]
       -- 获取的第二个 arg 为定时任务明细数据
       local task = ARGV[2]
       -- 获取的第三个 arg 为定时任务唯一键，用于将其从已删除任务 set 中移除
       local taskKey = ARGV[3]
       -- 每次添加定时任务时，都直接将其从已删除任务 set 中移除，不管之前是否在 set 中
       redis.call('srem',deleteSetKey,taskKey)
       -- 调用 zadd 指令，将定时任务添加到 zset 中
       return redis.call('zadd',zsetKey,score,task)
    `
)

3.6 删除任务

删除定时任务的方式是将定时任务追加到分钟级的已删除任务 set 中. 之后在检索定时任务时，会根据这个 set 对定时任务进行过滤，实现惰性删除机制.

// 从 redis 时间轮中删除一个定时任务
func (r *RTimeWheel) RemoveTask(ctx context.Context, key string, executeAt time.Time) error {
    // 执行 lua 脚本，将被删除的任务追加到 set 中.
    _, err := r.redisClient.Eval(ctx, LuaDeleteTask, 1, []interface{}{
        r.getDeleteSetKey(executeAt),
        key,
    })
    return err
}

lua 执行逻辑如下：

const(    
    // 删除定时任务 lua 脚本
    LuaDeleteTask = `
       -- 获取标识删除任务的 set 集合的 key
       local deleteSetKey = KEYS[1]
       -- 获取定时任务的唯一键
       local taskKey = ARGV[1]
       -- 将定时任务唯一键添加到 set 中
       redis.call('sadd',deleteSetKey,taskKey)
       -- 倘若是 set 中的首个元素，则对 set 设置 120 s 的过期时间
       local scnt = redis.call('scard',deleteSetKey)
       if (tonumber(scnt) == 1)
       then
           redis.call('expire',deleteSetKey,120)
       end
       return scnt
)    `

3.7 执行定时任务

在执行定时任务时，会通过 getExecutableTasks 方法批量获取到满足执行条件的定时任务 list，然后并发调用 execute 方法完成定时任务的回调执行.

// 批量执行定时任务
func (r *RTimeWheel) executeTasks() {
    defer func() {
        if err := recover(); err != nil {
            // log
        }
    }()


    // 并发控制，保证 30 s 之内完成该批次全量任务的执行，及时回收 goroutine，避免发生 goroutine 泄漏
    tctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
    defer cancel()
    // 根据当前时间条件扫描 redis zset，获取所有满足执行条件的定时任务
    tasks, err := r.getExecutableTasks(tctx)
    if err != nil {
        // log
        return
    }


    // 并发执行任务，通过 waitGroup 进行聚合收口
    var wg sync.WaitGroup
    for _, task := range tasks {
        wg.Add(1)
        // shadow
        task := task
        go func() {
            defer func() {
                if err := recover(); err != nil {
                }
                wg.Done()
            }()
            // 执行定时任务
            if err := r.executeTask(tctx, task); err != nil {
                // log
            }
        }()
    }
    wg.Wait()
}

3.8 检索定时任务

最后介绍一下，如何根据当前时间获取到满足执行条件的定时任务列表：

• 每次检索时，首先根据当前时刻，推算出所从属的分钟级时间片
• 然后获得当前的秒级时间戳，作为 zrange 指令检索的 score 范围
• 调用 lua 脚本，同时获取到已删除任务 set 以及 score 范围内的定时任务 list.
• 通过 set 过滤掉被删除的任务，然后返回满足执行条件的定时任务

func (r *RTimeWheel) getExecutableTasks(ctx context.Context) ([]*RTaskElement, error) {
    now := time.Now()
    // 根据当前时间，推算出其从属的分钟级时间片
    minuteSlice := r.getMinuteSlice(now)
    // 推算出其对应的分钟级已删除任务集合
    deleteSetKey := r.getDeleteSetKey(now)
    nowSecond := util.GetTimeSecond(now)
    // 以秒级时间戳作为 score 进行 zset 检索
    score1 := nowSecond.Unix()
    score2 := nowSecond.Add(time.Second).Unix()
    // 执行 lua 脚本，本质上是通过 zrange 指令结合秒级时间戳对应的 score 进行定时任务检索
    rawReply, err := r.redisClient.Eval(ctx, LuaZrangeTasks, 2, []interface{}{
        minuteSlice, deleteSetKey, score1, score2,
    })
    if err != nil {
        return nil, err
    }


    // 结果中，首个元素对应为已删除任务的 key 集合，后续元素对应为各笔定时任务
    replies := gocast.ToInterfaceSlice(rawReply)
    if len(replies) == 0 {
        return nil, fmt.Errorf("invalid replies: %v", replies)
    }


    deleteds := gocast.ToStringSlice(replies[0])
    deletedSet := make(map[string]struct{}, len(deleteds))
    for _, deleted := range deleteds {
        deletedSet[deleted] = struct{}{}
    }


    // 遍历各笔定时任务，倘若其存在于删除集合中，则跳过，否则追加到 list 中返回，用于后续执行
    tasks := make([]*RTaskElement, 0, len(replies)-1)
    for i := 1; i < len(replies); i++ {
        var task RTaskElement
        if err := json.Unmarshal([]byte(gocast.ToString(replies[i])), &task); err != nil {
            // log
            continue
        }


        if _, ok := deletedSet[task.Key]; ok {
            continue
        }
        tasks = append(tasks, &task)
    }


    return tasks, nil
}

lua 脚本的执行逻辑如下：

(    
    // 扫描 redis 时间轮. 获取分钟范围内,已删除任务集合 以及在时间上达到执行条件的定时任务进行返回
    LuaZrangeTasks = `
       -- 第一个 key 为存储定时任务的 zset key
       local zsetKey = KEYS[1]
       -- 第二个 key 为已删除任务 set 的 key
       local deleteSetKey = KEYS[2]
       -- 第一个 arg 为 zrange 检索的 score 左边界
       local score1 = ARGV[1]
       -- 第二个 arg 为 zrange 检索的 score 右边界
       local score2 = ARGV[2]
       -- 获取到已删除任务的集合
       local deleteSet = redis.call('smembers',deleteSetKey)
       -- 根据秒级时间戳对 zset 进行 zrange 检索，获取到满足时间条件的定时任务
       local targets = redis.call('zrange',zsetKey,score1,score2,'byscore')
       -- 检索到的定时任务直接从时间轮中移除，保证分布式场景下定时任务不被重复获取
       redis.call('zremrangebyscore',zsetKey,score1,score2)
       -- 返回的结果是一个 table
       local reply = {}
       -- table 的首个元素为已删除任务集合
       reply[1] = deleteSet
       -- 依次将检索到的定时任务追加到 table 中
       for i, v in ipairs(targets) do
           reply[#reply+1]=v
       end
       return reply
    `
)

4 总结

本期和大家探讨了如何基于 golang 从零到一实现时间轮算法，通过原理结合源码，详细展示了单机版和 redis 分布式版时间轮的实现方式.

最后我们展望一下，时间轮算法在工程实践中具体能得到怎样的应用呢？

此前我基于 redis zset 时间轮的模型实现了一个分布式定时任务调度系统 xtimer，大家感兴趣的话可以阅读一下我之前分享的这篇文章：基于协程池架构实现的分布式定时器 XTimer.

该项目源码已于 github 开源：http://github.com/xiaoxuxiansheng/xtimer

1、grep

1.1 什么是grep和egrep

Linux系统中grep命令是一种强大的文本搜索工具，它能使用正则表达式搜索文本，并把匹配的行打印出来（匹配到的标红）。grep全称是Global Regular Expression Print，表示全局正则表达式版本，它的使用权限是所有用户。

grep的工作方式是这样的，它在一个或多个文件中搜索字符串模板。如果模板包括空格，则必须被引用，模板后的所有字符串被看作文件名。搜索的结果被送到标准输出，不影响原文件内容。

grep可用于shell脚本，因为grep通过返回一个状态值来说明搜索的状态，如果模板搜索成功，则返回0，如果搜索不成功，则返回1，如果搜索的文件不存在，则返回2。我们利用这些返回值就可进行一些自动化的文本处理工作。

egrep = grep -E：扩展的正则表达式 （除了\< , \> , \b 使用其他正则都可以去掉\）

1.2 使用grep

1.2.1 命令格式

1.2.2 命令功能

用于过滤/搜索的特定字符。可使用正则表达式能多种命令配合使用，使用上十分灵活。

1.2.3 命令参数

常用参数已加粗

• -A<显示行数>：除了显示符合范本样式的那一列之外，并显示该行之后的内容。
• -B<显示行数>：除了显示符合样式的那一行之外，并显示该行之前的内容。
• -C<显示行数>：除了显示符合样式的那一行之外，并显示该行之前后的内容。
• -c：统计匹配的行数
• -e ：实现多个选项间的逻辑or 关系
• -E：扩展的正则表达式
• -f FILE：从FILE获取PATTERN匹配
• -F ：相当于fgrep
• -i --ignore-case #忽略字符大小写的差别。
• -n：显示匹配的行号
• -o：仅显示匹配到的字符串
• -q：静默模式，不输出任何信息
• -s：不显示错误信息。
• -v：显示不被pattern 匹配到的行，相当于[^] 反向匹配
• -w ：匹配 整个单词

1.3 grep实战演示

2、正则表达式

2.1 认识正则

（1）介绍

正则表达式应用广泛，在绝大多数的编程语言都可以完美应用，在Linux中，也有着极大的用处。

使用正则表达式，可以有效的筛选出需要的文本，然后结合相应的支持的工具或语言，完成任务需求。

在本篇博客中，我们使用grep/egrep来完成对正则表达式的调用

（2）正则表达式类型

正则表达式可以使用正则表达式引擎实现，正则表达式引擎是解释正则表达式模式并使用这些模式匹配文本的基础软件。

在Linux中，常用的正则表达式有：

• POSIX 基本正则表达式（BRE）引擎
• POSIX 扩展正则表达式（BRE）引擎

2.2 基本正则表达式

2.2.1 匹配字符

（1）格式

• . 匹配任意单个字符，不能匹配空行
• [] 匹配指定范围内的任意单个字符
• [^] 取反
• [:alnum:] 或 [0-9a-zA-Z]
• [:alpha:] 或 [a-zA-Z]
• [:upper:] 或 [A-Z]
• [:lower:] 或 [a-z]
• [:blank:] 空白字符（空格和制表符）
• [:space:] 水平和垂直的空白字符（比[:blank:]包含的范围广）
• [:cntrl:] 不可打印的控制字符（退格、删除、警铃...）
• [:digit:] 十进制数字 或[0-9]
• [:xdigit:]十六进制数字
• [:graph:] 可打印的非空白字符
• [:print:] 可打印字符
• [:punct:] 标点符号

（2）演示

2.2.2 配置次数

（1）格式

• * 匹配前面的字符任意次，包括0次，贪婪模式：尽可能长的匹配
• .* 任意长度的任意字符，不包括0次
• \? 匹配其前面的字符0 或 1次
• \+ 匹配其前面的字符至少1次
• \{n\} 匹配前面的字符n次
• \{m,n\} 匹配前面的字符至少m 次，至多n次
• \{,n\} 匹配前面的字符至多n次
• \{n,\} 匹配前面的字符至少n次

（2）演示

2.2.3 位置锚定：定位出现的位置

（1）格式

• ^ 行首锚定，用于模式的最左侧
• $ 行尾锚定，用于模式的最右侧
• ^PATTERN$，用于模式匹配整行
• ^$ 空行
• ^[[:space:]].*$ 空白行
• \< 或 \b 词首锚定，用于单词模式的左侧
• \> 或 \b 词尾锚定；用于单词模式的右侧
• \<PATTERN\>

（2）演示

2.2.4 分组和后向引用

（1）格式

① 分组：\(\) 将一个或多个字符捆绑在一起，当作一个整体进行处理

分组括号中的模式匹配到的内容会被正则表达式引擎记录于内部的变量中，这些变量的命名方式为: \1, \2, \3, ...

最近很多小伙伴找我要各种学习资料，但一个个私信太耽误大家时间了，于是我整理了一些优质经典学习资源，涵盖Linux、虚拟化、容器、云计算、网络、Python等书籍和视频，点击链接即可领取，祝学业有成！（无任何套路！）

10T学习资源大放送，涵盖Linux、虚拟化、容器、云计算、网络、Python、Go等书籍和视频​www.linuxyz.cn/2378.html

② 后向引用

引用前面的分组括号中的模式所匹配字符，而非模式本身

\1 表示从左侧起第一个左括号以及与之匹配右括号之间的模式所匹配到的字符

\2 表示从左侧起第2个左括号以及与之匹配右括号之间的模式所匹配到的字符，以此类推

\& 表示前面的分组中所有字符

③ 流程分析如下：

（2）演示

2.3 扩展正则表达式

（1）字符匹配：

• . 任意单个字符
• [] 指定范围的字符
• [^] 不在指定范围的字符
• 次数匹配：
• * ：匹配前面字符任意次
• ? : 0 或1次
• + ：1 次或多次
• {m} ：匹配m次 次
• {m,n} ：至少m ，至多n次

（2）位置锚定：

• ^ : 行首
• $ : 行尾
• \<, \b : 语首
• \>, \b : 语尾
• 分组：()
• 后向引用：\1, \2, ...

（3）总结

除了\<, \b : 语首、\>, \b : 语尾；使用其他正则都可以去掉\；上面有演示案例，不在进行演示

3、sed

3.1 认识sed

sed 是一种流编辑器，它一次处理一行内容。处理时，把当前处理的行存储在临时缓冲区中，称为“模式空间”（patternspace ），接着用sed 命令处理缓冲区中的内容，处理完成后，把缓冲区的内容送往屏幕。然后读入下行，执行下一个循环。如果没有使诸如‘D’ 的特殊命令，那会在两个循环之间清空模式空间，但不会清空保留空间。这样不断重复，直到文件末尾。文件内容并没有改变，除非你使用重定向存储输出或-i。

功能：主要用来自动编辑一个或多个文件, 简化对文件的反复操作

3.2 使用sed

3.2.1 命令格式

3.2.2 常用选项options

• -n：不输出模式空间内容到屏幕，即不自动打印，只打印匹配到的行
• -e：多点编辑，对每行处理时，可以有多个Script
• -f：把Script写到文件当中，在执行sed时-f 指定文件路径，如果是多个Script，换行写
• -r：支持扩展的正则表达式
• -i：直接将处理的结果写入文件
• -i.bak：在将处理的结果写入文件之前备份一份

3.2.3 地址定界

• 不给地址：对全文进行处理
• 单地址：
• #: 指定的行
• /pattern/：被此处模式所能够匹配到的每一行
• 地址范围：
• #,#
• #,+#
• /pat1/,/pat2/
• #,/pat1/
• ~：步进
• sed -n '1~2p' 只打印奇数行 （1~2 从第1行，一次加2行）
• sed -n '2~2p' 只打印偶数行

3.2.4 编辑命令command

• d：删除模式空间匹配的行，并立即启用下一轮循环
• p：打印当前模式空间内容，追加到默认输出之后
• a：在指定行后面追加文本，支持使用\n实现多行追加
• i：在行前面插入文本，支持使用\n实现多行追加
• c：替换行为单行或多行文本，支持使用\n实现多行追加
• w：保存模式匹配的行至指定文件
• r：读取指定文件的文本至模式空间中匹配到的行后
• =：为模式空间中的行打印行号
• !：模式空间中匹配行取反处理
• s///：查找替换，支持使用其它分隔符，如：s@@@，s###；
• 加g表示行内全局替换；
• 在替换时，可以加一下命令，实现大小写转换
• \l：把下个字符转换成小写。
• \L：把replacement字母转换成小写，直到\U或\E出现。
• \u：把下个字符转换成大写。
• \U：把replacement字母转换成大写，直到\L或\E出现。
• \E：停止以\L或\U开始的大小写转换

3.3 sed用法演示

3.3.1 常用选项options演示

3.3.2 地址界定演示

3.3.3 编辑命令command演示

3.4 sed高级编辑命令

（1）格式

• h：把模式空间中的内容覆盖至保持空间中
• H：把模式空间中的内容追加至保持空间中
• g：从保持空间取出数据覆盖至模式空间
• G：从保持空间取出内容追加至模式空间
• x：把模式空间中的内容与保持空间中的内容进行互换
• n：读取匹配到的行的下一行覆盖 至模式空间
• N：读取匹配到的行的下一行追加 至模式空间
• d：删除模式空间中的行
• D：删除 当前模式空间开端至\n 的内容（不再传 至标准输出），放弃之后的命令，但是对剩余模式空间重新执行sed

（2）一个案例+示意图演示

① 案例：倒序输出文本内容

② 示意图如下：

③ 总结模式空间与保持空间关系：

保持空间是模式空间一个临时存放数据的缓冲区，协助模式空间进行数据处理

（3）演示

① 显示偶数行

② 倒序显示

③ 显示奇数行

④ 显示最后一行

⑤ 每行之间加空行

⑥ 把每行内容替换成空行

⑦ 确保每一行下面都有一个空行

4、awk

4.1 认识awk

awk是一种编程语言，用于在linux/unix下对文本和数据进行处理。数据可以来自标准输入(stdin)、一个或多个文件，或其它命令的输出。它支持用户自定义函数和动态正则表达式等先进功能，是linux/unix下的一个强大编程工具。它在命令行中使用，但更多是作为脚本来使用。awk有很多内建的功能，比如数组、函数等，这是它和C语言的相同之处，灵活性是awk最大的优势。

awk其实不仅仅是工具软件，还是一种编程语言。不过，本文只介绍它的命令行用法，对于大多数场合，应该足够用了。

4.2 使用awk

4.2.1 语法

4.2.2 常用命令选项

• -F fs：fs指定输入分隔符，fs可以是字符串或正则表达式，如-F:
• -v var=value：赋值一个用户定义变量，将外部变量传递给awk
• -f scripfile：从脚本文件中读取awk命令

4.3 awk变量

变量：内置和自定义变量，每个变量前加 -v 命令选项

4.3.1 内置变量

（1）格式

• FS：输入字段分隔符，默认为空白字符
• OFS：输出字段分隔符，默认为空白字符
• RS ：输入记录分隔符，指定输入时的换行符，原换行符仍有效
• ORS ：输出记录分隔符，输出时用指定符号代替换行符
• NF ：字段数量，共有多少字段， $NF引用最后一列，$(NF-1)引用倒数第2列
• NR ：行号，后可跟多个文件，第二个文件行号继续从第一个文件最后行号开始
• FNR ：各文件分别计数, 行号，后跟一个文件和NR一样，跟多个文件，第二个文件行号从1开始
• FILENAME ：当前文件名
• ARGC ：命令行参数的个数
• ARGV ：数组，保存的是命令行所给定的各参数，查看参数

（2）演示

4.3.2 自定义变量

自定义变量( 区分字符大小写)

（1）-v var=value

① 先定义变量，后执行动作print

② 在执行动作print后定义变量

（2）在program 中直接定义

可以把执行的动作放在脚本中，直接调用脚本 -f

4.4 printf命令

比print更强大

4.4.1 格式

（1）格式化输出

① 必须指定FORMAT

② 不会自动换行，需要显式给出换行控制符，\n

③ FORMAT 中需要分别为后面每个item 指定格式符

（2）格式符：与item 一一对应

• %c: 显示字符的ASCII码
• %d, %i: 显示十进制整数
• %e, %E: 显示科学计数法数值
• %f ：显示为浮点数，小数 %5.1f，带整数、小数点、整数共5位，小数1位，不够用空格补上
• %g, %G ：以科学计数法或浮点形式显示数值
• %s ：显示字符串；例：%5s最少5个字符，不够用空格补上，超过5个还继续显示
• %u ：无符号整数
• %%: 显示% 自身

（3）修饰符：放在%c[/d/e/f...]之间

• #[.#]：第一个数字控制显示的宽度；第二个# 表示小数点后精度，%5.1f
• -：左对齐（默认右对齐） %-15s
• +：显示数值的正负符号 %+d

4.4.2 演示

4.5 操作符

4.5.1 格式

• 算术操作符：
• x+y, x-y, x*y, x/y, x^y, x%y
• -x: 转换为负数
• +x: 转换为数值
• 字符串操作符：没有符号的操作符，字符串连接
• 赋值操作符：
• =, +=, -=, *=, /=, %=, ^=
• ++, --
• 比较操作符：
• ==, !=, >, >=, <, <=
• 模式匹配符：~ ：左边是否和右边匹配包含 !~ ：是否不匹配
• 逻辑操作符：与&& ，或|| ，非!
• 函数调用： function_name(argu1, argu2, ...)
• 条件表达式（三目表达式）：selector?if-true-expression:if-false-expression
• 注释：先判断selector，如果符合执行 ? 后的操作；否则执行 : 后的操作

4.5.2 演示

（1）模式匹配符

（2）逻辑操作符

（3）条件表达式（三目表达式）

[root@along ~]# awk -F: '{$3 >= 1000?usertype="common user":usertype="sysadmin user";print usertype,$1,$3}' /etc/passwd

sysadmin user root 0

common user along 1000

4.6 awk PATTERN 匹配部分

4.6.1 格式

PATTERN：根据pattern 条件，过滤匹配的行，再做处理

（1）如果未指定：空模式，匹配每一行

（2）/regular expression/ ：仅处理能够模式匹配到的行，正则，需要用/ / 括起来

（3）relational expression：关系表达式，结果为“真”才会被处理

真：结果为非0值，非空字符串

假：结果为空字符串或0值

（4）line ranges：行范围

startline(起始行),endline(结束行)：/pat1/,/pat2/ 不支持直接给出数字，可以有多段，中间可以有间隔

（5）BEGIN/END 模式

BEGIN{}: 仅在开始处理文件中的文本之前执行一次

END{} ：仅在文本处理完成之后执行

4.6.2 演示

4.7 awk有意思的案例

5、awk高阶用法

5.1 awk控制语句—if-else判断

（1）语法

if(condition){statement;…}[else statement]  双分支

if(condition1){statement1}else if(condition2){statement2}else{statement3}  多分支

（2）使用场景：对awk 取得的整行或某个字段做条件判断

（3）演示

[root@along ~]# awk -F: '{if($3>10 && $3<1000)print $1,$3}' /etc/passwd

operator 11

games 1

[root@along ~]# awk -F: '{if($NF=="/bin/bash") print $1,$NF}' /etc/passwd

root /bin/bash

along /bin/bash

---输出总列数大于3的行

[root@along ~]# awk -F: '{if(NF>2) print $0}' awkdemo

linux:redhat:lalala:hahaha

along:love:you

---第3列>=1000为Common user，反之是root or Sysuser

[root@along ~]# awk -F: '{if($3>=1000) {printf "Common user: %s\n",$1} else{printf "root or Sysuser: %s\n",$1}}' /etc/passwd

root or Sysuser: root

root or Sysuser: bin

Common user: along

---磁盘利用率超过40的设备名和利用率

[root@along ~]# df -h|awk -F% '/^\/dev/{print $1}'|awk '$NF > 40{print $1,$NF}'

/dev/mapper/cl-root 43

---test=100和>90为very good; 90>test>60为good; test<60为no pass

[root@along ~]# awk 'BEGIN{ test=100;if(test>90){print "very good"}else if(test>60){ print "good"}else{print "no pass"}}'

very good

[root@along ~]# awk 'BEGIN{ test=80;if(test>90){print "very good"}else if(test>60){ print "good"}else{print "no pass"}}'

good

[root@along ~]# awk 'BEGIN{ test=50;if(test>90){print "very good"}else if(test>60){ print "good"}else{print "no pass"}}'

no pass

5.2 awk控制语句—while循环

（1）语法

注：条件“真”，进入循环；条件“假”， 退出循环

（2）使用场景

对一行内的多个字段逐一类似处理时使用

对数组中的各元素逐一处理时使用

（3）演示

---以along开头的行，以：为分隔，显示每一行的每个单词和其长度

[root@along ~]# awk -F: '/^along/{i=1;while(i<=NF){print $i,length($i); i++}}' awkdemo

along 5

love 4

you 3

---以：为分隔，显示每一行的长度大于6的单词和其长度

[root@along ~]# awk -F: '{i=1;while(i<=NF) {if(length($i)>=6){print $i,length($i)}; i++}}' awkdemo

redhat 6

lalala 6

hahaha 6

---计算1+2+3+...+100=5050

[root@along ~]# awk 'BEGIN{i=1;sum=0;while(i<=100){sum+=i;i++};print sum}'

5050

5.3 awk控制语句—do-while循环

（1）语法

意义：无论真假，至少执行一次循环体

（2）计算1+2+3+...+100=5050

5.4 awk控制语句—for循环

（1）语法

（2）特殊用法：遍历数组中的元素

（3）演示

5.5 和shell脚本中较相似的控制语句

5.5.1 switch语句

和shell中的case很像，就不在演示了

5.5.2 break和continue

5.5.3 next

next：提前结束对本行处理而直接进入下一行处理（awk 自身循环）

5.6 awk数组

5.6.1 关联数组：array[index-expression]

（1）可使用任意字符串；字符串要使用双引号括起来

（2）如果某数组元素事先不存在，在引用时，awk 会自动创建此元素，并将其值初始化为“空串”

（3）若要判断数组中是否存在某元素，要使用“index in array”格式进行遍历

（4）若要遍历数组中的每个元素，要使用for 循环：for(var in array) {for-body}

5.6.2 演示

（1）awk使用数组

分析：把每行作为下标，第一次进来，相当于print ias...一样结果为空，打印空，!取反结果为1，打印本行，并且++变为不空，下次进来相同的行就是相同的下标，本来上次的值，！取反为空，不打印，++变为不空，所以每次重复进来的行都不打印

（2）数组遍历

5.6.3 数值\字符串处理

（1）数值处理

• rand()：返回0和1之间一个随机数，需有个种子 srand()，没有种子，一直输出0.237788

演示：

（2）字符串处理：

• length([s]) ：返回指定字符串的长度
• sub(r,s,[t]) ：对t 字符串进行搜索r 表示的模式匹配的内容，并将第一个匹配的内容替换为s
• gsub(r,s,[t]) ：对t 字符串进行搜索r 表示的模式匹配的内容，并全部替换为s 所表示的内容
• plit(s,array,[r]) ：以r 为分隔符，切割字符串s ，并将切割后的结果保存至array 所表示的数组中，第一个索引值为1, 第二个索引值为2,…

演示：

5.7 awk自定义函数

（1）格式：和bash区别：定义函数（）中需加参数，return返回值不是$?，是相当于echo输出

（2）演示

5.8 awk中调用shell 命令

（1）system 命令

空格是awk 中的字符串连接符，如果system中需要使用awk中的变量可以使用空格分隔，或者说除了awk 的变量外其他一律用"" 引用 起来。

（2）awk 脚本

将awk 程序写成脚本，直接调用或执行

示例：

（3）向awk脚本传递参数

① 格式：

注意 ：在BEGIN 过程 中不可用。直到 首行输入完成以后，变量才可用 。可以通过-v 参数，让awk 在执行BEGIN 之前得到变量的值。命令行中每一个指定的变量都需要一个-v

② 示例

6、grep awk sed对比

grep 主要用于搜索某些字符串；

sed，awk 用于处理文本 ；

grep基本是以行为单位处理文本的； 而awk可以做更细分的处理，通过指定分隔符将一行（一条记录）划分为多个字段，以字段为单位处理文本。awk中支持C语法，可以有分支条件判断、循环语句等，相当于一个小型编程语言。

awk功能比较多是一个编程语言了。 grep功能简单，就是一个简单的正则表达式的匹配。 awk的功能依赖于grep。

grep可以理解为主要作用是在一个文件中查找过滤需要的内容。awk不是过滤查找，而是文本处理工具，是把一个文件处理成你想要的格式。

AWK的功能是什么？与sed和grep很相似，awk是一种样式扫描与处理工具。但其功能却大大强于sed和grep。awk提供了极其强大的功能：它几乎可以完成grep和sed所能完成的全部工作，同时，它还可以可以进行样式装入、流控制、数学运算符、进程控制语句甚至于内置的变量和函数。它具备了一个完整的语言所应具有的几乎所有精美特性。实际上，awk的确拥有自己的语言：awk程序设计语言，awk的三位创建者已将它正式定义为：样式扫描和处理语言。使用awk的第一个理由是基于文本的样式扫描和处理是我们经常做的工作，awk所做的工作有些象数据库，但与数据库不同的是，它处理的是文本文件，这些文件没有专门的存储格式，普通的人们就能编辑、阅读、理解和处理它们。而数据库文件往往具有特殊的存储格式，这使得它们必须用数据库处理程序来处理它们。既然这种类似于数据库的处理工作我们经常会遇到，我们就应当找到处理它们的简便易行的方法，UNIX有很多这方面的工具，例如sed 、grep、sort以及find等等，awk是其中十分优秀的一种。

使用awk的第二个理由是awk是一个简单的工具，当然这是相对于其强大的功能来说的。的确，UNIX有许多优秀的工具，例如UNIX天然的开发工具C语言及其延续C++就非常的优秀。但相对于它们来说，awk完成同样的功能要方便和简捷得多。这首先是因为awk提供了适应多种需要的解决方案：从解决简单问题的awk命令行到复杂而精巧的awk程序设计语言，这样做的好处是，你可以不必用复杂的方法去解决本来很简单的问题。例如，你可以用一个命令行解决简单的问题，而C不行，即使一个再简单的程序，C语言也必须经过编写、编译的全过程。其次，awk本身是解释执行的，这就使得awk程序不必经过编译的过程，同时，这也使得它与shell script程序能够很好的契合。最后，awk本身较C语言简单，虽然awk吸收了C语言很多优秀的成分，熟悉C语言会对学习awk有很大的帮助，但awk本身不须要会使用C语言——一种功能强大但需要大量时间学习才能掌握其技巧的开发工具。

使用awk的第三个理由是awk是一个容易获得的工具。与C和C++语言不同，awk只有一个文件(/bin/awk)，而且几乎每个版本的UNIX都提供各自版本的awk，你完全不必费心去想如何获得awk。但C语言却不是这样，虽然C语言是UNIX天然的开发工具，但这个开发工具却是单独发行的，换言之，你必须为你的UNIX版本的C语言开发工具单独付费（当然使用D版者除外），获得并安装它，然后你才可以使用它。

基于以上理由，再加上awk强大的功能，我们有理由说，如果你要处理与文本样式扫描相关的工作，awk应该是你的第一选择。在这里有一个可遵循的一般原则：如果你用普通的shell工具或shell script有困难的话，试试awk,如果awk仍不能解决问题，则便用C语言，如果C语言仍然失败，则移至C++。

sed是一个非交互性文本流编辑器。它编辑文件或标准输入导出的文本拷贝。sed编辑器按照一次处理 一行的方式来处理文件（或者输入）并把输出送到屏幕上。你可以在vi和ex/ed编辑器里识别他的命令。sed把当前正在处理的行保存在一个临时缓存里，这个缓存叫做模式空间。一但sed完成了对模式空间里的行的处理（即对该行执行sed命令），就把模式空间的行送到屏幕上（除非该命令要删除该行活禁止打印）。处理完该行之后，从模式空间里删除它，然后把下一行读入模式空间，进行处理，并显示。当输入文件的最后一行处理完后，sed终止。通过把每一行存在一个临时缓存里并编辑该行，初始文件不会被修改或被破坏。

本期和大家一起探讨一款完全基于 go 语言实现的分布式消息队列——nsq.

有关于消息队列 Message Queue 的概念及作用，大家可以参考我之前发表的文章——万字长文解析如何基于Redis实现消息队列，这些基础知识本文不再赘述.

nsq 是一款基于 go 语言开发实现的分布式消息队列组件，在 golang 世界中拥有着很高的认可度和流行性，截止今日在 github 上的 stars 数已高达 23 k. 本期会涉及大量对 nsq 的源码走读环节，主要包含：

• nsq 客户端：https://github.com/nsqio/go-nsq . 本文走读源码版本为：v1.1.0.
• nsq 服务端：https://github.com/nsqio/nsq . 本文走读源码版本为：v1.2.1.

此外有一个拓展性内容，涉及到 nsq 服务端与磁盘的存储交互操作，对应项目为 https://github.com/nsqio/go-diskqueue . 大家感兴趣可自行拓展学习，这部分内容本文不作展开.

本期分享内容的大纲罗列如下：

1 架构介绍

本章我们首先来理清楚 nsq 的整体架构和核心概念，在大的方向上建立一定的理解基础，再在后续章节中基于各个主要流程展开技术细节的探讨.

1.1 整体架构

nsq 可以拆分为三个组件：

• nsqd：最核心的消息队列后端模块，负责接收、存储和发送消息
• nsqlookupd：nsq 中的服务发现与注册中心，负责管理 nsqd 节点、topic、channel 之间拓扑映射关系
• nsqadmin：提供了用于实时监控 nsq 集群的 web 可视化界面

1.2 核心概念

nsq 整体架构如上图所示，其中涉及到的 topic、channel、producer、consumer 等概念下面我们逐一拆解：

• topic：人为定义的消息主题，所有被生产出来的消息首先指定其被发往哪个主题.
• channel：消费方定义的消息频道. channel 与 topic 是多对一关系，每个 channel 都会拥有一份 topic 下的全量完整数据
• producer：消息生产方，生产消息时需要显式指定其从属的 topic
• consumer：消息消费方. 消费消息时需要显式指定主题 topic 以及频道 channel

这里我们着重解释一下 channel 的概念. channel 可以与其他消费队列组件中的 consumer group 消费者组的概念类比来看：

• 首先，channel 与 topic 是一对多关系，每当 topic 有新消息到达时，都会拷贝成多份，逐一发送到每个 channel 当中，保证每个 channel 拥有独立完整的一份数据
• 其次，所有 consumer 在发起订阅时，都需要显式指定 topic + channel 的二维信息
• 最后，channel 下的每一条消息会被随机推送给订阅了该 channel 的 1 名 consumer

由此可见，所有 subcribe 了相同 channel 的 consumer 之间自动形成了一种类似消费者组的机制，大家各自消费 topic 下数据的一部分，形成数据分治与负载均衡.

倘若某个 consumer 需要获取到 topic 中的全量数据，那也容易，只需要 subscribe 一个不与他人共享的 channel 即可.

nsq 与其他消息队列组件的另一大差异是，其中的 topic 和 channel 采用的都是懒创建的机制，使用方无需显式执行 topic 或者 channel 的创建操作，channel 由首次针对该频道发起 subscribe 订阅操作的 consumer 创建；而 topic 则由首次针对该主题发起 publish 操作的 producer 或者 subscribe 操作的 consumer 创建.

到这里，有关 nsq 的核心概念就介绍得差不多了，建议大家略作消化后，可以回头看一眼本章最开始展示的整体架构图，巩固一下对概念的理解.

2 使用教程

本章开始，我们介绍一下 nsq 的基本用法.

2.1 启动服务端

首先我们介绍，如何在一台 linux 服务器中启动 nsq 服务端.

nsq 的官网下载安装包地址为 https://nsq.io/deployment/installing.html，可以从中下载到对应于 linux 系统的安装包压缩文件.

在 linux 服务器上解压文件，然后进入 bin 目录即可一键启动 nsq 服务. 为了避免因为启动服务而使得终端陷入阻塞，我们通过 nohup 指令以守护进程的方式分别启动 nsqlookupd、nsqd 和 nsqadmin 进程：

• 启动 nsqlookupd

脚本：start_lookup.sh

nohup ./nsqlookupd ./nsqlookupd &
exit

• 启动 nsqd ：

脚本：start_nsqd.sh

nohup ./nsqd --lookupd-tcp-address=127.0.0.1:4160 &
exit

• 启动 nsqadmin：

脚本：start_admin.sh

nohup ./nsqadmin --lookupd-http-address=127.0.0.1:4161 &
exit

在 nsqadmin 启动后，即可进入 http://localhost:4171 页面浏览 topic、channel 相关信息：

上述各个进程对应的端口号如下：

• nsqlookupd：tcp 端口：4160，用于 nsqd 集群广播；http 端口：4161，用于客户端访问
• nsqd：tcp 端口: 4150，用于生产者、消费者访问；http 端口: 4151，用于生产者访问
• nsqadmin：端口：4171

2.2 运行客户端

下面展示一下，nsq 客户端的示例代码，其中包括对 consumer 和 producer 的运行启动实例. 具体代码以及注释展示如下：

package main


import (
    "fmt"
    "sync"
    "testing"
    "time"


    "github.com/nsqio/go-nsq"
)


const (
    // 用于测试的消息主题
    testTopic = "my_test_topic"
    // nsqd 服务端地址
    nsqdAddr  = "localhost:4150"
    // 订阅 channelGroupA 的消费者 A1
    consumerA1 = "consumer_a1"
    // 订阅 channelGroupA 的消费者 A2
    consumerA2 = "consumer_a2"
    // 单独订阅 channelGroupB 的消费者 B
    consumerB  = "consumer_b"
    // testTopic 下的 channelA
    channelGroupA = "channel_a"
    // testTopic 下的 channelB
    channelGroupB = "channel_b"
)


// 建立 consumer 与 channel 之间的映射关系
// consumerA1、consumerA2 -> channelA
// consumerB -> channelB
var consumerToChannelGroup = map[string]string{
    consumerA1: channelGroupA,
    consumerA2: channelGroupA,
    consumerB:  channelGroupB,
}


func Test_nsq(t *testing.T) {
    // 消费者消费到消息后的执行逻辑
    msgCallback := func(consumerName string, msg []byte) error {
        t.Logf("i am %s, receive msg: %s", consumerName, msg)
        return nil
    }


    // 运行 producer
    if err := runProducer(); err != nil {
        t.Error(err)
        return
    }


    // 并发运行三个 consumer
    var wg sync.WaitGroup
    for consumer := range consumerToChannelGroup {
        // shadow
        wg.Add(1)
        go func(consumer string) {
            defer wg.Done()
            if err := runConsumer(consumer, msgCallback); err != nil {
                t.Error(err)
            }
        }(consumer)




    }
    wg.Wait()
}


// 运行生产者
func runProducer() error {
    // 通过 addr 直连 nsqd 服务端
    producer, err := nsq.NewProducer(nsqdAddr, nsq.NewConfig())
    if err != nil {
        return err
    }
    defer producer.Stop()


    // 通过 producer.Publish 方法，往 testTopic 中发送三条消息
    for i := 0; i < 3; i++ {
        msg := fmt.Sprintf("hello xiaoxu %d", i)
        if err := producer.Publish(testTopic, []byte(msg)); err != nil {
            return err
        }
    }
    return nil 
}


// 用于处理消息的 processor，需要实现 go-nsq 中定义的 msgProcessor interface，核心是实现消息回调处理方法： func HandleMessage(msg *nsq.Message) error
type msgProcessor struct {
    // 消费者名称
    consumerName string
    // 消息回调处理函数
    callback     func(consumerName string, msg []byte) error
}


func newMsgProcessor(consumerName string, callback func(consumerName string, msg []byte) error) *msgProcessor {
    return &msgProcessor{
        consumerName: consumerName,
        callback:     callback,
    }
}


// 消息回调处理
func (m *msgProcessor) HandleMessage(msg *nsq.Message) error {
    // 执行用户定义的业务处理逻辑
    if err := m.callback(m.consumerName, msg.Body); err != nil {
        return err
    }
    // 倘若业务处理成功，则调用 Finish 方法，发送消息的 ack
    msg.Finish()
    return nil
}


// 运行消费者
func runConsumer(consumerName string, callback func(consumerName string, msg []byte) error) error {
    // 根据消费者名获取到对应的 channel
    channel, ok := consumerToChannelGroup[consumerName]
    if !ok {
        return fmt.Errorf("bad name: %s", consumerName)
    }


    // 指定 topic 和 channel，创建 consumer 实例
    consumer, err := nsq.NewConsumer(testTopic, channel, nsq.NewConfig())
    if err != nil {
        return err
    }
    defer consumer.Stop()


    // 添加消息回调处理函数
    consumer.AddHandler(newMsgProcessor(consumerName, callback))
    
    // consumer 连接到 nsqd 服务端，开启消费流程
    if err = consumer.ConnectToNSQD(nsqdAddr); err != nil {
        return err
    }


    <-time.After(5 * time.Second)
}

按照上述流程运行单测代码后，对应的输出结果如下：

nsq_test.go:32: i am consumer_a1, receive msg: hello xiaoxu 0
    nsq_test.go:32: i am consumer_b, receive msg: hello xiaoxu 0
    nsq_test.go:32: i am consumer_a2, receive msg: hello xiaoxu 1
    nsq_test.go:32: i am consumer_b, receive msg: hello xiaoxu 1
    nsq_test.go:32: i am consumer_a1, receive msg: hello xiaoxu 2
    nsq_test.go:32: i am consumer_b, receive msg: hello xiaoxu 2

可以看到，共享了 channelA 的两个 consumerA1、consumerA2 分摊了 topic 下的消息数据；而独享 channelB 的 consumer 则消费到了 topic 下的全量消息.

3 客户端

本章中，我们带着大家深入到 nsq 客户端项目中，梳理生产者与消费者的运行流程.

nsq 客户端 lib 库开源地址为：https://github.com/nsqio/go-nsq，本文走读源码版本为 v1.1.0

3.1 连接交互

在 nsq 客户端部分，无论是生产者 producer 还是消费者 consumer，在与服务端交互时，都是通过客户端定义类 Conn 来封装表示两端之间建立的连接.

3.1.1 类定义

客户端定义的连接类 Conn 定义如下，其中涉及到的核心字段，我都补充了相应的注释.

这里稍微解释下 inFlight 的概念，其指的是某条消息已经被客户端接收到，但是还未给予服务端 ack 的状态.

type Conn struct {
    // 记录了有多少消息处于未 ack 状态
    messagesInFlight int64
    // ...


    // 互斥锁，保证临界资源一致性
    mtx sync.Mutex
    // ...


    // 真正的 tcp 连接
    conn    *net.TCPConn
    // ...
    // 连接的服务端地址
    addr    string
    // ...
    
    // 读入口
    r io.Reader
    // 写出口
    w io.Writer
    // writeLoop goroutine 用于接收用户指令的 channel
    cmdChan         chan *Command
    // write goroutine 通过此 channel 接收来自客户端的响应，发往服务端
    msgResponseChan chan *msgResponse
    // 控制 write goroutine 退出的 channel
    exitChan        chan int
    // 并发等待组，保证所有 goroutine 及时回收，conn 才能关闭
    wg        sync.WaitGroup
    // ...
}

3.1.2 创建连接

由客户端实际向服务端发起一笔连接，对应的方法是 Conn.Connect()，其核心步骤包括：

• 通过 net 包向服务端发起 tcp 连接
• 将 Conn 下的 writer 和 reader 设置为这笔 tcp 连接
• 异步启动 Conn 伴生的 readLoop 和 writeLoop goroutine，持续负责接收和发送与服务端之间的往来数据

func (c *Conn) Connect() (*IdentifyResponse, error) {
    dialer := &net.Dialer{
        LocalAddr: c.config.LocalAddr,
        Timeout:   c.config.DialTimeout,
    }




    conn, err := dialer.Dial("tcp", c.addr)
    // ...
    c.conn = conn.(*net.TCPConn)
    c.r = conn
    c.w = conn
    // var MagicV2 = []byte("  V2")
    _, err = c.Write(MagicV2)
    // ...
    c.wg.Add(2)
    // ...
    // 接收来自服务端的响应
    go c.readLoop()
    // 发送发往服务端的请求
    go c.writeLoop()
    return resp, nil
}

有关 readLoop 和 writeLoop 的设计，其核心优势包括：

• 解耦发送请求与接收响应的串行流程，能够实现更加自由的双工通信
• 充分利用 goroutine 和 channel 的优势，通过 for 自旋 + select 多路复用的方式，保证两个 loop goroutinie 能够在监听到退出指令时熔断流程，比如 context 的 cancel、timeout 事件，比如 exitChan 的关闭事件等

这种读写 loop 模式在很多 go 语言底层通信框架中都有采用，比较经典的案例包括 go 语言中的 net/http 标准库，大家如果感兴趣可以阅读我之前发表的这篇文章——Golang HTTP 标准库实现原理.

有关 Conn 中的 readLoop、writeLoop 部分内容，在本文 3.2、3.3 小节介绍生产者 producer 和 消费者 consumer 模块时，会略作展开.

3.2 生产者

下面我们来介绍一下，在客户端视角下，生产者 producer 向服务端生产消息的 publish 流程.

3.2.1 类定义

生产者 producer 类定义如下，核心字段均已给出相应注释.

值得一提的是，producer 在创建好与服务端交互的连接 Conn 后，也会启动一个常驻的 router goroutine，负责持续接收来自客户端的指令，并复用 Conn 将指令发送到服务端.

// 生产者
type Producer struct {
    // 生产者标识 id
    id     int64
    // 连接的 nsqd 服务器地址
    addr   string
    // 内置的客户端连接，其实现类是 3.1 小节中的 Conn. 次数声明为 producerConn interface，是为 producer 屏蔽了一些生产者无需感知的细节
    conn   producerConn
    // ...
    // 生产者 router goroutine 接收服务端响应的 channel 
    responseChan chan []byte
    // 生产者 router goroutine 接收错误的 channel
    errorChan    chan []byte
    // 生产者 router goroutine 接收生产者关闭信号的 channel
    closeChan    chan int
    // 生产者 router goroutine 接收 publish 指令的 channel
    transactionChan chan *ProducerTransaction
    // ...
    // 生产者的状态
    state           int32
    // 当前有多少 publish 指令并发执行
    concurrentProducers int32
    // 生产者 router goroutine 接收退出信号的 channel
    exitChan            chan int
    // 并发等待组，保证 producer 关闭前及时回收所有 goroutine
    wg                  sync.WaitGroup
    // 互斥锁
    guard               sync.Mutex
}

3.2.2 publish 指令

生产者 producer 生产消息的入口方法为 Producer.Publish()，底层会组装出一个 PUB 指令，并调用 Producer.sendCommand() 方法发送指令.

// Publish synchronously publishes a message body to the specified topic, returning
// an error if publish failed
func (w *Producer) Publish(topic string, body []byte) error {
    return w.sendCommand(Publish(topic, body))
}

PUB 指令：

// Publish creates a new Command to write a message to a given topic
func Publish(topic string, body []byte) *Command {
    var params = [][]byte{[]byte(topic)}
    return &Command{[]byte("PUB"), params, body}
}

3.2.3 发送指令

producer 发送指令的主流程如上图所示，核心步骤包括：

• 首次发送指令时，需要调用 Producer.connect() 方法创建一笔与服务端通信的 Conn
• 通过 Producer.transactionChan 通道，将指令发送到 Producer 的守护协程 router goroutine 当中
• router goroutine 接收到指令后，会调用 Conn.WriteCommand() 方法，将指令发放服务端

func (w *Producer) sendCommand(cmd *Command) error {
    doneChan := make(chan *ProducerTransaction)
    err := w.sendCommandAsync(cmd, doneChan, nil)
    // ...
    t := <-doneChan
    return t.Error
}
func (w *Producer) sendCommandAsync(cmd *Command, doneChan chan *ProducerTransaction, args []interface{}) error {
    // keep track of how many outstanding producers we're dealing with
    // in order to later ensure that we clean them all up...
    atomic.AddInt32(&w.concurrentProducers, 1)
    defer atomic.AddInt32(&w.concurrentProducers, -1)


    if atomic.LoadInt32(&w.state) != StateConnected {
        err := w.connect()
        // ...
    }


    t := &ProducerTransaction{
        cmd:      cmd,
        doneChan: doneChan,
        Args:     args,
    }


    select {
    case w.transactionChan <- t:
    case <-w.exitChan:
        return ErrStopped
    }


    return nil
}

可以看到，在 Producer.connect() 方法中：

• 创建了一个 Conn 实例
• 调用 Conn.Connect() 方法实际向服务端发起连接（这个过程中也会完成 readLoop 和 writeLoop goroutine 的启动）
• 异步启动了 router goroutine，负责持续接收指令，并通过 Conn 与服务端交互

func (w *Producer) connect() error {
    w.guard.Lock()
    defer w.guard.Unlock()


    // ...
    w.conn = NewConn(w.addr, &w.config, &producerConnDelegate{w})
    // ...
    // producer 创建和 nsqd 之间的连接
    _, err := w.conn.Connect()
    // ...
    atomic.StoreInt32(&w.state, StateConnected)
    w.closeChan = make(chan int)
    w.wg.Add(1)
    go w.router()


    return nil
}

router goroutine 的运行框架如下，主要通过 for + select 的形式，持续接收来自 Producer.transactionChan 通道的指令，将其发往服务端.

func (w *Producer) router() {
    for {
        select {
        case t := <-w.transactionChan:
            w.transactions = append(w.transactions, t)
            err := w.conn.WriteCommand(t.cmd)
            // ...
        // ...
        case <-w.closeChan:
            goto exit
        case <-w.exitChan:
            goto exit
        }
    }


// ...
}

3.2.4 接收响应

下面我们来理一下，producer 在发送完 PUB 指令后，又要如何接收到来自服务端的响应. 大家可以结合上面这张流程图，和我们深入到下面的源码走读环节：

步骤 I：在 Conn.readLoop() 方法中，读取到来自服务端的响应，通过调用 Producer.onConnResponse() 方法，将数据发送到 Producer.responseChan 当中

func (c *Conn) readLoop() {
    delegate := &connMessageDelegate{c}
    for {
        // ...


        frameType, data, err := ReadUnpackedResponse(c, c.config.MaxMsgSize)
        // ...


        switch frameType {
        case FrameTypeResponse:
            c.delegate.OnResponse(c, data)
        // ...
        }
    }
    // ...
}
func (d *producerConnDelegate) OnResponse(c *Conn, data []byte)       { d.w.onConnResponse(c, data) }
func (w *Producer) onConnResponse(c *Conn, data []byte) { w.responseChan <- data }

步骤II：Producer 的 router goroutine 通过 Producer.responseChan 接收到响应数据，通过调用 Producer.popTransaction(...) 方法，将响应推送到 doneChan 当中

func (w *Producer) router() {
    for {
        select {
        // ...
        case data := <-w.responseChan:
            w.popTransaction(FrameTypeResponse, data)
        case data := <-w.errorChan:
            w.popTransaction(FrameTypeError, data)
        // ...
    }


    // ...
}
func (w *Producer) popTransaction(frameType int32, data []byte) {
    // ...
    t := w.transactions[0]
    // ...
    t.finish()
}
func (t *ProducerTransaction) finish() {
    if t.doneChan != nil {
        t.doneChan <- t
    }
}

步骤III：客户端通过在 Producer.sendCommand 方法中阻塞等待来自 doneChan 的数据，接收到后将其中的错误返回给上层.

func (w *Producer) sendCommand(cmd *Command) error {
    doneChan := make(chan *ProducerTransaction)
    err := w.sendCommandAsync(cmd, doneChan, nil)
    // ...
    t := <-doneChan
    return t.Error
}

3.3 消费者

接下来梳理一下，消费者 consumer 订阅、消费以及 ack 的流程.

3.3.1 类定义

客户端关于消费者的类定义如下. 每当用户注册消息回调处理函数 handler 时，consumer 就会启动一个 handleLoop goroutine，负责接收消息并调用相应的处理函数.

需要注意，consumer 在被创建出来时，就需要明确指定其订阅的 topic 以及 chanel.

type Consumer struct {
    // ...
    // 互斥锁
    mtx sync.RWMutex


    // ...
    // 消费者标识 id
    id      int64
    // 消费者订阅的 topic
    topic   string
    // 消费者订阅的 channel
    channel string
    // ...
    // 用于接收消息的 channel
    incomingMessages chan *Message


    // ...
    pendingConnections map[string]*Conn
    connections        map[string]*Conn
  
    // 连接的 nsqd 地址
    nsqdTCPAddrs []string


    // ...
}

3.3.2 添加 handler

用户可以通过调用 Consumer.AddHandler(...) 方法，针对每个 handler 会启动一个 handleLoop goroutine 用于在消费到消息时执行预定义的回调函数.

需要注意的是，倘若用户注册了多个 handler，最终每条消息只会随机被其中一个 handler 处理.

func (r *Consumer) AddHandler(handler Handler) {
    r.AddConcurrentHandlers(handler, 1)
}
func (r *Consumer) AddConcurrentHandlers(handler Handler, concurrency int) {
    // ...
    atomic.AddInt32(&r.runningHandlers, int32(concurrency))
    for i := 0; i < concurrency; i++ {
        go r.handlerLoop(handler)
    }
}
func (r *Consumer) handlerLoop(handler Handler) {
    // ...
    for {
        message, ok := <-r.incomingMessages
        // ...
        err := handler.HandleMessage(message)
        // ...
    }


    // ...
}

3.3.3 连接服务端

消费者通过调用 Consumer.ConnectToNSQD(...) 方法，实现与 nsqd 服务端的连接交互，其中核心步骤包括：

• 调用 NewConn(...) 方法，创建 Conn 实例
• 调用 Conn.Connect(...) 方法，实际向服务端发起连接
• 调用 Conn.WriteCommand(...) 方法，向服务端发送 SUB 指令

func (r *Consumer) ConnectToNSQD(addr string) error {
    // 创建链接
    conn := NewConn(addr, &r.config, &consumerConnDelegate{r})
    // ...
    r.mtx.Lock()
    _, pendingOk := r.pendingConnections[addr]
    _, ok := r.connections[addr]
    // ...
    r.pendingConnections[addr] = conn
    if idx := indexOf(addr, r.nsqdTCPAddrs); idx == -1 {
        r.nsqdTCPAddrs = append(r.nsqdTCPAddrs, addr)
    }
    r.mtx.Unlock()


    // ...
    resp, err := conn.Connect()
    // ...


    // ...
    cmd := Subscribe(r.topic, r.channel)
    err = conn.WriteCommand(cmd)
    // ...
    r.mtx.Lock()
    delete(r.pendingConnections, addr)
    r.connections[addr] = conn
    r.mtx.Unlock()
    // ...
    return nil
}

3.3.4 消费消息

当指定 topic channel 下有新消息产生时，consumer 持有的 conn 会通过 readLoop goroutine 接收到对应的消息，并调用 Consumer.onConnMessage(...) 方法，将消息推送到 Consumer.incomingMessages 通道中.

func (c *Conn) readLoop() {
    delegate := &connMessageDelegate{c}
    for {
        // ...


        frameType, data, err := ReadUnpackedResponse(c, c.config.MaxMsgSize)
        // ...


        switch frameType {
        // ...
        case FrameTypeMessage:
            msg, err := DecodeMessage(data)
            // ...           
            c.delegate.OnMessage(c, msg)
        // ...
    }


    // ...
}
func (d *consumerConnDelegate) OnMessage(c *Conn, m *Message)         { d.r.onConnMessage(c, m) }
func (r *Consumer) onConnMessage(c *Conn, msg *Message) {
    // ...
    r.incomingMessages <- msg
}

此前在注册 consumer handler 时启动的 handlerLoop goroutine 便会通过 Consumer.incomingMessages 通道获取到消息，并调用相应的 handler 执行回调处理逻辑.

func (r *Consumer) handlerLoop(handler Handler) {
    // ...
    for {
        message, ok := <-r.incomingMessages
        // ...
        err := handler.HandleMessage(message)
        // ...
    }




    // ...
}

3.3.5 消息 ack

为防止消息丢失，nsq 中同样支持了 consumer ack 机制.

consumer 在接收到消息并成功调用 handler 回调处理函数后，可以通过调用 Message.Finish() 方法，向服务端发送 ack 指令，确认消息已成功接收.

倘若服务端超时未收到 ack 响应，则会默认消息已丢失，会重新推送一轮消息.

consumer ack 流程的核心代码展示如下，在 Conn.onMessageFinish(...) 方法中，会通过 Conn.msgResponseChan 通道将数据推送到 Conn writeLoop goroutine 当中，由其负责将请求发往 nsqd 服务端.

func (m *Message) Finish() {
    .// ..
    m.Delegate.OnFinish(m)
}
func (d *connMessageDelegate) OnFinish(m *Message) { d.c.onMessageFinish(m) }
func (c *Conn) onMessageFinish(m *Message) {
    c.msgResponseChan <- &msgResponse{msg: m, cmd: Finish(m.ID), success: true}
}

Conn.writeLoop 是 Conn 负责向服务端发送指令的常驻 goroutine，其在接收到来自 Consumer.msgResponseChan 通道的数据后，会根据其成功状态，选择调用 OnMessageFinished 或者 OnMessageRequeued 将其封装成 ack 或者重试指令，最后调用 Conn.WriteCommand(...) 方法将指令发往服务端.

func (c *Conn) writeLoop() {
    for {
        select {
        // ...
        // ...  
        case resp := <-c.msgResponseChan:
            // ...
            msgsInFlight := atomic.AddInt64(&c.messagesInFlight, -1)


            if resp.success {
                c.log(LogLevelDebug, "FIN %s", resp.msg.ID)
                c.delegate.OnMessageFinished(c, resp.msg)
                c.delegate.OnResume(c)
            } else {
                c.log(LogLevelDebug, "REQ %s", resp.msg.ID)
                c.delegate.OnMessageRequeued(c, resp.msg)
                if resp.backoff {
                    c.delegate.OnBackoff(c)
                } else {
                    c.delegate.OnContinue(c)
                }
            }


            err := c.WriteCommand(resp.cmd)
            // ...
        }
    }
    // ...
}

4 服务端

接下来进入本文最核心的章节，我们来揭示一下 nsq 服务端的底层实现原理.

nsq 服务端开源地址：https://github.com/nsqio/nsq 本文走读源码的版本为: v1.2.1

4.1 核心类

首先统一梳理一下，nsq 服务端所涉及到的几个核心类.

4.1.1 nsqd

NSQD 类对应于一个 nsqd 节点，其中封装了各类重要信息：

• topicMap：当前 nsqd 节点下包含的 topic 集合
• lookupPeers：倘若启用集群模式，该字段标识出集群中其他节点的信息
• tcpServer：nsqd 节点下的 tcp 服务，用于接收处理来自客户端的各种请求指令

type NSQD struct {
    // 节点内递增的客户端 id 序号，为每个到来的客户端请求分配唯一的标识 id
    clientIDSequence int64
    // 一把读写锁，保证临界资源并发安全
    sync.RWMutex
    
    // ...
    // 存在的 topic 集合
    topicMap map[string]*Topic


    // 集群中的其他 nsqd 节点 
    lookupPeers atomic.Value
    // 运行的 tcp server
    tcpServer     *tcpServer
    // tcp server 使用的端口监听器
    tcpListener   net.Listener
    // http server 使用的端口监听器
    httpListener  net.Listener
    // https server 使用的端口监听器
    httpsListener net.Listener
    // ...


    // 用于回收派生出去的 goroutine  
    exitChan             chan int
    // 并发等待组工具，保证派发出去的 goroutine 都能得到及时回收
    waitGroup            util.WaitGroupWrapper
    // 集群信息
    ci *clusterinfo.ClusterInfo
}

4.1.2 topic

Topic 对应为一个消息主题，其下包含核心字段：

• channelMap：topic 下的 channel 集合
• memoryMsgChan：有新消息到达时，在内存中通过此通道向 topic 传递消息
• backend：当 memoryMsgChan 容量达到上限时，则通过磁盘文件向 topic 传递消息
• nsqd：从属的 nsqd 模块

type Topic struct {
    // 统计消息数量和大小
    messageCount uint64
    messageBytes uint64
    
    // 读写锁，保证临界资源并发安全
    sync.RWMutex


    // 当前 topic 名称
    name              string
    // 当前 topic 下存在的 channel 集合
    channelMap        map[string]*Channel
    // 当 memoryMsgChan 满了，消息通过该组件持久化落盘
    backend           BackendQueue
    // 用于在内存中传递当前 topic 下的消息
    memoryMsgChan     chan *Message
    // 通知 pump goroutine 启动
    startChan         chan int
    // 通知 pump goroutine 退出
    exitChan          chan int
    // 通知 pump goroutine channels 有更新
    channelUpdateChan chan int
    // 等待组工具，保证当前 topic 下派生出的 goroutine 能够被回收
    waitGroup         util.WaitGroupWrapper
    // 消息 id 生成器
    idFactory         *guidFactory


    // ...
    // 从属的 nsqd 模块
    nsqd *NSQD
}

在 topic 被构造出来时，会调用 Topic.messagePump(...) 方法，异步启动一个 goroutine，负责持续接收分配给该 topic 的消息，并将消息逐一发送给该 topic 下的每个 channel

// Topic constructor
func NewTopic(topicName string, nsqd *NSQD, deleteCallback func(*Topic)) *Topic {
    // 构造 topic 实例
    t := &Topic{
        // ...
    }
    // ...


    // 启动 topic 下的 pump goroutine
    t.waitGroup.Wrap(t.messagePump)


    // ...
    // 返回 topic 实例
    return t
}

topic.messagePump 方法非常核心，这里先展示一下其核心源码，该方法在后续 4.3 小节的 publish 流程中还会有所涉及.

func (t *Topic) messagePump() {
    var msg *Message
    var buf []byte
    var err error
    var chans []*Channel
    var memoryMsgChan chan *Message
    var backendChan <-chan []byte


    // ...
    t.RLock()
    for _, c := range t.channelMap {
        chans = append(chans, c)
    }
    t.RUnlock()
    if len(chans) > 0 && !t.IsPaused() {
        // 获取基于内存的 memoryChan 和基于磁盘的 backendChan
        memoryMsgChan = t.memoryMsgChan
        backendChan = t.backend.ReadChan()
    }




    // main message loop
    for {
        select {
        // 通过内存通道接收消息
        case msg = <-memoryMsgChan:
        // 通过磁盘通道接收消息
        case buf = <-backendChan:
            msg, err = decodeMessage(buf)
            // ...
        // 倘若和 lookupd 通信发送 channel 有变更，则需要对 channels list 进行更新
        case <-t.channelUpdateChan:
            chans = chans[:0]
            t.RLock()
            for _, c := range t.channelMap {
                chans = append(chans, c)
            }
            t.RUnlock()
            if len(chans) == 0 || t.IsPaused() {
                memoryMsgChan = nil
                backendChan = nil
            } else {
                memoryMsgChan = t.memoryMsgChan
                backendChan = t.backend.ReadChan()
            }
            continue
        // ...
        }


        // 遍历 topic 下所有 channel，一一将消息投递过去
        for i, channel := range chans {
            chanMsg := msg
            // ...
            // 倘若消息类型为延时消息，则将其添加到延时队列
            if chanMsg.deferred != 0 {
                channel.PutMessageDeferred(chanMsg, chanMsg.deferred)
                continue
            }
            // 将消息逐一投递到 topic 下的每个 channel 中
            err := channel.PutMessage(chanMsg)
            // ...
        }
    }


    // ...
}

4.1.3 channel

Channel 对应为一个消息频道，

• memoryMsgChan：有新消息到达时，在内存中通过此通道向 channel 传递消息
• backend：当 memoryMsgChan 容量达到上限时，则通过磁盘文件向 channel 传递消息
• clients: 订阅该 channel 的 consumer 集合
• deferredPQ：channel 下的延时消息队列，基于小顶堆实现
• inFlightPQ：channel 下的待 ack 确认消息队列，基于小顶堆实现

值得一提的是，channel 下的 memoryMsgChan 和 backend 会同时被所有订阅了该 channel 的 consumer client goroutine 所接收，因此 channel 下的同一条消息只会随机被某个 consumer 消费到.

type Channel struct {
    // 一些计数器
    requeueCount uint64
    messageCount uint64
    timeoutCount uint64


    // 读写锁，保证临界资源并发安全
    sync.RWMutex
    // 从属的 topic 名称
    topicName string
    // 当前 channel 名称
    name      string
    // 从属的 nsqd 模块
    nsqd      *NSQD
    // 当 memoryMsgChan 满了，则通过该组件将消息落盘传递
    backend BackendQueue
    // 用于在内存中传递当前 channel 下的消息
    memoryMsgChan chan *Message
    // ...
    // 记录当前 channel 下的 consumer 集合
    clients        map[int64]Consumer
    // 延时消息集合
    deferredMessages map[MessageID]*pqueue.Item
    // 延时消息队列，底层基于一个小顶堆实现，以执行时间戳作为排序的键
    deferredPQ       pqueue.PriorityQueue
    // 保护延时消息队列的互斥锁
    deferredMutex    sync.Mutex
    // 重试(待确认)消息集合
    inFlightMessages map[MessageID]*Message
    // 重试(待确认)消息队列，底层基于一个小顶堆实现，以执行时间戳作为排序的键
    inFlightPQ       inFlightPqueue
    // 保护重试(待确认)消息队列的互斥锁
    inFlightMutex    sync.Mutex
}

4.1.4 deferredPQ

延时队列 deferredQ 和待 ack 队列 inFlightPQ 底层数据结构类似，都是基于时间戳进行排序的小顶堆. 这里我们以 deferredPQ 为例展开介绍.

在 deferredQ 中使用了 golang container/heap 标准库中的堆结构，其中定义的堆 interface 如下：

type Interface interface {
    sort.Interface
    Push(x any) // add x as element Len()
    Pop() any   // remove and return element Len() - 1.
}
type Interface interface {
    // 返回堆的长度
    Len() int


    // 判断堆中元素大小的规则
    Less(i, j int) bool


    // 交换 index = i 和 index = j 的两个元素
    Swap(i, j int)
}

在 nsq 中定义了延时队列类型 PriorityQueue，逐一实现了上述 interface 的各个方法. 其中，用于比较元素大小的 Item.Priority 对应就是一条消息的执行时间戳.

type Item struct {
    Value    interface{}
    Priority int64
    Index    int
}


// this is a priority queue as implemented by a min heap
// ie. the 0th element is the *lowest* value
type PriorityQueue []*Item




func New(capacity int) PriorityQueue {
    return make(PriorityQueue, 0, capacity)
}




func (pq PriorityQueue) Len() int {
    return len(pq)
}




func (pq PriorityQueue) Less(i, j int) bool {
    return pq[i].Priority < pq[j].Priority
}




func (pq PriorityQueue) Swap(i, j int) {
    pq[i], pq[j] = pq[j], pq[i]
    pq[i].Index = i
    pq[j].Index = j
}




func (pq *PriorityQueue) Push(x interface{}) {
    n := len(*pq)
    c := cap(*pq)
    if n+1 > c {
        npq := make(PriorityQueue, n, c*2)
        copy(npq, *pq)
        *pq = npq
    }
    *pq = (*pq)[0 : n+1]
    item := x.(*Item)
    item.Index = n
    (*pq)[n] = item
}




func (pq *PriorityQueue) Pop() interface{} {
    n := len(*pq)
    c := cap(*pq)
    if n < (c/2) && c > 25 {
        npq := make(PriorityQueue, n, c/2)
        copy(npq, *pq)
        *pq = npq
    }
    item := (*pq)[n-1]
    item.Index = -1
    *pq = (*pq)[0 : n-1]
    return item
}

在 PriorityQueue 中专门实现了一个 PriorityQueue.PeekAndShift(...) 方法，其作用是传入一个指定时间戳 max，倘若堆顶消息时间戳小于等于 max，则 pop 弹出，否则不执行任何操作.

func (pq *PriorityQueue) PeekAndShift(max int64) (*Item, int64) {
    if pq.Len() == 0 {
        return nil, 0
    }


    item := (*pq)[0]
    if item.Priority > max {
        return nil, item.Priority - max
    }
    heap.Remove(pq, 0)




    return item, 0
}

往延时队列中添加消息的方法入口为 Channel.addToDeferredPQ(...)，其中会把一条延时消息封装成一个 pqueue.Item，然后调用 container/heap 包下的 Push 方法，里面会实现新元素入堆的 heapInsert 操作.

func (c *Channel) addToDeferredPQ(item *pqueue.Item) {
    c.deferredMutex.Lock()
    heap.Push(&c.deferredPQ, item)
    c.deferredMutex.Unlock()
}

介绍完几个核心类之后，下面我们开始针对几个核心流程进行梳理总结.

4.2 服务运行

首先我们介绍一下，nsqd 服务端是如何启动运行的.

4.2.1 宏观流程

首先，nsqd 服务端的启动入口位于 apps/nsqd/main.go 文件的 main 方法中：

func main() {
    prg := &program{}
    if err := svc.Run(prg, syscall.SIGINT, syscall.SIGTERM); err != nil {
        logFatal("%s", err)
    }
}

此处 nsq 使用了 go-svc 脚手架：https://github.com/judwhite/go-svc .

在 svc 框架中，会分别调用 program 的 Init() 和 Start() 方法，用于初始化和运行 nsqd 后台程序.

I program.Init

在 program.Init() 方法中，会读取用户输入的配置，然后构造出 nsqd 实例：

func (p *program) Init(env svc.Environment) error {
    // ...
    
    // 初始化 nsqd 实例
    nsqd, err := nsqd.New(opts)
    // ...
    p.nsqd = nsqd


    return nil
}

II program.Start

在 program.Start() 方法中，会异步调用 NSQD.Main() 方法，并分别在其中启动如下几个异步任务：

• 启动 tcp server：面向客户端提供服务，处理各种请求指令
• 启动 http/https server：面向客户端提供生产消息部分服务
• 启动 queueScanLoop：执行延时队列、待 ack 队列的扫描任务
• 启动 lookupLoop：与 nsqlookupd 交互，更新元数据信息

func (p *program) Start() error {
    // program 启动前会加载之前存储的历史数据信息
    err := p.nsqd.LoadMetadata()
    // ...
    err = p.nsqd.PersistMetadata()
    // ...


    go func() {
        // 调用 nsqd.Main 方法，启动 nsqd 常驻 goroutine
        err := p.nsqd.Main()
        // ...
    }()


    return nil
}
func (n *NSQD) Main() error {
    // ...
    // 启动 tcp 服务，提供 publish 和 subscribe 
    n.waitGroup.Wrap(func() {
        exitFunc(protocol.TCPServer(n.tcpListener, n.tcpServer, n.logf))
    })    
    // 启动 http server，提供 publish 流程有关的 api
    if n.httpListener != nil {
        httpServer := newHTTPServer(n, false, n.getOpts().TLSRequired == TLSRequired)
        n.waitGroup.Wrap(func() {
            exitFunc(http_api.Serve(n.httpListener, httpServer, "HTTP", n.logf))
        })
    }
    // 启动 https server
    if n.httpsListener != nil {
        httpsServer := newHTTPServer(n, true, true)
        n.waitGroup.Wrap(func() {
            exitFunc(http_api.Serve(n.httpsListener, httpsServer, "HTTPS", n.logf))
        })
    }


    n.waitGroup.Wrap(n.queueScanLoop)
    // 启动
    n.waitGroup.Wrap(n.lookupLoop)
    // ...


    err := <-exitCh
    return err
}

4.2.2 tcp server

nsqd 的 tcp server 部分采用的是服务端中经典的 for + listen 模式，每当通过 listener.Accept() 方法接收到一笔来自于客户端的连接后，会为这笔连接分配一个 goroutine 处理后续到来的请求.

这部分实现与 go语言 net/http 标准库类似，listener.Accept() 方法向下延伸会使用到 linux 的 epoll 多路复用指令，这部分内容大家如果感兴趣的话，可以阅读一下我之前发表的两篇文章：

• Golang HTTP 标准库实现原理
• 解析 Golang 网络 IO 模型之 EPOLL

回归正题，tcp server 的运行方法展示如下：

func TCPServer(listener net.Listener, handler TCPHandler, logf lg.AppLogFunc) error {
    // ...


    var wg sync.WaitGroup


    for {
        // 接收新的 tcp 请求
        clientConn, err := listener.Accept()
        // ...
        wg.Add(1)
        go func() {
            // 处理请求
            handler.Handle(clientConn)
            wg.Done()
        }()
    }


    // ...
    wg.Wait()
    // ...


    return nil
}

针对于每笔到来的连接，nsqd 服务端会：

• 将 conn 封装成一个 client
• 调用 protocolV2.IOLoop() 方法，为这个连接提供服务

func (p *tcpServer) Handle(conn net.Conn) {
    // ...
    buf := make([]byte, 4)
    _, err := io.ReadFull(conn, buf)
    // ...
    protocolMagic := string(buf)
    // ...
    var prot protocol.Protocol
    switch protocolMagic {
    case "  V2":
        prot = &protocolV2{nsqd: p.nsqd}
    default:
        // ...
        return
    }


    // 基于当前请求的 conn，创建一个新的 client 实例
    client := prot.NewClient(conn)
    p.conns.Store(conn.RemoteAddr(), client)
 
    // 当前请求维度的常驻处理流程
    err = prot.IOLoop(client)
    // ...
    p.conns.Delete(conn.RemoteAddr())
    client.Close()
}

protocolV2.IOLoop() 方法是处理一笔客户端连接的主方法，核心步骤包括：

• 异步运行 protocolV2.messagePump(...) 方法，在其中会持续负责将需要发送的消息推送该客户端
• 读取来自客户端的请求指令，调用 protocolV2.Exec(...) 方法分类处理每一种指令

func (p *protocolV2) IOLoop(c protocol.Client) error {
    var err error
    var line []byte
    var zeroTime time.Time


    client := c.(*clientV2)


    // synchronize the startup of messagePump in order
    // to guarantee that it gets a chance to initialize
    // goroutine local state derived from client attributes
    // and avoid a potential race with IDENTIFY (where a client
    // could have changed or disabled said attributes)
    messagePumpStartedChan := make(chan bool)
    go p.messagePump(client, messagePumpStartedChan)
    <-messagePumpStartedChan


    // 启动
    for {
        // ...
        line, err = client.Reader.ReadSlice('\n')
        // ...
        var response []byte
        response, err = p.Exec(client, params)
        // ...


        if response != nil {
            err = p.Send(client, frameTypeResponse, response)
            // ...
        }
    }


    // ...      
    close(client.ExitChan)
    if client.Channel != nil {
        client.Channel.RemoveClient(client.ID)
    }
    return err
}

protocolV2.Exec(...) 方法中，会根据各类指令，dispatch 到各种处理方法当中. 本文重点介绍的指令包括：

• PUB：生产者 publish 消息的指令，本文 4.3 小节详细展开
• SUB：消费者 subscribe topic + channel 的指令，本文 4.4. 小节详细展开
• FIN：消费者 ack 消息的指令，本文 4.5.3 小节中详细展开

func (p *protocolV2) Exec(client *clientV2, params [][]byte) ([]byte, error) {
    // ...
    switch {
    // ...
    case bytes.Equal(params[0], []byte("PUB")):
        return p.PUB(client, params)
    // ...
    case bytes.Equal(params[0], []byte("SUB")):
        return p.SUB(client, params)
    // ...
    case bytes.Equal(params[0], []byte("FIN")):
        return p.FIN(client, params)    
    }
    return nil, protocol.NewFatalClientErr(nil, "E_INVALID", fmt.Sprintf("invalid command %s", params[0]))
}

4.3 publish 流程

下面我们展开 nsqd 服务端针对 PUB 指令的处理流程. 这部分属于本文中绝对核心的内容.

4.3.1 topic -> channel

在 protocolV2.PUB(...) 方法中：

• 调用 io.ReadFull(...) 方法，读取消息数据
• 调用 NSQD.GetTopic(...) 方法获取到消息所属的 topic（一条被生产出来的消息会明确从属于某一个 topic）
• 调用 topic.PutMessage(...) 方法，将消息推送给对应的 topic. 其本质上会通过 Topic.memoryMsgChan 或 Topic.backend 进行消息传递，让 topic 的常驻运行 goroutine Topic.messagePump(...) 方法接收到这条消息

func (p *protocolV2) PUB(client *clientV2, params [][]byte) ([]byte, error) {
    var err error


    // ...
    topicName := string(params[1])
    // ...


    bodyLen, err := readLen(client.Reader, client.lenSlice)
    // ...
    messageBody := make([]byte, bodyLen)
    _, err = io.ReadFull(client.Reader, messageBody)
    // ...


    topic := p.nsqd.GetTopic(topicName)
    msg := NewMessage(topic.GenerateID(), messageBody)
    err = topic.PutMessage(msg)
    // ...


    client.PublishedMessage(topicName, 1)


    return okBytes, nil
}
// PutMessage writes a Message to the queue
func (t *Topic) PutMessage(m *Message) error {
    t.RLock()
    defer t.RUnlock()
    // ...
    err := t.put(m)
    // ...
    return nil
}

nsqd 服务端处理 sub 指令流程：

func (t *Topic) put(m *Message) error {
    // ...
    if cap(t.memoryMsgChan) > 0 || t.ephemeral || m.deferred != 0 {
        select {
        // 写到内存中
        case t.memoryMsgChan <- m:
            return nil
        // 内存 chan 已满，则写入到磁盘
        default:
            break // write to backend
        }
    }
    err := writeMessageToBackend(m, t.backend)
    // ...
    return nil
}

接下来我们再观察一下 Topic.messagePump() 方法. 其中在读取到新消息后，会遍历当前 topic 下的 channel，逐一调用 Channel.PutMessage(...) 方法，将消息传递到每个 channel 中.

func (t *Topic) messagePump() {
    // ...
    for {
        select {
        case msg = <-memoryMsgChan:
        case buf = <-backendChan:
            msg, err = decodeMessage(buf)
            // ...
        // ...
        }


        for i, channel := range chans {
            chanMsg := msg
            // ...            
            // 延迟消息特殊处理
            if chanMsg.deferred != 0 {
                channel.PutMessageDeferred(chanMsg, chanMsg.deferred)
                continue
            }
            err := channel.PutMessage(chanMsg)
            // ...
        }
    }
// ...
}

在 Channel.PutMessage(...) 方法中，会将消息通过 Channel.memoryMsgChan 或者 Channel.backend，将消息随机传递到某个订阅了该 channel 的 consumer client 对应的 protocolV2.messagePump(...) goroutine 当中.

此处呼应了 4.2.2 小节，该方法会于 4.3.2 小节详细展开.

func (c *Channel) PutMessage(m *Message) error {
    c.exitMutex.RLock()
    defer c.exitMutex.RUnlock()
    // ...
    err := c.put(m)
    // ...
    atomic.AddUint64(&c.messageCount, 1)
    return nil
}
func (c *Channel) put(m *Message) error {
    select {
    case c.memoryMsgChan <- m:
    default:
        err := writeMessageToBackend(m, c.backend)
        c.nsqd.SetHealth(err)
        // ...
    }
    return nil
}

4.3.2 channel -> consumer

当一条消息从 topic 被一一推送到每个 channel 的 memoryMsgChan 和 backendChan 中时，这笔数据会随机被一个订阅了该 channel 的 consumer 消费.

这种随机与互斥的性质得以实现的原因是，是多个 consumer 竞争消费同一个 memoryMsgChan 和 backendChan 中的数据，因此同一条数据一定只能被一个 consumer client 所获取到.

consumer 消费 channel 数据的内容可以参见方法 protocolV2.messagePump()，该方法是与每一个连接 nsqd 服务端的客户端连接一一对应的.

值得一提的是，只有某个 client 执行了订阅 channel 的指令，此时才能从 client.SubEventChan 下接收到该 channel 对应的 subChannel，并进一步获得该 channel 的 memoryMsgChan 和 backendChan. 这部分会与 3.4 小节的内容形成呼应.

func (p *protocolV2) messagePump(client *clientV2, startedChan chan bool) {
    var err error
    var memoryMsgChan chan *Message
    var backendMsgChan <-chan []byte
    var subChannel *Channel
    // ...
    // 当前 client 用于接收 subChannel 的通道
    subEventChan := client.SubEventChan
    // ...  


    for {
       
        // ...
        // 倘若当前 client 执行 sub 指令，次数获取到订阅 channel 下用于传递消息的 memoryMsgChan
        memoryMsgChan = subChannel.memoryMsgChan
        backendMsgChan = subChannel.backend.ReadChan()
        flusherChan = outputBufferTicker.C
        
        // ...
        select {
        // ...      
        // 首先，倘若当前 client 执行的是 sub 指令，则此处会获取到该客户端订阅的 channel
        // 倘若 client 发送的指令是 SUB，则其订阅的 channel 会通过 subEventChan 发送过来
        case subChannel = <-subEventChan:
            // you can't SUB anymore
            subEventChan = nil
        // ...
        // 尝试接收订阅 channel 下的 backendMsgChan
        case b := <-backendMsgChan:
            // ...
            msg, err := decodeMessage(b)
            // ...
            // 发送消息前，先将消息添加到待 ack 确认队列中
            subChannel.StartInFlightTimeout(msg, client.ID, msgTimeout)
            client.SendingMessage()
            // 将消息发送到 client
            err = p.SendMessage(client, msg)
            // ...
        // 尝试接收订阅 channel 下的 memoryMsgChan
        case msg := <-memoryMsgChan:
            // ...
            // 发送消息前，先将消息添加到待 ack 确认队列中
            subChannel.StartInFlightTimeout(msg, client.ID, msgTimeout)
            client.SendingMessage()
            // 将消息发送到 client
            err = p.SendMessage(client, msg)
            // ...
        }
    }


    // ...
}

向客户端发送消息的方法链路展示如下，核心是获取 client 中的 tcp 连接，然后向 io writer 中写入数据发往客户端：

func (p *protocolV2) SendMessage(client *clientV2, msg *Message) error {
    // ...
    buf := bufferPoolGet()
    defer bufferPoolPut(buf)


    _, err := msg.WriteTo(buf)
    // ...


    err = p.Send(client, frameTypeMessage, buf.Bytes())
    // ...
}
func (p *protocolV2) Send(client *clientV2, frameType int32, data []byte) error {
    // ...
    _, err := protocol.SendFramedResponse(client.Writer, frameType, data)
    // ...
}
func SendFramedResponse(w io.Writer, frameType int32, data []byte) (int, error) {
    // ...
    n, err = w.Write(data)
    // ...
}

4.4 subscribe 流程

接下来我们一起梳理下 consumer 向 nsqd 服务端发起 subscribe 指令的流程.

4.4.1 sub 主流程

处理 SUB 指令的入口方法为 protocolV2.SUB(...)，其核心步骤是：

• 在订阅 channel 的 client 集合中添加当前 client
• 将订阅的 channel 发送到当前 client 的 SubEventChan 中，供 protocolV2.messagePump() goroutine 接收和处理

func (p *protocolV2) SUB(client *clientV2, params [][]byte) ([]byte, error) {
    // ...
    topicName := string(params[1])
    // ...


    channelName := string(params[2])
    // ...
    var channel *Channel
    for i := 1; ; i++ {
        topic := p.nsqd.GetTopic(topicName)
        channel = topic.GetChannel(channelName)
        if err := channel.AddClient(client.ID, client); err != nil {
            // ...
        }


        // ...
        break
    }
    atomic.StoreInt32(&client.State, stateSubscribed)
    client.Channel = channel
    // update message pump
    client.SubEventChan <- channel


    return okBytes, nil
}

在 1.2 小节中，我们有提到，channel 是在首次被 consumer 订阅时自动创建的，该流程对应的方法链路如下：

func (t *Topic) GetChannel(channelName string) *Channel {
    t.Lock()
    channel, isNew := t.getOrCreateChannel(channelName)
    t.Unlock()
    // ...
    return channel
}
// this expects the caller to handle locking
func (t *Topic) getOrCreateChannel(channelName string) (*Channel, bool) {
    channel, ok := t.channelMap[channelName]
    if !ok {
        // ...
        channel = NewChannel(t.name, channelName, t.nsqd, deleteCallback)
        t.channelMap[channelName] = channel
        // ...
        return channel, true
    }
    return channel, false
}

4.4.2 pump 协程接收 chan

对应于每个 consumer client 会有一个运行 protocolV2.messagePump(...) 方法的 goroutine，其中会通过 client.SubEventChan 接收到订阅的 channel，然后从对应的 memoryMsgChan 和 backendChan 中接收消息：

func (p *protocolV2) messagePump(client *clientV2, startedChan chan bool) {
    var err error
    var memoryMsgChan chan *Message
    var backendMsgChan <-chan []byte
    var subChannel *Channel
    // ...
    subEventChan := client.SubEventChan
    // ...  


    for {
       
        // ...
        // 倘若当前 client 执行 sub 指令，次数获取到订阅 channel 下用于传递消息的 memoryMsgChan
        memoryMsgChan = subChannel.memoryMsgChan
        backendMsgChan = subChannel.backend.ReadChan()
        flusherChan = outputBufferTicker.C
        
        // ...
        select {        
        case subChannel = <-subEventChan:
            // you can't SUB anymore
            subEventChan = nil
        // ...         
        }
    }
    // ...
}

4.5 inFlight&ack 机制

下面我们捋一下，nsq 为保证消息不丢失而提供的 inFlight 和 ack 机制.

4.5.1 StartInFlight

在 protocolV2.messagePump(...) 方法中，每当从 channel 中接收到消息要发往客户端之前，都会先调用 Channel.StartInFlightTimeout(...) 方法，将消息添加到待 ack 确认队列 inFlightPQ 当中：

func (p *protocolV2) messagePump(client *clientV2, startedChan chan bool) {
    // ...
    for {
        // ...
        select {
        // ...
        case b := <-backendMsgChan:
            // ...
            subChannel.StartInFlightTimeout(msg, client.ID, msgTimeout)
            // ...
            err = p.SendMessage(client, msg)
            // ...
        case msg := <-memoryMsgChan:
            // ...
            subChannel.StartInFlightTimeout(msg, client.ID, msgTimeout)
            // ...
            err = p.SendMessage(client, msg)
            // ...
        }
    }
    // ...
}

添加消息进入 inFlightPQ 前，会根据用户设置的超时重试时间，推算出消息重试的执行时间，以时间戳作为排序的键，将消息添加进入小顶堆中.

func (c *Channel) StartInFlightTimeout(msg *Message, clientID int64, timeout time.Duration) error {
    now := time.Now()
    msg.clientID = clientID
    msg.deliveryTS = now
    msg.pri = now.Add(timeout).UnixNano()
    err := c.pushInFlightMessage(msg)
    if err != nil {
        return err
    }
    c.addToInFlightPQ(msg)
    return nil
}
func (c *Channel) addToInFlightPQ(msg *Message) {
    c.inFlightMutex.Lock()
    c.inFlightPQ.Push(msg)
    c.inFlightMutex.Unlock()
}

4.5.2 queueScan

在 nsqd 节点启动时，会异步启动一个 gorouting 执行 NSQD.queueScanLoop() 方法，该方法的用途就是定时轮询 inFlightPQ 和 deferredPQ，取出其中达到时间条件的消息进行执行：

func (n *NSQD) Main() error {
    // ...
    n.waitGroup.Wrap(n.queueScanLoop)
    // ...
}
func (n *NSQD) queueScanLoop() {
    workCh := make(chan *Channel, n.getOpts().QueueScanSelectionCount)
    // ...
    // 轮询时间间隔
    workTicker := time.NewTicker(n.getOpts().QueueScanInterval)
    // ...
    refreshTicker := time.NewTicker(n.getOpts().QueueScanRefreshInterval)


    channels := n.channels()
    // 在方法中会启动 scan worker goroutine
    n.resizePool(len(channels), workCh, responseCh, closeCh)
    // ...
    for {
        select {
        case <-workTicker.C:
            // ...
        // ...
        }


        num := n.getOpts().QueueScanSelectionCount
        if num > len(channels) {
            num = len(channels)
        }


        // ...
        // 将需要扫描的 channel 通过 workCh 发给 scan worker goroutine
        for _, i := range util.UniqRands(num, len(channels)) {
            workCh <- channels[i]
        }


         // ...
    }   
}
func (n *NSQD) resizePool(num int, workCh chan *Channel, responseCh chan bool, closeCh chan int) {
    // ...
    for {
        // ...
        n.waitGroup.Wrap(func() {
            n.queueScanWorker(workCh, responseCh, closeCh)
        })
        n.poolSize++      
    }
}

queueScanWorker 是 nsqd 节点下异步运行的 goroutine，每当通过 workCh 接收到 channel 时，会调用 Channel.processInFlightQueue(...) 方法，完成扫描任务：

func (n *NSQD) queueScanWorker(workCh chan *Channel, responseCh chan bool, closeCh chan int) {
    for {
        select {
        case c := <-workCh:
            now := time.Now().UnixNano()
            dirty := false
            if c.processInFlightQueue(now) {
                dirty = true
            }
            // ...
            responseCh <- dirty
        // ...
        }
    }
}

在扫描任务中，每次会尝试从 inFlightPQ 堆顶弹出达到执行时间的消息，然后调用 Channel.put(...) 方法将消息传递到某个订阅了该 channel 的 client 手中：

func (c *Channel) processInFlightQueue(t int64) bool {
    c.exitMutex.RLock()
    defer c.exitMutex.RUnlock()


    if c.Exiting() {
        return false
    }


    dirty := false
    for {
        c.inFlightMutex.Lock()
        // 倘若堆顶的消息已经达到执行时间，则弹出对应的消息
        msg, _ := c.inFlightPQ.PeekAndShift(t)
        c.inFlightMutex.Unlock()


        // 倘若堆顶消息未达到执行时间，则结束本次任务
        if msg == nil {
            goto exit
        }
        dirty = true


        _, err := c.popInFlightMessage(msg.clientID, msg.ID)
        if err != nil {
            goto exit
        }
        atomic.AddUint64(&c.timeoutCount, 1)
        c.RLock()
        client, ok := c.clients[msg.clientID]
        c.RUnlock()
        if ok {
            client.TimedOutMessage()
        }
        // 发送重试消息
        c.put(msg)
    }


exit:
    return dirty
}

4.5.3 ack

倘若某条消息已经得到了 consumer 的 ack，则 nsqd 服务端会接收到来自 consumer 发送的 FIN 指令，然后步入到 protocolV2.FIN(...) 方法当中：

func (p *protocolV2) Exec(client *clientV2, params [][]byte) ([]byte, error) {
    // ...
    switch {
    case bytes.Equal(params[0], []byte("FIN")):
        return p.FIN(client, params)
    // ...
    }
    // ...
}

在 protocolV2.FIN(...) 方法中，会获取到消息的唯一 id，然后调用 Channel.FinishMessage(...) 方法，将消息从 channel 的 inFlightPQ 当中移除：

func (p *protocolV2) FIN(client *clientV2, params [][]byte) ([]byte, error) {
    // ...
    id, err := getMessageID(params[1])
    // ...
    err = client.Channel.FinishMessage(client.ID, *id)
    // ...
    client.FinishedMessage()


    return nil, nil
}
// FinishMessage successfully discards an in-flight message
func (c *Channel) FinishMessage(clientID int64, id MessageID) error {
    msg, err := c.popInFlightMessage(clientID, id)
    // ...
    c.removeFromInFlightPQ(msg)
    // ...
}
func (c *Channel) removeFromInFlightPQ(msg *Message) {
    c.inFlightMutex.Lock()
    // ...
    c.inFlightPQ.Remove(msg.index)
    c.inFlightMutex.Unlock()
}

4.6 deferred 机制

下面我们梳理一下 nsq 实现延时消息的流程.

4.6.1 添加延时队列

首先在 publish 流程中，消息由 topic 发放每个 channel 前，会先判断消息是否为延时消息类型. 倘若为延时消息，则会调用 Channel.PutMessageDeferred(...) 方法追加到延时队列当中.

func (t *Topic) messagePump() {
    // ...
    for {
        select {
        case msg = <-memoryMsgChan:
        case buf = <-backendChan:
            msg, err = decodeMessage(buf)
            // ...
        // ...
        }


        for i, channel := range chans {
            chanMsg := msg
            // ...          
            // 延迟消息特殊处理
            if chanMsg.deferred != 0 {
                channel.PutMessageDeferred(chanMsg, chanMsg.deferred)
                continue
            }
            // ...
        }
    }
    // ...
}

添加消息进入延时队列的方法链路如下，其中主要是通过当前时刻结合延时设置，推算出消息的执行时刻，然后以该时间戳为排序键，将消息添加到 deferredPQ 的小顶堆中.

func (c *Channel) PutMessageDeferred(msg *Message, timeout time.Duration) {
    atomic.AddUint64(&c.messageCount, 1)
    c.StartDeferredTimeout(msg, timeout)
}
func (c *Channel) StartDeferredTimeout(msg *Message, timeout time.Duration) error {
    absTs := time.Now().Add(timeout).UnixNano()
    item := &pqueue.Item{Value: msg, Priority: absTs}
    err := c.pushDeferredMessage(item)
    // ...
    c.addToDeferredPQ(item)
    return nil
}
func (c *Channel) pushDeferredMessage(item *pqueue.Item) error {
    c.deferredMutex.Lock()
    // TODO: these map lookups are costly
    id := item.Value.(*Message).ID
    _, ok := c.deferredMessages[id]
    if ok {
        c.deferredMutex.Unlock()
        return errors.New("ID already deferred")
    }
    c.deferredMessages[id] = item
    c.deferredMutex.Unlock()
    return nil
}
func (c *Channel) addToDeferredPQ(item *pqueue.Item) {
    c.deferredMutex.Lock()
    heap.Push(&c.deferredPQ, item)
    c.deferredMutex.Unlock()
}

4.6.2 处理延时任务

扫描执行延时队列的流程与 4.5.2 小节介绍的处理待确认消息的流程类似，都是以 NSQD.queueScanWorker(...) 方法为入口，定时轮询延时队列，找到满足时间条件的消息，调用 Channel.put(...) 方法发送消息.

func (n *NSQD) queueScanWorker(workCh chan *Channel, responseCh chan bool, closeCh chan int) {
    for {
        select {
        case c := <-workCh:
            now := time.Now().UnixNano()
            // ...
            if c.processDeferredQueue(now) {
                dirty = true
            }
        // ...
        }
    }
}

执行

func (c *Channel) processDeferredQueue(t int64) bool {
    c.exitMutex.RLock()
    defer c.exitMutex.RUnlock()


    // ...
    dirty := false
    for {
        c.deferredMutex.Lock()
        // 倘若堆顶的消息已经达到执行时间条件，则取出进行执行
        item, _ := c.deferredPQ.PeekAndShift(t)
        c.deferredMutex.Unlock()
        // 倘若堆顶的消息都尚未达到执行时间条件，则结束本次任务
        if item == nil {
            goto exit
        }
        dirty = true




        msg := item.Value.(*Message)
        _, err := c.popDeferredMessage(msg.ID)
        if err != nil {
            goto exit
        }
        // 发送延时消息，由  protocolV2.messagePump(..) 方法接收处理
        c.put(msg)
    }


exit:
    return dirty
}

5 总结

本期向大家详细分享了一款完全基于 go 语言实现的分布式消息队列——nsq. 文中分别向大家详细介绍了有关 nsq 的核心概念、使用教程，并且和大家一起深入到 nsq 项目源码当中，逐一向大家揭示了 nsq 客户端与服务端项目的底层技术细节.

这里比较自信地说一句，只要大家认真阅读完本文，对于 nsq 的理解程度一定能做到“遥遥领先”！

在使用了像 NGINX 等专业的负载均衡器后，我自己也尝试着用 Golang 创建了一个简易负载均衡器。Go 是一种现代语言，第一特性是支持并发。Go 有丰富的标准库，使用这些库你可以用更少的代码写出高性能的应用程序。对每一个发行版本它都有静态链接库。

我们的简易负载均衡器工作原理

负载均衡器有不同的策略用来在一组后端机器中分摊负载。

例如：

• 轮询 平等分摊，认为后端的所有机器处理能力相同
• 加权轮询 基于后端机器不同的处理能力，为其加上不同的权重
• 最少连接数 负载被分流到活跃连接最少的服务器

至于我们的简易负载均衡器，我们会实现这里边最简单的方式 轮询。

轮询选择

轮询无疑是很简单的。它轮流给每个 worker 相同的执行任务的机会。

上图已经说明了，负载均衡器周期性地选择某台服务器。但是我们不能直接使用它，不是吗？

如果后端机器宕机了怎么办？恐怕我们不会希望流量被路由到挂掉的机器上去。因此除非我们添加一些条件，否则不能直接使用这个算法。我们需要把流量只路由到没有挂掉且正常运行的后端机器上。

定义几个结构体

修正思路后，现在我们清楚我们是想要一种能跟踪后端机器状态信息的方法。我们需要检查机器是否存活，也需要跟踪 Url。

我们可以简单地定义一个下面的结构体来维护我们的后端机器。

type Backend struct {
    URL          *url.URL
    Alive        bool
    mux          sync.RWMutex
    ReverseProxy *httputil.ReverseProxy
}

不要担心，后面我会解释 Backend 里的字段。

现在我们要在负载均衡器中跟踪所有后端机器的状态，可以简单地使用一个切片来实现。另外还需要一个计算变量。我们可以定义为 ServerPool

type ServerPool struct {
    backends []*Backend
    current  uint64
}

ReverseProxy 的使用

前面已经声明过了，负载均衡器是专门用来把流量路由到不同的后端机器以及把结果返回给来源客户端的。

Go 官方文档的描述：

ReverseProxy 是一种 HTTP Handler，接收请求并发送到另一台服务器，把响应代理回客户端。

**而这正是我们需要的。**我们不需要重复造轮子了。我们可以简单地通过 ReverseProxy 转发原始请求。

u, _ := url.Parse("http://localhost:8080")
rp := httputil.NewSingleHostReverseProxy(u)

// initialize your server and add this as handler
http.HandlerFunc(rp.ServeHTTP)

通过 httputil.NewSingleHostReverseProxy(url) 我们可以初始化一个把请求转发给 url 的反向代理。在上面的例子中，所有的请求会被转发到 localhost:8080，结果会发回到来源客户端。这里你可以找到更多例子。

如果我们看一下 ServeHTTP 方法的签名，它有 HTTP handler 的签名，因此我们可以把它传给 http 的 HandlerFunc。

你可以在文档中找到更多例子。

在我们的简易负载均衡器中，我们可以用与 ReverseProxy 相关联的 Backend 中的 URL 初始化 ReverseProxy，这样 ReverseProxy 就会把我们请求路由到 URL.

选择处理过程

我们要在下一次轮询中跳过挂掉的后端机器。但是无论如何我们需要一种计数的方式。

很多客户端会连接到负载均衡器，当某一个客户端发来请求时，我们要转发流量的目标机器会出现竞争。我们可以使用 mutex 为 ServerPool 加锁来避免这种现象。但这是一种过犹不及的手段，毕竟我们不希望锁住 ServerPool。我们的需求只是让计数器加 1.

为了满足这个需求，最理想的解决方案是让加 1 成为原子操作。Go 的 atomic 包能完美支持。

func (s *ServerPool) NextIndex() int {
    return int(atomic.AddUint64(&s.current, uint64(1)) % uint64(len(s.backends)))
}

这里我们的加 1 是原子操作，通过对切片的长度取模返回了 index。这意味着返回的值一定在 0 与切片长度之间。归根结底，我们需要的是一个特定的 index，而不是所有数。

选中存活的后端机器

我们已经知道我们的请求是被周期性的路由到每台后端机器上的。我们要做的就是跳过挂掉的机器。

GetNext() 返回的一定是 0 与 切片长度之间的值。每次我们要转发请求到后端某台机器时，如果它挂掉了，我们必须循环地查找整个切片。

从上图可以看出，我们要想从 next 开始遍历整个 list，只需要遍历 next + length。但是我们要覆盖整个切片的长度才能选中一个 index。我们可以用取模操作很容易地实现。

当我们通过搜索找到了一台正常工作的后端机器时，我们把它标记为 current。

下面是对应上面操作的代码。

// GetNextPeer returns next active peer to take a connection
func (s *ServerPool) GetNextPeer() *Backend {
    // loop entire backends to find out an Alive backend
    next := s.NextIndex()
    l := len(s.backends) + next // start from next and move a full cycle
    for i := next; i < l; i++ {
        idx := i % len(s.backends) // take an index by modding with length
        // if we have an alive backend, use it and store if its not the original one
        if s.backends[idx].IsAlive() {
            if i != next {
                atomic.StoreUint64(&s.current, uint64(idx)) // mark the current one
            }
            return s.backends[idx]
        }
    }
    return nil
}

在 Backend 结构体中避免竞争条件

我们需要考虑一个很严重的问题。我们的 Backend 结构体有一个可能被不同协程同时修改或访问的变量。

我们知道读的协程数比写的多。因此我们用 RWMutex 来串行化对 Alive 的读写。

// SetAlive for this backend
func (b *Backend) SetAlive(alive bool) {
    b.mux.Lock()
    b.Alive = alive
    b.mux.Unlock()
}

// IsAlive returns true when backend is alive
func (b *Backend) IsAlive() (alive bool) {
    b.mux.RLock()
    alive = b.Alive
    b.mux.RUnlock()
    return
}

让负载均衡器发请求

所有的准备工作都做完了，我们可以用下面的方法对我们的请求实现负载均衡。只有在所有的后端机器都离线后它才会返回失败。

// lb load balances the incoming request
func lb(w http.ResponseWriter, r *http.Request) {
    peer := serverPool.GetNextPeer()
    if peer != nil {
        peer.ReverseProxy.ServeHTTP(w, r)
        return
    }
    http.Error(w, "Service not available", http.StatusServiceUnavailable)
}

这个方法可以简单地作为一个 HandleFunc 传给 http server。

server := http.Server{
    Addr:    fmt.Sprintf(":%d", port),
    Handler: http.HandlerFunc(lb),
}

仅把流量路由到健康的后端机器

现在我们的 lb 有个严重的问题。我们不知道后端某台机器是否健康。我们必须向后端机器发送请求再检查它是否存活才能知道。

我们可以用两种方法实现：

• **主动：**在处理当前的请求时，选中的某台机器没有响应，我们把它标记为挂掉。
• **被动：**我们可以以固定的周期 ping 后端机器，检查其状态

主动检查后端健康机器

ReverseProxy 在有错误时会触发一个回调函数 ErrorHandler。我们可以用它来检测失败。下面是其实现：

proxy.ErrorHandler = func(writer http.ResponseWriter, request *http.Request, e error) {
    log.Printf("[%s] %s\n", serverUrl.Host, e.Error())
    retries := GetRetryFromContext(request)
    if retries < 3 {
        select {
            case <-time.After(10 * time.Millisecond):
                ctx := context.WithValue(request.Context(), Retry, retries+1)
                proxy.ServeHTTP(writer, request.WithContext(ctx))
            }
            return
        }

    // after 3 retries, mark this backend as down
    serverPool.MarkBackendStatus(serverUrl, false)

    // if the same request routing for few attempts with different backends, increase the count
    attempts := GetAttemptsFromContext(request)
    log.Printf("%s(%s) Attempting retry %d\n", request.RemoteAddr, request.URL.Path, attempts)
    ctx := context.WithValue(request.Context(), Attempts, attempts+1)
    lb(writer, request.WithContext(ctx))
}

这里我们利用闭包的特性来设计这个错误 handler。我们能把 serverUrl 等外部的变量捕获到方法内。它会检查已存在的重试次数，如果小于 3，就会再发送同样的请求到同一台机器。这么做的原因是可能会有临时的错误，服务器可能暂时拒绝你的请求而在短暂的延迟之后又变得可用了（可能服务器为了接受更多的客户端耗尽了 socket）。因此我们加了一个定时器，延迟 10 毫秒左右进行重试。每次请求都会加一次重试次数的计数。

3 次请求都失败后，我们把这台后端机器标记为挂掉。

下一步要做的是，把这个请求发送到另外一台后端机器。我们通过使用 context 包来维护一个尝试次数的计数来实现。增加了尝试次数的计数后，我们把它返回给 lb，寻找下一台可用的后端机器来处理这个请求。

我们不能无限地重复这个过程，因此我们需要在 lb 中检查在继续处理该请求之前是否已达到了最大尝试次数。

我们可以简单地从请求中取得尝试次数，如果它超过了最大数，取消这次请求。

// lb load balances the incoming request
func lb(w http.ResponseWriter, r *http.Request) {
    attempts := GetAttemptsFromContext(r)
    if attempts > 3 {
        log.Printf("%s(%s) Max attempts reached, terminating\n", r.RemoteAddr, r.URL.Path)
        http.Error(w, "Service not available", http.StatusServiceUnavailable)
        return
    }

    peer := serverPool.GetNextPeer()
    if peer != nil {
        peer.ReverseProxy.ServeHTTP(w, r)
        return
    }
    http.Error(w, "Service not available", http.StatusServiceUnavailable)
}

这个实现是递归的。

context 的使用

context 包能让你保存 HTTP 请求中有用的数据。我们大量地使用了它来跟踪请求中特定的数据，如尝试次数和重试次数。

首先，我们需要指定 context 的 key。推荐用不重复的整型而不是字符串作为 key。Go 提供了 iota 关键字实现常量的增加，每个 iota 含有一个独一无二的值。这是一个定义整型 key 的完美解决方案。

const (
    Attempts int = iota
    Retry
)

然后我们可以像在 HashMap 中检索值一样检索定义的值。返回的默认值随实际用例的不同而不同。

// GetAttemptsFromContext returns the attempts for request
func GetRetryFromContext(r *http.Request) int {
    if retry, ok := r.Context().Value(Retry).(int); ok {
        return retry
    }
    return 0
}

被动健康检查

通过被动健康检查我们可以恢复挂掉的后端机器或识别它们。我们以固定的时间周期 ping 后端机器来检查它们的状态。

我们尝试建立 TCP 连接来 ping 机器。如果后端机器有响应，我们把它比较为存活的。如果你愿意，这种方法可以修改为请求一个类似 /status 的特定的服务终端。建立连接之后不要忘记关闭连接，以免对服务器造成额外的负载。否则，它会尝试一直维持连接最终耗尽资源。

// isAlive checks whether a backend is Alive by establishing a TCP connection
func isBackendAlive(u *url.URL) bool {
    timeout := 2 * time.Second
    conn, err := net.DialTimeout("tcp", u.Host, timeout)
    if err != nil {
        log.Println("Site unreachable, error: ", err)
        return false
    }
    _ = conn.Close() // close it, we dont need to maintain this connection
    return true
}

现在我们可以像下面这样遍历服务器并标记它们的状态。

// HealthCheck pings the backends and update the status
func (s *ServerPool) HealthCheck() {
    for _, b := range s.backends {
        status := "up"
        alive := isBackendAlive(b.URL)
        b.SetAlive(alive)
        if !alive {
            status = "down"
        }
        log.Printf("%s [%s]\n", b.URL, status)
    }
}

在 Go 中，我们可以起一个定时器来周期性地运行它。当定时器创建后，你可以用通道监听事件。

// healthCheck runs a routine for check status of the backends every 20 secs
func healthCheck() {
    t := time.NewTicker(time.Second * 20)
    for {
        select {
        case <-t.C:
            log.Println("Starting health check...")
            serverPool.HealthCheck()
            log.Println("Health check completed")
        }
    }
}

在上一段中，<-t.C 通道会每 20 秒接收一次数据。select 探测这个事件。如果没有 default 分支，select 会一直等待，直到至少一个分支执行。

最后，在一个单独的协程中运行。

总结

本文讲了很多

• 轮询选择
• 标准库中的 ReverseProxy
• Mutex
• 原子操作
• 闭包
• 回调
• select 操作

还有很多可以做的来改进我们的简易负载均衡器。

例如：

• 用堆来对后端机器进行排序，减少搜索范围
• 采集统计信息
• 实现加权轮询/最少连接
• 支持配置文件

等等。

你可以在这里找到代码库。

由于 Go 协程相对于传统操作系统中的线程 (thread) 是非常轻量级的，因此对于一个典型的 Go 应用来说，有数以千计的 Go 协程并发运行的情形是十分常见的。并发可以显著地提升应用的运行速度，并且可以帮助我们编写关注点分离（Separation of concerns，Soc）的代码。

什么是 Go 协程？

我们也许在理论上已经知晓 Go 协程是如何工作的，但是在代码层级上，go 协程何许物也？其实，go 协程看起来只是一个与其他众 Go 协程并发运行的一个简单函数或者方法，但是我们并不能想当然地从函数或者方法中的定义来确定一个 Go 协程，go 协程的确定还是要取决于我们如何去调用。

Go 中提供了一个关键字 go 来让我们创建一个 Go 协程，当我们在函数或方法的调用之前添加一个关键字 go，这样我们就开启了一个 Go 协程，该函数或者方法就会在这个 Go 协程中运行。

举个简单的栗子：

https://play.golang.org/p/pIGsToIA2hL

在上面的代码中，我们定义了一个可以在控制台输出 Hello World 字符串的 printHello 的函数，在 main 函数中，我们就像平时那样调用 printHello 函数，最终也是理所当然地获得了期望的结果。

下面，让我们尝试从同一个函数创建 Go 协程：

https://play.golang.org/p/LWXAgDpTcJP

根据 Go 协程的语法，我们在函数调用的前面增加了一个 go 关键字，之后程序运行正常，输出了以下的结果：

main execution started
main execution stopped

奇怪的是，Hello World 并没有如同我们预料的那样输出，这期间究竟发生了什么？

go 协程总是在后台运行，当一个 Go 协程执行的时候（在这个例子中是 go printHello()）, Go 并不会像在之前的那个例子中在执行 printHello 中的功能时阻塞 main 函数中剩下语句的执行，而是直接忽略了 Go 协程的返回并继续执行 main 函数剩下的语句。即便如此，我们为什么没法看到函数的输出呢？

在默认情况下，每个独立的 Go 应用运行时就创建了一个 Go 协程，其 main 函数就在这个 Go 协程中运行，这个 Go 协程就被称为 go 主协程（main Goroutine，下面简称主协程）。在上面的例子中，主协程 中又产生了一个 printHello 这个函数的 Go 协程，我们暂且叫它 printHello 协程 吧，因而我们在执行上面的程序的时候，就会存在两个 Go 协程（main 和 printHello）同时运行。正如同以前的程序那样，go 协程们会进行协同调度。因此，当 主协程 运行的时候，Go 调度器在 主协程 执行完之前并不会将控制权移交给 printHello 协程。不幸的是，一旦 主协程 执行完毕，整个程序会立即终止，调度器再也没有时间留给 printHello 协程 去运行了。

但正如我们从其他课程所知，通过阻塞条件，我们可以手动将控制权转移给其他的 Go 协程 , 也可以说是告诉调度器让它去调度其他可用空闲的 Go 协程。让我们调用 time.Sleep() 函数去实现它吧。

https://play.golang.org/p/ujQKjpALlRJ

如上图所示，我们修改了程序，程序在 main 函数的最后一条语句之前调用了 time.Sleep(10 * time.Millisecond)，使得 主协程 在执行最后一条指令之前调度器就将控制权转移给了 printhello 协程。在这个例子中，我们通过调用 time.Sleep(10 * time.Millisecond) 强行让 主协程 休眠 10ms 并且在在这个 10ms 内不会再被调度器重新调度运行。

一旦 printHello 协程 执行，它就会向控制台打印‘ Hello World ！’，然后该 Go 协程（printHello 协程）就会随之终止，接下来 主协程 就会被重新调度（因为 main Go 协程已经睡够 10ms 了），并执行最后一条语句。因此，运行上面的程序就会得到以下的输出 :

main execution started
Hello World!
main execution stopped

下面我稍微修改一下例子，我在 printHello 函数的输出语句之前添加了一条 time.Sleep(time.Millisecond)。我们已经知道了如果我们在函数中调用了休眠（sleep）函数，这个函数就会告诉 Go 调度器去调度其他可被调度的 Go 协程。在上一课中提到，只有非休眠（non-sleeping）的 Go 协程才会被认为是可被调度的，所以主协程在这休眠的 10ms 内是不会被再次调度的。因此 主协程 先打印出“ main execution started ” 接着就创建了一个 printHello 协程，需要注意此时的 主协程 还是非休眠状态的，在这之后主协程就要调用休眠函数去睡 10ms，并且把这个控制权让出来给printHello 协程。printHello 协程会先休眠 1ms 告诉调度器看看有没有其他可调度的 Go 协程，在这个例子里显然没有其他可调度的 Go 协程了，所以在printHello协程结束了这 1ms 的休眠户就会被调度器调度，接着就输出了“ Hello World ”字符串，之后这个 Go 协程运行结束。之后，主协程会在之后的几毫秒被唤醒，紧接着就会输出“ main execution stopped ”并且结束整个程序。

https://play.golang.org/p/rWvzS8UeqD6

上面的程序依旧和之前的例子一样，输出以下相同的结果：

main execution started
Hello World!
main execution stopped

要是，我把这个printHello 协程中的休眠 1 毫秒改成休眠 15 毫秒，这个结果又是如何呢？

https://play.golang.org/p/Pc2nP2BtRiP

在这个例子中，与其他的例子最大的区别就是printHello 协程比主协程的休眠时间还要长，很明显，主协程要比 printHello 协程唤醒要早，这样的结果就是主协程即使唤醒后执行完所有的语句，printHello 协程还是在休眠状态。之前提到过，主协程比较特殊，如果主协程执行结束后整个程序就要退出，所以 printHello 协程得不到机会去执行下面的输出的语句了，所以以上的程序的数据结果如下：

main execution started
main execution stopped

使用多 Go 协程

就像之前我所提到过的，你可以随心所欲地创建多个 Go 协程。下面让我们定义两个简单的函数，一个是用于顺序打印某个字符串中的每个字符，另一个是顺序打印出某个整数切片中的每个数字。

https://play.golang.org/p/SJano_g1wTV

在上图中的程序中，我们连续地创建了两个 Go 协程，程序输出的结果如下：

main execution started
H e l l o 1 2 3 4 5
main execution stopped

上面的结果证实了 Go 协程是以合作式调度来运作的。下面我们在每个函数中的输出语句的下面添加一行 time.Sleep，让函数在输出每个字符或数字后休息一段时间，好让调度器调度其他可用的 Go 协程。

https://play.golang.org/p/lrSIEdNxSaH

在上面的程序中，我又修改了一下输出语句使得我们可以看到每个字符或数字的输出时刻。理论上主协程会休眠 200ms，因此其他 Go 协程要赶在主协程唤醒之前做完自己的工作，因为主协程唤醒之后就会导致程序退出。getChars 协程每打印一个字符就会休眠 10ms，之后控制权就会传给 getDigits 协程，getDigits 协程每打印一个数字后就休眠 30ms，若 getChars 协程唤醒，则会把控制权传回 getChars 协程，如此往复。在代码中可以看到，getChars 协程会在其他协程休眠的时候多次进行打印字符以及休眠操作，所以我们预计可以看到输出的字符比数字更具有连续性。

我们在 Windows 上运行上面的程序，得到了以下的结果：

main execution started at time 0s
H at time 1.0012ms                         <-|
1 at time 1.0012ms                           | almost at the same time
e at time 11.0283ms                        <-|
l at time 21.0289ms                          | ~10ms apart
l at time 31.0416ms
2 at time 31.0416ms
o at time 42.0336ms
3 at time 61.0461ms                        <-|
4 at time 91.0647ms                          |
5 at time 121.0888ms                         | ~30ms apart
main execution stopped at time 200.3137ms    | exiting after 200ms

通过以上输出结果可以证明我们之前对输出的讨论。对于这个结果，我们可以通过下面的的程序运行图来解释。需要注意的是，我们在图中定义一个输出语句大约会花费 1ms 的 CPU 时间，而这个时间相对于 200ms 来说是可以忽略不计的。

现在我们已经知道了如何去创建 Go 协程以及去如何去使用它。但是使用 time.Sleep 只是一个让我们获取理想结果的一个小技巧。在实际生产环境中，我们无法知晓一个 Go 协程到底需要执行多长的时间，因而在 main 函数里面添加一个 time.Sleep 并不是一个解决问题的方法。我们希望 Go 协程在执行完毕后告知主协程运行的结果。在目前阶段，我们还不知道如何向其他 Go 协程传递以及获取数据，简而言之，就是与其他 Go 协程进行通信。这就是 channels 引入的原因。我们会在下一次课中讨论这个东西。

匿名 Go 协程

如果一个匿名函数可以退出，那么匿名 Go 协程也同样可以退出。请参照functions 课程中的 即时调用函数（Immedietly invoked function） 来理解本节。让我们修改一下之前 printHello 协程的例子：

结果非常明显，因为我们定义了匿名函数，并在同一语句中作为 Go 协程执行。

需要注意的是，所有的 Go 协程都是匿名的，因为我们从并发（concurrency 一课中学到，go 协程是不存在标识符的，在这里所谓的匿名 Go 协程只是通过匿名函数来创建的 Go 协程罢了

我们可以使用已经存在的试验性的 Go 包 x/time/rate。

在本教程中，我们将创建一个基于用户 IP 地址进行速率限制的简单的中间件。

「干净的」HTTP 服务

让我们从构建一个简单的 HTTP 服务开始，该服务具有非常简单的 endpiont。这可能是个非常「重」的 endpoint，因此我们想在这里添加速率限制。

package main

import (
    "log"
    "net/http"
)

func main() {
    mux := http.NewServeMux()
    mux.HandleFunc("/", okHandler)

    if err := http.ListenAndServe(":8888", mux); err != nil {

    // log.Fatalf("unable to start server: %s", err.Error())
    }
}

func okHandler(w http.ResponseWriter, r *http.Request) {
    // Some very expensive database call
    w.Write([]byte("alles gut"))
}

在 main.go 中，我们启动服务，该服务监听 :8888 并拥有一个 endpoint /。

http://golang.org/x/time/rate

我们将使用 Go 中 x/time/rate 包，该包提供了令牌桶限速算法。rate#Limiter 控制事件发生的频率。它实现了一个大小为 b 的「令牌桶」，初始化时是满的，并且以每秒 r 个令牌的速率重新填充。非正式地，在任意足够长的时间间隔中，限速器将速率限制在每秒 r 个令牌，最大突发事件为 b 个。

既然我们想要实现基于 IP 地址的限速器，我们还需要维护一个限速器的 map。

package main

import (
    "sync"

    "golang.org/x/time/rate"
)

// IPRateLimiter .
type IPRateLimiter struct {
    ips map[string]*rate.Limiter
    mu  *sync.RWMutex
    r   rate.Limit
    b   int
}

// NewIPRateLimiter .
func NewIPRateLimiter(r rate.Limit, b int) *IPRateLimiter {
    i := &IPRateLimiter{
        ips: make(map[string]*rate.Limiter),
        mu:  &sync.RWMutex{},
        r:   r,
        b:   b,
    }

    return i
}

// AddIP creates a new rate limiter and adds it to the ips map,
// using the IP address as the key
func (i *IPRateLimiter) AddIP(ip string) *rate.Limiter {
    i.mu.Lock()
    defer i.mu.Unlock()

    limiter := rate.NewLimiter(i.r, i.b)

    i.ips[ip] = limiter

    return limiter
}

// GetLimiter returns the rate limiter for the provided IP address if it exists.
// Otherwise calls AddIP to add IP address to the map
func (i *IPRateLimiter) GetLimiter(ip string) *rate.Limiter {
    i.mu.Lock()
    limiter, exists := i.ips[ip]

    if !exists {
        i.mu.Unlock()
        return i.AddIP(ip)
    }

    i.mu.Unlock()

    return limiter
}

NewIPRateLimiter 创建了一个 IP 限速器的实例，HTTP 服务需要调用该实例的 GetLimiter 方法去获取特定 IP 的限速器（从 map 中获取，或者生成一个新的）。

中间件

让我们升级我们的 HTTP 服务，将中间件添加到所有的 endpoints 中，因此，如果某 IP 达到了限制速率，服务将会返回 429 Too Many Requests，否则服务将处理该请求。

在 limitMiddleware 函数中，每次中间件收到 HTTP 请求，我们都会调用全局限速器的 Allow() 方法。如果令牌桶中没有剩余的令牌，Allow() 将返回 false，我们返回给用户 429 Too Many Requests 响应。否则，调用 Allow() 将消耗桶中的一个令牌，我们将控制权传递给调用链的下一个处理器。

package main

import (
    "log"
    "net/http"
)

var limiter = NewIPRateLimiter(1, 5)

func main() {
    mux := http.NewServeMux()
    mux.HandleFunc("/", okHandler)

    if err := http.ListenAndServe(":8888", limitMiddleware(mux)); err != nil {
        log.Fatalf("unable to start server: %s", err.Error())
    }
}

func limitMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        limiter := limiter.GetLimiter(r.RemoteAddr)
        if !limiter.Allow() {
            http.Error(w, http.StatusText(http.StatusTooManyRequests), http.StatusTooManyRequests)
            return
        }

        next.ServeHTTP(w, r)
    })
}

func okHandler(w http.ResponseWriter, r *http.Request) {
    // Some very expensive database call
    w.Write([]byte("alles gut"))
}

构建 & 运行

go get golang.org/x/time/rate
go build -o server .
./server

测试

有一个非常棒的工具称作 vegeta，我喜欢在 HTTP 负载测试中使用（它也是用 Go 编写的）

brew install vegeta

我们需要创建一个简单的配置文件，声明我们想要发送的请求。

GET http://localhost:8888/

然后，以每个时间单元 100 个请求的速率攻击 10 秒。

vegeta attack -duration=10s -rate=100 -targets=vegeta.conf | vegeta report

结果，你将看到一些请求返回 200，但大多数返回 429。

Directory

Directory
前言
项目重构
共识机制
添加Nonce
PoW实现
调试带PoW的区块链系统
总结

前言

在上一章中我们了解了区块是什么以及区块与区块链之间的关系。在这一章中我们将拓宽区块的头部信息，并讲解区块如何合法的被添加进区块链中。

项目重构

在上一章中，我们所有的代码都写在了main.go中，这显然不利于我们继续构建项目。我们希望main.go只用于最后启动我们的区块链系统，为此我们需要将设计的区块与区块链移植其它文件夹中，这也会帮你go语言的代码管理机制。

首先我们在goblockchain文件夹下建立一个新的文件夹blockchain，在blockchain下分别创建block.go,blockchain.go，之后我们会将相关结构体与函数放入其中。创建utils文件夹，在其下创建util.go用以存放一些龙套函数。创建constcoe文件夹在其下创建constcoe.go用于储存一些全局常量。

打开utils.go，引入以下包。

//util.go

package utils

import (
	"bytes"
	"encoding/binary"
	"log"
)

然后构建一个简单的错误处理函数。

//util.go

func Handle(err error) {
	if err != nil {
		log.Panic(err)
	}
}

然后将我们之前写的int64转字节串函数移动过来。

//util.go

func ToHexInt(num int64) []byte {
	buff := new(bytes.Buffer)
	err := binary.Write(buff, binary.BigEndian, num)
	Handle(err)
	return buff.Bytes()
}

然后我们转至constcoe.go,设置一个后面会用到的全局常量，也就是实现PoW时的难度（后面会详细讲）。

//constcoe.go

package constcoe

const (
	Difficulty = 12
)

打开block.go，将之前写的结构体与函数放入。

//block.go

package blockchain

import (
	"bytes"
	"crypto/sha256"
	"goblockchain/utils"
	"time"
)

type Block struct {
	Timestamp int64
	Hash      []byte
	PrevHash  []byte
	Data      []byte
}

func (b *Block) SetHash() {
	information := bytes.Join([][]byte{utils.ToHexInt(b.Timestamp), b.PrevHash, b.Data}, []byte{})
	hash := sha256.Sum256(information)
	b.Hash = hash[:]
}

func CreateBlock(prevhash, data []byte) *Block {
	block := Block{time.Now().Unix(), []byte{}, prevhash, data}
	block.SetHash()
	return &block
}

func GenesisBlock() *Block {
	genesisWords := "Hello, blockchain!"
	return CreateBlock([]byte{}, []byte(genesisWords))
}

打开blockchain.go，将结构体与函数放入。

//blockchain.go

package blockchain

type BlockChain struct {
	Blocks []*Block
}

func (bc *BlockChain) AddBlock(data string) {
	newBlock := CreateBlock(bc.Blocks[len(bc.Blocks)-1].Hash, []byte(data))
	bc.Blocks = append(bc.Blocks, newBlock)
}

func CreateBlockChain() *BlockChain {
	blockchain := BlockChain{}
	blockchain.Blocks = append(blockchain.Blocks, GenesisBlock())
	return &blockchain
}

回到main.go，只需要调用我们写的blockchain包，就可以启动之前写的区块链系统了。

//main.go

package main

import (
	"fmt"
	"goblockchain/blockchain"
	"time"
)

func main() {
	blockchain := blockchain.CreateBlockChain()
	time.Sleep(time.Second)
	blockchain.AddBlock("After genesis, I have something to say.")
	time.Sleep(time.Second)
	blockchain.AddBlock("Leo Cao is awesome!")
	time.Sleep(time.Second)
	blockchain.AddBlock("I can't wait to follow his github!")
	time.Sleep(time.Second)

	for _, block := range blockchain.Blocks {
		fmt.Printf("Timestamp: %d\n", block.Timestamp)
		fmt.Printf("hash: %x\n", block.Hash)
		fmt.Printf("Previous hash: %x\n", block.PrevHash)
		fmt.Printf("data: %s\n", block.Data)

	}

}

注意此时调用函数CreateBlockChain需要在前面加blockchain的前缀。

共识机制

我们常说区块链是一个分布式系统，系统中每个节点都有机会储存数据信息构造一个区块然后追加到区块链尾部。这里就存在一个问题，那就是当区块链系统中有多个节点都想将自己的区块追加到区块链是我们该怎么办？我们将这些等待添加的区块统称为候选区块，显然我们不能对候选区块全盘照收，否则区块链就不再是一条链而是不同分叉成区块树。那么我们如何确定一种方法来从候选区块中选择一个加入到区块链中了？这里就需要用到区块链的共识机制，后文将以比特币使用的最经典PoW共识机制进行讲解。

共识机制说的通俗明白一点就是要在相对公平的条件下让想要添加区块进区块链的节点内卷，通过竞争选择出一个大家公认的节点添加它的区块进入区块链。整个共识机制被分为两部分，首先是竞争，然后是共识。中本聪在比特币中设计了如下的一个Game来实现竞争：每个节点去寻找一个随机值（也就是nonce），将这个随机值作为候选区块的头部信息属性之一，要求候选区块对自身信息（注意这里是包含了nonce的）进行哈希后表示为数值要小于一个难度目标值（也就是Target），最先寻找到nonce的节点即为卷王，可以将自己的候选区块发布并添加到区块链尾部。这个Game设计的非常巧妙，首先每个节点要寻找到的nonce只对自己候选区块有效，防止了其它节点同学抄答案；其次，nonce的寻找是完全随机的没有技巧，寻找到nonce的时间与目标难度值与节点本身计算性能有关，但不妨碍性能较差的节点也有机会获胜；最后寻找nonce可能耗费大量时间与资源，但是验证卷王是否真的找到了nonce却非常却能够很快完成并几乎不需要耗费资源，这个寻找到的nonce可以说就是卷王真的是卷王的证据。现在我们就来一步一步实现这个Game。

添加Nonce

如前文所说，我们要先增加一些区块的头部信息。

//block.go

import (
	"bytes"
	"crypto/sha256"
	"goblockchain/utils"
	"time"
)

type Block struct {
	Timestamp int64
	Hash      []byte
	PrevHash  []byte
	Target    []byte //This line is new
	Nonce     int64  //This line is new
	Data      []byte
}

Nonce就是节点寻找到的作为卷王的证据。Target就是我们前文说到的目标难度值，将它保存到区块中便于其他节点快速验证Nonce是否正确。

这样一来之前创建的几个函数会报错，我们先暂时不理会。

PoW实现

在blockchain文件夹下创建proofofwork.go，我们来实现之前说到的Game。首先引入以下包。

//proofofwork.go

package blockchain

import (
	"bytes"
	"crypto/sha256"
	"goblockchain/constcoe"
	"goblockchain/utils"
	"math"
	"math/big"
)

我们现在来构建一个可以返回目标难度值的函数。我们这里使用的之前设定的一个常量Difficulty来构造目标难度值，但是在实际的区块链中目标难度值会根据网络情况定时进行调整，且能够保证各节点在同一时间在同一难度下进行竞争，故这里的GetTarget可以理解为预留API，期待一下之后的分布式网络实现。

//proofofwork.go

func (b *Block) GetTarget() []byte {
	target := big.NewInt(1)
	target.Lsh(target, uint(256-constcoe.Difficulty))
	return target.Bytes()
}

Lsh函数就是向左移位，移的越多目标难度值越大，哈希取值落在的空间就更多就越容易找到符合条件的nonce。

每次我们输入一个nonce对应的区块的哈希值都会改变，如下。

//proofofwork.go

func (b *Block) GetBase4Nonce(nonce int64) []byte {
	data := bytes.Join([][]byte{
		utils.ToHexInt(b.Timestamp),
		b.PrevHash,
		utils.ToHexInt(int64(nonce)),
		b.Target,
		b.Data,
	},
		[]byte{},
	)
	return data
}

现在对于任意一个区块，我们都能去寻找一个合适的nonce了。

//proofofwork.go

func (b *Block) FindNonce() int64 {
	var intHash big.Int
	var intTarget big.Int
	var hash [32]byte
	var nonce int64
	nonce = 0
	intTarget.SetBytes(b.Target)

	for nonce < math.MaxInt64 {
		data := b.GetBase4Nonce(nonce)
		hash = sha256.Sum256(data)
		intHash.SetBytes(hash[:])
		if intHash.Cmp(&intTarget) == -1 {
			break
		} else {
			nonce++
		}
	}
	return nonce
}

可以看到，神秘的nonce不过是从0开始取的整数而已，随着不断尝试，每次失败nonce就加1直到由当前nonce得到的区块哈希转化为数值小于目标难度值为止。

我们再来实现一个快速验证卷王是卷王的函数。

//proofofwork.go

func (b *Block) ValidatePoW() bool {
	var intHash big.Int
	var intTarget big.Int
	var hash [32]byte
	intTarget.SetBytes(b.Target)
	data := b.GetBase4Nonce(b.Nonce)
	hash = sha256.Sum256(data)
	intHash.SetBytes(hash[:])
	if intHash.Cmp(&intTarget) == -1 {
		return true
	}
	return false
}

好了，PoW我们已经实现了。回到block.go，调整以下函数。

//block.go

func (b *Block) SetHash() {
	information := bytes.Join([][]byte{utils.ToHexInt(b.Timestamp), b.PrevHash, b.Target, utils.ToHexInt(b.Nonce), b.Data}, []byte{})
	hash := sha256.Sum256(information)
	b.Hash = hash[:]
}

func CreateBlock(prevhash, data []byte) *Block {
	block := Block{time.Now().Unix(), []byte{}, prevhash, []byte{}, 0, data}
	block.Target = block.GetTarget()
	block.Nonce = block.FindNonce()
	block.SetHash()
	return &block
}

调试带PoW的区块链系统

现在打开main.go，我们可以编写程序启动我们的区块链系统了。

//main.go

package main

import (
	"fmt"
	"goblockchain/blockchain"
	"time"
)

func main() {
	chain := blockchain.CreateBlockChain()
	time.Sleep(time.Second)
	chain.AddBlock("After genesis, I have something to say.")
	time.Sleep(time.Second)
	chain.AddBlock("Leo Cao is awesome!")
	time.Sleep(time.Second)
	chain.AddBlock("I can't wait to follow his github!")
	time.Sleep(time.Second)

	for _, block := range chain.Blocks {
		fmt.Printf("Timestamp: %d\n", block.Timestamp)
		fmt.Printf("hash: %x\n", block.Hash)
		fmt.Printf("Previous hash: %x\n", block.PrevHash)
		fmt.Printf("nonce: %d\n", block.Nonce)
		fmt.Printf("data: %s\n", block.Data)
		fmt.Println("Proof of Work validation:", block.ValidatePoW())
	}
}

老样子在terminal中敲下go run main.go，得到如下结果。

D:\learngo\goblockchain>go run main.go
Timestamp: 1632558933
hash: 5ee3e13ce051362c7fa4a5c0dcf4883fec97affb66521d02dcd1bfc5ac14c142
Previous hash:
nonce: 1623
data: Hello, blockchain!
Proof of Work validation: true
Timestamp: 1632558934
hash: 611739fb51de47b7bb472444517a4a9c7d590f9f17fc54ef13fe79ddebefa0e2
Previous hash: 5ee3e13ce051362c7fa4a5c0dcf4883fec97affb66521d02dcd1bfc5ac14c142
nonce: 3369
data: After genesis, I have something to say.
Proof of Work validation: true
Timestamp: 1632558935
hash: eca9fe17530d867ee8212084eae6d54581d152fd01f0c0c64f2143adbbf6a9cb
Previous hash: 611739fb51de47b7bb472444517a4a9c7d590f9f17fc54ef13fe79ddebefa0e2
nonce: 4541
data: Leo Cao is awesome!
Proof of Work validation: true
Timestamp: 1632558936
hash: edaff517ca77b1f6705227b1bf40f311aeb0207014581318f2c98a0fe3842723
Previous hash: eca9fe17530d867ee8212084eae6d54581d152fd01f0c0c64f2143adbbf6a9cb
nonce: 1325
data: I can't wait to follow his github!
Proof of Work validation: true

可以看到我们的PoW验证运行正常。你可以尝试将全局变量Difficulty改大来增加难度值，然后观察一下各个区块的timestamp变化。

总结

本章讲解了PoW共识机制，需要重点理解nonce与目标难度值，以及卷王。在一章中我们将会讲解区块中的数据信息存储方式，涉及UTXO模型。

Directory

Directory
前言
创建项目
区块与区块链
哈希
区块创建与创始区块
运行区块链系统
总结

前言

有时觉得前言什么的可以省略，但作为一个教程还是应该在最开始的时候说两句。

这个系列的教程目的是使用Golang由浅入深地还原PoW共识机制最基础区块链系统（参照比特币），适合想要快速入门区块链核心技术的读者，当然也适合刚学完Go基础语法希望练手的读者。相较于网络上其它Go语言实现区块链的教程，本系列教程以最终建立一个可以分布式运行的区块链系统为目标，使用新版本的Golang（v1.17）及相关包，在还原的过程中会尽量说明一些常规教程忽视的细节，给出所有的代码。

大部分人对于区块链技术的学习常常停留在表象，了解了UTXO模型，共识机制，P2P网络后，总是迫不及待地就想用区块链往所有涉及隐私与安全的问题里套。我始终持有的观点是，可以允许区块链在各种应用场景中试错，但不应该过于推崇区块链，它不是万能的，要学习区块链就应该了解其本质，从事物的两面性去研究它。举个例子，如果我们把区块链当作一个分布式的数据库看待，那么它的性能无疑是拉跨的，但是如果充分理解区块链的本质，就能够明白比特币为什么仅仅用一段代码就能够在全球没有第三方机构的参与下实现资产信息的长久保存，稳定运行超过五年，感叹其精妙之处。

如果想要了解区块链技术的本质，了解区块链技术的优缺点以及可能的研究方向，最直接的办法无疑是阅读比特币源码。但是比特币源码使用C进行编写，人们常常不知道从哪里开始进行学习理解，在耗费大量时间与精力的过程中，渐渐消磨学习者对区块链的兴趣。本人在学习区块链的过程中发现，通过理解区块链原理一步一步构建区块链系统比直接阅读比特币源码要有乐趣的多，学习速度也更快，而且最终都能达到同样的目的，就像B树的建立往往比B树的查找容易学习理解一样。

为了达到使读者快速入门并理解区块链核心技术的目的，本教程不太注重对Go语言相关问题的讲解，关注的是代码背后的区块链原理与实现细节。读者需要做的只是创建一个空项目，从零开始跟随本教程敲下一行又一行的代码。对读者来说，重要的不是对我所写的代码进行改动，而是理解每一行代码的意义。

由于本人Go语言也还处于学习阶段，故一些部分代码可能较为冗余。区块链实现细节有误的地方也随时欢迎讨论指正。

完整的项目地址：https://github.com/leo201313/Blockchain_with_Go

创建项目

首先我们需要创建一个项目。创建一个文件夹命名为goblockchain（当然你也可以取一个霸气的名字，比如lighteningchain,goXchain等等），然后使用VS（Visual Studio Code，推荐使用VS作为IDE）打开文件夹，如下图

此时文件中什么都没有，我们使用go mod来初始化项目，点击VS左下方的小三角，在terminal中输入go mod init goblockchain.

此时文件夹中将会多出一个go.mod文件，证明项目已经初始化成功。在goblockchain文件夹下创建main.go文件。

区块与区块链

区块链以区块（block）的形式储存数据信息，一个区块记录了一段时间内系统或网络中产生的重要数据信息，区块通过引用上一个区块的hash值来连接上一个区块这样区块就按时间顺序排列形成了一条链。每个区块应该包含头部（head）信息用于总结性的描述这个区块，然后在区块的数据存放区（body）中存放要保存的重要数据。首先我们需要初始化main.go，并导入一些基本的包。

//main.go

package main

import (
	"bytes"
	"crypto/sha256"
	"encoding/binary"
	"fmt"
	"log"
	"time"
)

func main {
	
}

然后定义区块的结构体。

//main.go

type Block struct{
	Timestamp int64
	Hash []byte
	PrevHash []byte
	Data []byte
}

我们定义的区块中有时间戳，本身的哈希值，指向上一个区块的哈希这三个属性构成头部信息，而区块中的数据以Data属性表示。在获得了区块后，我们可以定义区块链。

//main.go

type BlockChain struct{
	Blocks []*Block
}

可以看到我们这里的区块链就是区块的一个集合。好了，现在你已经掌握了区块与区块链了，现在就可以去搭建自己的区块链系统了。

哈希

QVQ，好吧，我们现在来给我们的区块增加点细节，来看看它们是怎么连接起来的。对于一个区块而言，可以通过哈希算法概括其所包含的所有信息，哈希值就相当于区块的ID值，同时也可以用来检查区块所包含信息的完整性。哈希函数构造如下。

//main.go

func (b *Block) SetHash() {
	information := bytes.Join([][]bytes{ToHexInt(b.Timestamp),b.PrevHash,b.Data},[]byte{})
	hash := sha256.Sum256(information)
	b.Hash = hash[:]
}

func ToHexInt(num int64) []byte {
	buff := new(bytes.Buffer)
	err := binary.Write(buff, binary.BigEndian, num)
	if err != nil {
		log.Panic(err)
	}
	return buff.Bytes()
}

information变量是将区块的各项属性串联之后的字节串。这里提醒一下bytes.Join可以将多个字节串连接，第二个参数是将字节串连接时的分隔符，这里设置为[]byte{}即为空，ToHexInt将int64转换为字节串类型。然后我们对information做哈希就可以得到区块的哈希值了。

区块创建与创始区块

既然我们可以获得区块的哈希值了，我们就能够创建区块了。

//main.go

func CreateBlock(prevhash, data []byte) *Block {
	block := Block{time.Now().Unix(), []byte{}, prevhash, data}
	block.SetHash()
	return &block
}

可以看到在创建一个区块时一定要引用前一个区块的哈希值，这里会有一个问题，那就是区块链中的第一个区块怎么创建？其实，在区块链中有一个创世区块，随着区块链的创建而添加，它指向的上一个区块的哈希值为空。

//main.go

func GenesisBlock() *Block {
	genesisWords := "Hello, blockchain!"
	return CreateBlock([]byte{}, []byte(genesisWords))
}

可以看到我们在创始区块中存放了 Hello, blockchain! 这段信息。现在我们来构建函数，使得区块链可以根据其它信息创建区块进行储存。

 //main.go

func (bc *BlockChain) AddBlock(data string) {
	newBlock := CreateBlock(bc.Blocks[len(bc.Blocks)-1].Hash, []byte(data))
	bc.Blocks = append(bc.Blocks, newBlock)
}

最后我们构建一个区块链初始化函数，使其返回一个包含创始区块的区块链。

//main.go

func CreateBlockChain() *BlockChain {
	blockchain := BlockChain{}
	blockchain.Blocks = append(blockchain.Blocks, GenesisBlock())
	return &blockchain
}

运行区块链系统

现在我们已经拥有了所有创建区块链需要的函数了，来看看我们的区块链是怎么运作的。

//main.go

func main() {
	blockchain := CreateBlockChain()
	time.Sleep(time.Second)
	blockchain.AddBlock("After genesis, I have something to say.")
	time.Sleep(time.Second)
	blockchain.AddBlock("Leo Cao is awesome!")
	time.Sleep(time.Second)
	blockchain.AddBlock("I can't wait to follow his github!")
	time.Sleep(time.Second)

	for _, block := range blockchain.Blocks {
		fmt.Printf("Timestamp: %d\n", block.Timestamp)
		fmt.Printf("hash: %x\n", block.Hash)
		fmt.Printf("Previous hash: %x\n", block.PrevHash)
		fmt.Printf("data: %s\n", block.Data)

	}

}

在terminal中输入go run main.go，输出如下。

D:\learngo\goblockchain>go run main.go
Timestamp: 1632471455
hash: 289c596026a32c6ac5702fd2d3c96104d6b7178de49beb70a71c100ee839ac26
Previous hash:
data: Hello, blockchain!
Timestamp: 1632471456
hash: a29d04ef59529bb50b1526393203ebf7cc60d8f0ddfbb09900475c9dcf180d3b
Previous hash: 289c596026a32c6ac5702fd2d3c96104d6b7178de49beb70a71c100ee839ac26
data: After genesis, I have something to say.
Timestamp: 1632471457
hash: 69eb263ab680cc0d45530c5ba0db1514255c891e084c3f04bfb416f0f1b06a59
Previous hash: a29d04ef59529bb50b1526393203ebf7cc60d8f0ddfbb09900475c9dcf180d3b
data: Leo Cao is awesome!
Timestamp: 1632471458
hash: 453ff251f95183c92ace277dec4181d5c71582129dc31cce3ceb37c7b1377efc
Previous hash: 69eb263ab680cc0d45530c5ba0db1514255c891e084c3f04bfb416f0f1b06a59
data: I can't wait to follow his github!

你需要注意的是创始区块没有Previous Hash，同时后面的每一个区块都保留了前一个区块的哈希值。

总结

在本章中，我们构建了一个最简单的区块链模型。本章需要重点理解区块与区块链的关系，区块的哈希值的意义，以及创世区块的构建。在下一章中，我们将讲解PoW(Proof of Work)共识机制，并增加一些区块结构体的头部信息。

北京网电创新科技有限公司，为此次活动全程提供室内，户外，有线和无线网络全覆盖，保证活动的顺利进行。

北京网电创新科技有限公司，为此次活动全程提供室内，户外，有线和无线网络全覆盖，保证活动的顺利进行。