这是关于以太坊交易池的第三篇文章，第一篇是[整体概况以太坊交易池]({{< ref "/part2/miner">}})，第二篇是讲解[以太坊本地交易存储]({{< ref "txJournal.md" >}})。而第三篇文章详解一笔交易时如何进入交易池，以及影响。内容较多，请坐好板凳。

交易进入交易池分三步走：校验、入队列、容量检查。拿 AddLocalTx举例。核心代码集中在交易池的func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error)方法。

校验交易合法性

任何交易进入交易池之前均需要校验交易数据的合法性。如果交易校验失败则拒绝此交易。

//core/tx_pool.go:662
if err := pool.validateTx(tx, local); err != nil {
   log.Trace("Discarding invalid transaction", "hash", hash, "err", err)
   invalidTxCounter.Inc(1)
   return false, err
}

那么是如何进行校验的呢？代码逻辑集中在func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error方法中。

首先是防止DOS攻击，不允许交易数据超过32KB。

if tx.Size() > 32*1024 {
   return ErrOversizedData
}

接着不允许交易的转账金额为负数，实际上这次判断难以命中，原因是从外部接收的交易数据属RLP编码，是无法处理负数的。当然这里做一次校验，更加保险。

if tx.Value().Sign() < 0 {
   return ErrNegativeValue
}

交易在虚拟机中执行时将消耗GAS，为了防止程序错误，允许用户在交易中携带一个GAS上限，防止意外发生。同样，为了避免区块总消耗异常，和控制区块数据大小。也同样存在区块GAS上限。而区块中的GAS量是每笔交易执行消耗GAS之和，故不可能一笔交易的GAS上限超过区块GAS限制。一旦超过，这笔交易不可能会打包到区块中，则可在交易池中直接拒绝超过限制的交易。

if pool.currentMaxGas < tx.Gas() {
   return ErrGasLimit
}

每笔交易都需要携带[交易签名]({{< ref "part3/sign-and-valid.md" >}})信息，并从签名中解析出签名者地址。只有合法的签名才能成功解析出签名者。一旦解析失败拒绝此交易。

from, err := types.Sender(pool.signer, tx)
if err != nil {
   return ErrInvalidSender
}

既然知道是交易发送者(签名者)，那么该发送者也可能是来自于交易池所标记的local账户。因此当交易不是local交易时，还进一步检查是否属于local账户。

local = local || pool.locals.contains(from)

如果不是local交易，那么交易的GasPrice 也必须不小于交易池设定的最低GasPrice。这样的限制检查，允许矿工自行决定GasPrice。有些矿工，可能只愿意处理愿意支付高手续费的交易。当然local交易则忽略，避免将本地产生的交易拦截。

if !local && pool.gasPrice.Cmp(tx.GasPrice()) > 0 {
   return ErrUnderpriced
}

以太坊中每个[账户]({{< ref "part1/account.md" >}})都有一个数字类型的 Nonce 字段。是一个有序数字，一次比一次大。虚拟机每执行一次该账户的交易，则新 Nonce 将在此交易的Nonce上加1。如果使用恰当，该 Nonce 可间接表示已打包了 Nonce 笔该账户交易。既然不会变小，那么在交易池中不允许出现交易的Nonce 小于此账户当前Nonce的交易。

if pool.currentState.GetNonce(from) > tx.Nonce() {
   return ErrNonceTooLow
}

如果交易被打包到区块中，应该花费多少手续费呢？虽然无法知道最终花费多少，但至少花费多少手续费是可预知的。手续费加上本次交易转移的以太币数量，将会从该账户上扣除。那么账户至少需要转移多少以太坊是明确的。

因此在交易池中，将检查该账户余额，只有账户资产充足时，才允许交易继续，否则在虚拟机中执行交易，交易也必将失败。

if pool.currentState.GetBalance(from).Cmp(tx.Cost()) < 0 {
   return ErrInsufficientFunds
}

我们不但知道最低花费，也可以知道将最低花费多少GAS。因此也检查交易所设置的Gas上限是否正确。一旦交易至少需要2万Gas，而交易中设置的Gas上限确是 1万GAS。那么交易必然失败，且剩余了 1万GAS。

intrGas, err := IntrinsicGas(tx.Data(), tx.To() == nil, pool.homestead)
if err != nil {
   return err
}
if tx.Gas() < intrGas {
   return ErrIntrinsicGas
}

因此，在最后。如果交易GAS上限低于已知的最低GAS开销，则拒绝这笔必将失败的交易。

交易入队列

在交易池中并不是一个队列管理数据，而是由多个数据集一起管理交易。

如上图所示，交易池先采用一个 txLookup (内部为map）跟踪所有交易。同时将交易根据本地优先，价格优先原则将交易划分为两部分 queue 和 pending。而这两部交易则按账户分别跟踪。

在进入交易队列前，将判断所有交易队列 all 是否已经达到上限。如果到底上限，则需要从交易池或者当前交易中移除优先级最低交易 。

//core/tx_pool.go:668
if uint64(pool.all.Count()) >= pool.config.GlobalSlots+pool.config.GlobalQueue { //❶
   if !local && pool.priced.Underpriced(tx, pool.locals) {//❷
      log.Trace("Discarding underpriced transaction", "hash", hash, "price", tx.GasPrice())
      underpricedTxCounter.Inc(1)
      return false, ErrUnderpriced
   }
   drop := pool.priced.Discard(pool.all.Count()-int(pool.config.GlobalSlots+pool.config.GlobalQueue-1), pool.locals) //❸
   for _, tx := range drop {
      log.Trace("Discarding freshly underpriced transaction", "hash", tx.Hash(), "price", tx.GasPrice())
      underpricedTxCounter.Inc(1)
      pool.removeTx(tx.Hash(), false)
   }
}

那么哪些交易的优先级最低呢？首先，本地交易是受保护的，因此如果交易来自remote 时，将检查该交易的价格是否是整个交易池中属于最低价格的。如果是，则拒绝该交易❷。否则在加入此交易前，将从交易队列 all 中删除价格最低的一部分交易❸。为了高效获得不同价格的交易，交易池已经将交易按价格从低到高实施排列存储在 pool.priced中。

解决交易容量问题后，这笔交易过关斩将，立马将驶入交易内存池中。上图中，交易是有根据 from 分组管理，且一个 from 由分非可执行交易队列（queue）和可执行交易队列（pending）。新交易默认是要在非可执行队列中等待指示，但是一种情况时，如果该 from 的可执行队列中存在一个相同 nonce 的交易时，需要进一步识别是否能替换❹。

怎样的交易才能替换掉已在等待执行的交易呢？以太坊早起的默认设计是，只要价格(gasPrice)高于原交易，则允许替换。但是17年7月底在 #15401被改进。人们愿意支付更多手续费的原因有两种情况，一是急于处理交易，但如果真是紧急交易，那么在发送交易之处，会使用高于推荐的gasprice来处理交易。另一种情况时，以太坊价格下跌，人们愿意支付更多手续费。上调多少手续费是合理的呢？以太币下跌10%，那么便可以上调10%的手续费，毕竟对于用户来说，手续费的面值是一样的。交易池的默认配置（pool.config.PriceBump）是10%，只有上调10%手续费的交易才允许替换掉已在等待执行的交易❺。一旦可以替换，则替换掉旧交易❺，移除旧交易❻，并将交易同步存储到 all 交易内存池中。

//core/tx_pool.go:685
if list := pool.pending[from]; list != nil && list.Overlaps(tx) {//❹
   inserted, old := list.Add(tx, pool.config.PriceBump)//❺
   if !inserted {
      pendingDiscardCounter.Inc(1)
      return false, ErrReplaceUnderpriced
   }
   if old != nil { //❻
      pool.all.Remove(old.Hash())
      pool.priced.Removed()
      pendingReplaceCounter.Inc(1)
   }
   pool.all.Add(tx)
   pool.priced.Put(tx)
   pool.journalTx(from, tx)
   //...
   return old != nil, nil
}
replace, err := pool.enqueueTx(hash, tx)//❼
if err != nil {
    return false, err
}

检查完是否需要替换 pending 交易后，则将交易存入非可执行队列❼。同样，在进入非可执行队列之前，也要检查是否需要替换掉相同 nonce 的交易❽。

func (pool *TxPool) enqueueTx(hash common.Hash, tx *types.Transaction) (bool, error) {
   //...
   inserted, old := pool.queue[from].Add(tx, pool.config.PriceBump) //❽
   if !inserted {
      queuedDiscardCounter.Inc(1)
      return false, ErrReplaceUnderpriced
   }
   if old != nil {
      pool.all.Remove(old.Hash())
      pool.priced.Removed()
      queuedReplaceCounter.Inc(1)
   }
   if pool.all.Get(hash) == nil {
      pool.all.Add(tx)
      pool.priced.Put(tx)
   }
   return old != nil, nil
}

最后，如果交易属于本地交易还需要额外关照。如果交易属于本地交易，但是本地账户集中不存在此 from 时，更新本地账户集❾，避免交易无法被存储⑩。另外，如果已开启存储本地交易，则实时存储本地交易⑪。

// core/tx_pool.go:715
if local {
   if !pool.locals.contains(from) {
      log.Info("Setting new local account", "address", from)
      pool.locals.add(from)//❾
   }
}
pool.journalTx(from, tx)
//....
//core/tx_pool.go:757
func (pool *TxPool) journalTx(from common.Address, tx *types.Transaction) {
    // Only journal if it's enabled and the transaction is local
    if pool.journal == nil || !pool.locals.contains(from) {//⑩
        return
    }
    if err := pool.journal.insert(tx); err != nil {//⑪
        log.Warn("Failed to journal local transaction", "err", err)
    }
}

至此，一笔交易经过千山万水，进入了交易内存池，等待执行。

另外，不难看出，priced 队列是在交易进入队列内存池时便被编排到priced 队列，已让 priced 队列是对 all 交易内存池的同步排序。且交易是在进入pending 队列或者 queue 队列后，才同步更新到 all 交易内存池中。

这里不打算讲解 pending 和 queue 队列的内部实现，请自行研究。因为忽略技术细节不会影响你对以太坊各个技术点，模块的理解。下一讲讲解交易池内存容量处理。
转载自：https://learnblockchain.cn/books/geth/part2/txpool/txaddtx.html

在交易存入交易池后，将影响pending格局。原因是在交易排队等待处理时，需要确定交易优先级。 如果交易池已有一万笔交易排队，该如何按优先级排队来处理呢？如果第一万零一笔交易刚刚加入交易池，需要优先处理此交易吗？如果待执行交易已经有十笔交易被执行完毕，如何从队列中转移一部分交易来添补空缺呢？如果交易将超过交易池配置上限呢？

不管如果变化，以太坊以不变应万变。用统一的优先级规则来应当各种情况，只要有交易加入或者清理出交易池都将立即激活对可执行交易队列的更新（promoteExecutables）。

规则是：删除无效和超上限交易、转移一部分、容量控制。虽然概括为一句话，但逻辑的整个处理确是整个交易池中最复杂的部分，也是最核心部分。

删除旧交易

当新区块来到时，很有可能包含交易内存池中一些账户的交易。一旦存在，则意味着账户的 nonce 和账户余额被存在变动。而只有高于当前新 nonce 交易才能被执行，且账户余额不足以支撑交易执行时，交易也将执行失败。

因此，在新区块来到后，删除所有低于新nonce的交易❶。 再根据账户可用余额，来移除交易开销（amount+gasLimit*gasPrice）高于此余额的交易❷。

//core/tx_pool.go:982
for _, tx := range list.Forward(pool.currentState.GetNonce(addr)) {//❶
   hash := tx.Hash()
   log.Trace("Removed old queued transaction", "hash", hash)
   pool.all.Remove(hash)
   pool.priced.Removed()
}
drops, _ := list.Filter(pool.currentState.GetBalance(addr), pool.currentMaxGas)//❷
for _, tx := range drops {
    hash := tx.Hash()
    log.Trace("Removed unpayable queued transaction", "hash", hash)
    pool.all.Remove(hash)
    pool.priced.Removed()
    queuedNofundsCounter.Inc(1)
}

转移交易或释放

在非可执行队列中的交易有哪些可以转移到可执行队列呢？因为交易 nonce 的缘故，如果queue队列中存在低于 pending 队列的最小nonce的交易❸，则可直接转移到pending中❹。

for _, tx := range list.Ready(pool.pendingState.GetNonce(addr)) {//
   hash := tx.Hash()
   if pool.promoteTx(addr, hash, tx) {//❹
      log.Trace("Promoting queued transaction", "hash", hash)
      promoted = append(promoted, tx)
   }
}

转移后，该账户的交易可能超过所允许的排队交易笔数，如果超过则直接移除超过上限部分的交易❺。当然这仅仅针对remote交易。

if !pool.locals.contains(addr) {
   for _, tx := range list.Cap(int(pool.config.AccountQueue)) {//❺
      hash := tx.Hash()
      pool.all.Remove(hash)
      pool.priced.Removed() 
      //...
   }
}

至此，每个账户的非可执行交易更新完毕。随后，需要检查可执行交易队列情况。

检查pending 交易数量

[交易池配置]({{< ref "./txPool.md" >}}#交易池配置)有设置总pending量上限(pool.config.GlobalSlots)。如果超过上限❶，则分两种策略移除超限部分。

pending := uint64(0)
for _, list := range pool.pending {
   pending += uint64(list.Len())
}
if pending > pool.config.GlobalSlots {//❶
    //...
}

优先从超上限(pool.config.AccountSlots)的账户❷中移除交易。在移除交易时，并非将某个账户的交易全部删除，而是每个账户轮流❸删除一笔交易❹，直到低于交易上限❺。同时，还存在一个特殊删除策略，并非直接轮流每个账户，而是通过一个动态阀值控制❻，阀值控制遍历顺序，存在一定的随机性。

//core/tx_pool.go:1032
spammers := prque.New(nil)
for addr, list := range pool.pending { 
   if !pool.locals.contains(addr) && uint64(list.Len()) > pool.config.AccountSlots {
      spammers.Push(addr, int64(list.Len()))//❷
   }
} 
offenders := []common.Address{}
for pending > pool.config.GlobalSlots && !spammers.Empty() {
   offender, _ := spammers.Pop()
   offenders = append(offenders, offender.(common.Address))

   if len(offenders) > 1 { 
      threshold := pool.pending[offender.(common.Address)].Len()//❻

      for pending > pool.config.GlobalSlots && pool.pending[offenders[len(offenders)-2]].Len() > threshold { //❺
         for i := 0; i < len(offenders)-1; i++ {//❸
            list := pool.pending[offenders[i]]
            for _, tx := range list.Cap(list.Len() - 1) {//❹ 
              //delete tx
            }
            pending--
         }
      }
   }
}

如果仍然还超限，则继续采用直接遍历方式❼，删除交易，直到低于限制❽。

//core/tx_pool.go:1073
if pending > pool.config.GlobalSlots && len(offenders) > 0 {
   for pending > pool.config.GlobalSlots && uint64(pool.pending[offenders[len(offenders)-1]].Len()) > pool.config.AccountSlots {//❽
      for _, addr := range offenders {//❼
         list := pool.pending[addr]
         for _, tx := range list.Cap(list.Len() - 1) { 
                        //delete tx
         }
         pending--
      }
   }
}

检查 queue 交易数量

同样，交易池对于非可执行交易数量也存在上限控制。如果超过上限❶，同样需要删除超限部分。

//core/tx_pool.go:1096
queued := uint64(0)
for _, list := range pool.queue {
   queued += uint64(list.Len())
}
if queued > pool.config.GlobalQueue {//❶
    //...
}

删除交易的策略完成根据每个账户pending交易的时间处理，依次删除长时间存在于pending的账户交易。在交易进入pending 时会更新账户级的心跳时间，代表账户最后pending交易活动时间。时间越晚，说明交易越新。

当交易池的交易过多时，以太坊首先根据账户活动时间，从早到晚排列❷。 再按时间从晚到早❸依次交易。删除时，如果queue交易笔数不够待删除量时❹，直接清理该账户所有queue交易。否则逐个删除，直到到达删除任务❺。

addresses := make(addressesByHeartbeat, 0, len(pool.queue))
for addr := range pool.queue {
   if !pool.locals.contains(addr) { // don't drop locals
      addresses = append(addresses, addressByHeartbeat{addr, pool.beats[addr]})
   }
}
sort.Sort(addresses)//❷

for drop := queued - pool.config.GlobalQueue; drop > 0 && len(addresses) > 0; {
   addr := addresses[len(addresses)-1]//❸
   list := pool.queue[addr.address]

   addresses = addresses[:len(addresses)-1] 
   if size := uint64(list.Len()); size <= drop {//❹
      for _, tx := range list.Flatten() {
         pool.removeTx(tx.Hash(), true)
      }
      drop -= size
      queuedRateLimitCounter.Inc(int64(size))
      continue
   } 
   txs := list.Flatten()
   for i := len(txs) - 1; i >= 0 && drop > 0; i-- {//❺
      pool.removeTx(txs[i].Hash(), true)
      drop--
      queuedRateLimitCounter.Inc(1)
   }
}

也许，你所有疑惑。为何是删除最新活动账户的Queue交易呢？这是因为账户是最新活动，意味着该账户有刚交易进入 pending ，此账户的交易是更有机会被执行的。那么公平起见，哪些迟迟未能进入 pending 的账户的 queue 交易应该继续保留，以便账户交易有机会进入 pending 。这样对于每个账户来说，长时间等待过程中都是有机会进入 pending 被矿工处理的。
转载自：https://learnblockchain.cn/books/geth/part2/txpool/txpromote.html

稍后补充

参考

https://github.com/MyCryptoHQ/eth-scan
https://github.com/ethereum-lists/tokens
https://github.com/ethfaction/erc20_tokens
https://github.com/wbobeirne/eth-balance-checker
https://github.com/web3labs/erc20-rest-service
https://blog.csdn.net/mongo_node/article/details/82146665

前言

这篇文章从区块传播策略入手，介绍新区块是如何传播到远端节点，以及新区块加入到远端节点本地链的过程，同时会介绍fetcher模块，fetcher的功能是处理Peer通知的区块信息。在介绍过程中，还会涉及到p2p，eth等模块，不会专门介绍，而是专注区块的传播和加入区块链的过程。

当前代码是以太坊Release 1.8，如果版本不同，代码上可能存在差异。

总体过程和传播策略

本节从宏观角度介绍，节点产生区块后，为了传播给远端节点做了啥，远端节点收到区块后又做了什么，每个节点都连接了很多Peer，它传播的策略是什么样的？

总体流程和策略可以总结为，传播给远端Peer节点，Peer验证区块无误后，加入到本地区块链，继续传播新区块信息。具体过程如下。

先看总体过程。产生区块后，miner模块会发布一个事件NewMinedBlockEvent，订阅事件的协程收到事件后，就会把新区块的消息，广播给它所有的peer，peer收到消息后，会交给自己的fetcher模块处理，fetcher进行基本的验证后，区块没问题，发现这个区块就是本地链需要的下一个区块，则交给blockChain进一步进行完整的验证，这个过程会执行区块所有的交易，无误后把区块加入到本地链，写入数据库，这个过程就是下面的流程图，图1。

总体流程图，能看到有个分叉，是因为节点传播新区块是有策略的。它的传播策略为：

1. 假如节点连接了N个Peer，它只向Peer列表的sqrt(N)个Peer广播完整的区块消息。
2. 向所有的Peer广播只包含区块Hash的消息。

策略图的效果如图2，红色节点将区块传播给黄色节点：

收到区块Hash的节点，需要从发送给它消息的Peer那里获取对应的完整区块，获取区块后就会按照图1的流程，加入到fetcher队列，最终插入本地区块链后，将区块的Hash值广播给和它相连，但还不知道这个区块的Peer。非产生区块节点的策略图，如图3，黄色节点将区块Hash传播给青色节点：

至此，可以看出以太坊采用以石击水的方式，像水纹一样，层层扩散新产生的区块。

Fetcher模块是干啥的

fetcher模块的功能，就是收集其他Peer通知它的区块信息：1）完整的区块2）区块Hash消息。根据通知的消息，获取完整的区块，然后传递给eth模块把区块插入区块链。

如果是完整区块，就可以传递给eth插入区块，如果只有区块Hash，则需要从其他的Peer获取此完整的区块，然后再传递给eth插入区块。

源码解读

本节介绍区块传播和处理的细节东西，方式仍然是先用图解释流程，再是代码流程。

产块节点的传播新区块

节点产生区块后，广播的流程可以表示为图4：

1. 发布事件
2. 事件处理函数选择要广播完整的Peer，然后将区块加入到它们的队列
3. 事件处理函数把区块Hash添加到所有Peer的另外一个通知队列
4. 每个Peer的广播处理函数，会遍历它的待广播区块队列和通知队列，把数据封装成消息，调用P2P接口发送出去

再看下代码上的细节。

1. worker.wait()函数发布事件NewMinedBlockEvent。

2. ProtocolManager.minedBroadcastLoop()是事件处理函数。它调用了2次pm.BroadcastBlock()。

   // Mined broadcast loop
   func (pm *ProtocolManager) minedBroadcastLoop() {
    // automatically stops if unsubscribe
    for obj := range pm.minedBlockSub.Chan() {
        switch ev := obj.Data.(type) {
        case core.NewMinedBlockEvent:
            pm.BroadcastBlock(ev.Block, true)  // First propagate block to peers
            pm.BroadcastBlock(ev.Block, false) // Only then announce to the rest
        }
    }
   }

3. pm.BroadcastBlock()的入参propagate为真时，向部分Peer广播完整的区块，调用peer.AsyncSendNewBlock()，否则向所有Peer广播区块头，调用peer.AsyncSendNewBlockHash()，这2个函数就是把数据放入队列，此处不再放代码。

   // BroadcastBlock will either propagate a block to a subset of it's peers, or
   // will only announce it's availability (depending what's requested).
   func (pm *ProtocolManager) BroadcastBlock(block *types.Block, propagate bool) {
    hash := block.Hash()
    peers := pm.peers.PeersWithoutBlock(hash)
   
    // If propagation is requested, send to a subset of the peer
    // 这种情况，要把区块广播给部分peer
    if propagate {
        // Calculate the TD of the block (it's not imported yet, so block.Td is not valid)
        // 计算新的总难度
        var td *big.Int
        if parent := pm.blockchain.GetBlock(block.ParentHash(), block.NumberU64()-1); parent != nil {
            td = new(big.Int).Add(block.Difficulty(), pm.blockchain.GetTd(block.ParentHash(), block.NumberU64()-1))
        } else {
            log.Error("Propagating dangling block", "number", block.Number(), "hash", hash)
            return
        }
        // Send the block to a subset of our peers
        // 广播区块给部分peer
        transfer := peers[:int(math.Sqrt(float64(len(peers))))]
        for _, peer := range transfer {
            peer.AsyncSendNewBlock(block, td)
        }
        log.Trace("Propagated block", "hash", hash, "recipients", len(transfer), "duration", common.PrettyDuration(time.Since(block.ReceivedAt)))
        return
    }
    // Otherwise if the block is indeed in out own chain, announce it
    // 把区块hash值广播给所有peer
    if pm.blockchain.HasBlock(hash, block.NumberU64()) {
        for _, peer := range peers {
            peer.AsyncSendNewBlockHash(block)
        }
        log.Trace("Announced block", "hash", hash, "recipients", len(peers), "duration", common.PrettyDuration(time.Since(block.ReceivedAt)))
    }
   }

4. peer.broadcase()是每个Peer连接的广播函数，它只广播3种消息：交易、完整的区块、区块的Hash，这样表明了节点只会主动广播这3中类型的数据，剩余的数据同步，都是通过请求-响应的方式。

   // broadcast is a write loop that multiplexes block propagations, announcements
   // and transaction broadcasts into the remote peer. The goal is to have an async
   // writer that does not lock up node internals.
   func (p *peer) broadcast() {
    for {
        select {
        // 广播交易
        case txs := <-p.queuedTxs:
            if err := p.SendTransactions(txs); err != nil {
                return
            }
            p.Log().Trace("Broadcast transactions", "count", len(txs))
        // 广播完整的新区块
        case prop := <-p.queuedProps:
            if err := p.SendNewBlock(prop.block, prop.td); err != nil {
                return
            }
            p.Log().Trace("Propagated block", "number", prop.block.Number(), "hash", prop.block.Hash(), "td", prop.td)
   
        // 广播区块Hash
        case block := <-p.queuedAnns:
            if err := p.SendNewBlockHashes([]common.Hash{block.Hash()}, []uint64{block.NumberU64()}); err != nil {
                return
            }
            p.Log().Trace("Announced block", "number", block.Number(), "hash", block.Hash())
   
        case <-p.term:
            return
        }
    }
   }

   Peer节点处理新区块

   本节介绍远端节点收到2种区块同步消息的处理，其中NewBlockMsg的处理流程比较清晰，也简洁。NewBlockHashesMsg消息的处理就绕了2绕，从总体流程图1上能看出来，它需要先从给他发送消息Peer那里获取到完整的区块，剩下的流程和NewBlockMsg又一致了。

这部分涉及的模块多，画出来有种眼花缭乱的感觉，但只要抓住上面的主线，代码看起来还是很清晰的。通过图5先看下整体流程。

消息处理的起点是ProtocolManager.handleMsg，NewBlockMsg的处理流程是蓝色标记的区域，红色区域是单独的协程，是fetcher处理队列中区块的流程，如果从队列中取出的区块是当前链需要的，校验后，调用blockchian.InsertChain()把区块插入到区块链，最后写入数据库，这是黄色部分。最后，绿色部分是NewBlockHashesMsg的处理流程，代码流程上是比较复杂的，为了能通过图描述整体流程，我把它简化掉了。

仔细看看这幅图，掌握整体的流程后，接下来看每个步骤的细节。

NewBlockMsg的处理

本节介绍节点收到完整区块的处理，流程如下：

1. 首先进行RLP编解码，然后标记发送消息的Peer已经知道这个区块，这样本节点最后广播这个区块的Hash时，不会再发送给该Peer。
2. 将区块存入到fetcher的队列，调用fetcher.Enqueue。
3. 更新Peer的Head位置，然后判断本地链是否落后于Peer的链，如果是，则通过Peer更新本地链。

只看handle.Msg()的NewBlockMsg相关的部分。

case msg.Code == NewBlockMsg:
    // Retrieve and decode the propagated block
    // 收到新区块，解码，赋值接收数据
    var request newBlockData
    if err := msg.Decode(&request); err != nil {
        return errResp(ErrDecode, "%v: %v", msg, err)
    }
    request.Block.ReceivedAt = msg.ReceivedAt
    request.Block.ReceivedFrom = p

    // Mark the peer as owning the block and schedule it for import
    // 标记peer知道这个区块
    p.MarkBlock(request.Block.Hash())
    // 为啥要如队列？已经得到完整的区块了
    // 答：存入fetcher的优先级队列，fetcher会从队列中选取当前高度需要的块
    pm.fetcher.Enqueue(p.id, request.Block)

    // Assuming the block is importable by the peer, but possibly not yet done so,
    // calculate the head hash and TD that the peer truly must have.
    // 截止到parent区块的头和难度
    var (
        trueHead = request.Block.ParentHash()
        trueTD   = new(big.Int).Sub(request.TD, request.Block.Difficulty())
    )
    // Update the peers total difficulty if better than the previous
    // 如果收到的块的难度大于peer之前的，以及自己本地的，就去和这个peer同步
    // 问题：就只用了一下块里的hash值，为啥不直接使用这个块呢，如果这个块不能用，干嘛不少发送些数据，减少网络负载呢。
    // 答案：实际上，这个块加入到了优先级队列中，当fetcher的loop检查到当前下一个区块的高度，正是队列中有的，则不再向peer请求
    // 该区块，而是直接使用该区块，检查无误后交给block chain执行insertChain
    if _, td := p.Head(); trueTD.Cmp(td) > 0 {
        p.SetHead(trueHead, trueTD)

        // Schedule a sync if above ours. Note, this will not fire a sync for a gap of
        // a singe block (as the true TD is below the propagated block), however this
        // scenario should easily be covered by the fetcher.
        currentBlock := pm.blockchain.CurrentBlock()
        if trueTD.Cmp(pm.blockchain.GetTd(currentBlock.Hash(), currentBlock.NumberU64())) > 0 {
            go pm.synchronise(p)
        }
    }
//------------------------ 以上 handleMsg

// Enqueue tries to fill gaps the the fetcher's future import queue.
// 发给inject通道，当前协程在handleMsg，通过通道发送给fetcher的协程处理
func (f *Fetcher) Enqueue(peer string, block *types.Block) error {
    op := &inject{
        origin: peer,
        block:  block,
    }
    select {
    case f.inject <- op:
        return nil
    case <-f.quit:
        return errTerminated
    }
}

//------------------------ 以下 fetcher.loop处理inject部分
case op := <-f.inject:
    // A direct block insertion was requested, try and fill any pending gaps
    // 区块加入队列，首先也填入未决的间距
    propBroadcastInMeter.Mark(1)
    f.enqueue(op.origin, op.block)

//------------------------  如队列函数

// enqueue schedules a new future import operation, if the block to be imported
// has not yet been seen.
// 把导入的新区块加入到队列，主要操作queue, queues, queued这3个变量，quque用来保存要插入的区块，
// 按高度排序，queues记录了在队列中某个peer传来的区块的数量，用来做对抗DoS攻击，queued用来
// 判断某个区块是否已经在队列，防止2次插入，浪费时间
func (f *Fetcher) enqueue(peer string, block *types.Block) {
    hash := block.Hash()

    // Ensure the peer isn't DOSing us
    // 防止peer的DOS攻击
    count := f.queues[peer] + 1
    if count > blockLimit {
        log.Debug("Discarded propagated block, exceeded allowance", "peer", peer, "number", block.Number(), "hash", hash, "limit", blockLimit)
        propBroadcastDOSMeter.Mark(1)
        f.forgetHash(hash)
        return
    }
    // Discard any past or too distant blocks
    // 高度检查：未来太远的块丢弃
    if dist := int64(block.NumberU64()) - int64(f.chainHeight()); dist < -maxUncleDist || dist > maxQueueDist {
        log.Debug("Discarded propagated block, too far away", "peer", peer, "number", block.Number(), "hash", hash, "distance", dist)
        propBroadcastDropMeter.Mark(1)
        f.forgetHash(hash)
        return
    }
    // Schedule the block for future importing
    // 块先加入优先级队列，加入链之前，还有很多要做
    if _, ok := f.queued[hash]; !ok {
        op := &inject{
            origin: peer,
            block:  block,
        }
        f.queues[peer] = count
        f.queued[hash] = op
        f.queue.Push(op, -float32(block.NumberU64()))
        if f.queueChangeHook != nil {
            f.queueChangeHook(op.block.Hash(), true)
        }
        log.Debug("Queued propagated block", "peer", peer, "number", block.Number(), "hash", hash, "queued", f.queue.Size())
    }
}

fetcher队列处理

本节我们看看，区块加入队列后，fetcher如何处理区块，为何不直接校验区块，插入到本地链？

由于以太坊又Uncle的机制，节点可能收到老一点的一些区块。另外，节点可能由于网络原因，落后了几个区块，所以可能收到“未来”的一些区块，这些区块都不能直接插入到本地链。

区块入的队列是一个优先级队列，高度低的区块会被优先取出来。fetcher.loop是单独协程，不断运转，清理fecther中的事务和事件。首先会清理正在fetching的区块，但已经超时。然后处理优先级队列中的区块，判断高度是否是下一个区块，如果是则调用f.insert()函数，校验后调用BlockChain.InsertChain()，成功插入后，广播新区块的Hash。

// Loop is the main fetcher loop, checking and processing various notification
// events.
func (f *Fetcher) loop() {
    // Iterate the block fetching until a quit is requested
    fetchTimer := time.NewTimer(0)
    completeTimer := time.NewTimer(0)

    for {
        // Clean up any expired block fetches
        // 清理过期的区块
        for hash, announce := range f.fetching {
            if time.Since(announce.time) > fetchTimeout {
                f.forgetHash(hash)
            }
        }
        // Import any queued blocks that could potentially fit
        // 导入队列中合适的块
        height := f.chainHeight()
        for !f.queue.Empty() {
            op := f.queue.PopItem().(*inject)
            hash := op.block.Hash()
            if f.queueChangeHook != nil {
                f.queueChangeHook(hash, false)
            }
            // If too high up the chain or phase, continue later
            // 块不是链需要的下一个块，再入优先级队列，停止循环
            number := op.block.NumberU64()
            if number > height+1 {
                f.queue.Push(op, -float32(number))
                if f.queueChangeHook != nil {
                    f.queueChangeHook(hash, true)
                }
                break
            }
            // Otherwise if fresh and still unknown, try and import
            // 高度正好是我们想要的，并且链上也没有这个块
            if number+maxUncleDist < height || f.getBlock(hash) != nil {
                f.forgetBlock(hash)
                continue
            }
            // 那么，块插入链
            f.insert(op.origin, op.block)
        }

        //省略
    }
}

func (f *Fetcher) insert(peer string, block *types.Block) {
    hash := block.Hash()

    // Run the import on a new thread
    log.Debug("Importing propagated block", "peer", peer, "number", block.Number(), "hash", hash)
    go func() {
        defer func() { f.done <- hash }()

        // If the parent's unknown, abort insertion
        parent := f.getBlock(block.ParentHash())
        if parent == nil {
            log.Debug("Unknown parent of propagated block", "peer", peer, "number", block.Number(), "hash", hash, "parent", block.ParentHash())
            return
        }
        // Quickly validate the header and propagate the block if it passes
        // 验证区块头，成功后广播区块
        switch err := f.verifyHeader(block.Header()); err {
        case nil:
            // All ok, quickly propagate to our peers
            propBroadcastOutTimer.UpdateSince(block.ReceivedAt)
            go f.broadcastBlock(block, true)

        case consensus.ErrFutureBlock:
            // Weird future block, don't fail, but neither propagate

        default:
            // Something went very wrong, drop the peer
            log.Debug("Propagated block verification failed", "peer", peer, "number", block.Number(), "hash", hash, "err", err)
            f.dropPeer(peer)
            return
        }
        // Run the actual import and log any issues
        // 调用回调函数，实际是blockChain.insertChain
        if _, err := f.insertChain(types.Blocks{block}); err != nil {
            log.Debug("Propagated block import failed", "peer", peer, "number", block.Number(), "hash", hash, "err", err)
            return
        }
        // If import succeeded, broadcast the block
        propAnnounceOutTimer.UpdateSince(block.ReceivedAt)
        go f.broadcastBlock(block, false)

        // Invoke the testing hook if needed
        if f.importedHook != nil {
            f.importedHook(block)
        }
    }()
}

NewBlockHashesMsg的处理

本节介绍NewBlockHashesMsg的处理，其实，消息处理是简单的，而复杂一点的是从Peer哪获取完整的区块，下节再看。

流程如下:

1. 对消息进行RLP解码，然后标记Peer已经知道此区块。
2. 寻找出本地区块链不存在的区块Hash值，把这些未知的Hash通知给fetcher。
3. fetcher.Notify记录好通知信息，塞入notify通道，以便交给fetcher的协程。
4. fetcher.loop()会对notify中的消息进行处理，确认区块并非DOS攻击，然后检查区块的高度，判断该区块是否已经在fetching或者comleting(代表已经下载区块头，在下载body)，如果都没有，则加入到announced中，触发0s定时器，进行处理。

关于announced下节再介绍。

// handleMsg()部分
case msg.Code == NewBlockHashesMsg:
    var announces newBlockHashesData
    if err := msg.Decode(&announces); err != nil {
        return errResp(ErrDecode, "%v: %v", msg, err)
    }
    // Mark the hashes as present at the remote node
    for _, block := range announces {
        p.MarkBlock(block.Hash)
    }
    // Schedule all the unknown hashes for retrieval
    // 把本地链没有的块hash找出来，交给fetcher去下载
    unknown := make(newBlockHashesData, 0, len(announces))
    for _, block := range announces {
        if !pm.blockchain.HasBlock(block.Hash, block.Number) {
            unknown = append(unknown, block)
        }
    }
    for _, block := range unknown {
        pm.fetcher.Notify(p.id, block.Hash, block.Number, time.Now(), p.RequestOneHeader, p.RequestBodies)
    }

// Notify announces the fetcher of the potential availability of a new block in
// the network.
// 通知fetcher（自己）有新块产生，没有块实体，有hash、高度等信息
func (f *Fetcher) Notify(peer string, hash common.Hash, number uint64, time time.Time,
    headerFetcher headerRequesterFn, bodyFetcher bodyRequesterFn) error {
    block := &announce{
        hash:        hash,
        number:      number,
        time:        time,
        origin:      peer,
        fetchHeader: headerFetcher,
        fetchBodies: bodyFetcher,
    }
    select {
    case f.notify <- block:
        return nil
    case <-f.quit:
        return errTerminated
    }
}

// fetcher.loop()的notify通道消息处理
case notification := <-f.notify:
    // A block was announced, make sure the peer isn't DOSing us
    propAnnounceInMeter.Mark(1)
    count := f.announces[notification.origin] + 1
    if count > hashLimit {
        log.Debug("Peer exceeded outstanding announces", "peer", notification.origin, "limit", hashLimit)
        propAnnounceDOSMeter.Mark(1)
        break
    }

    // If we have a valid block number, check that it's potentially useful
    // 高度检查
    if notification.number > 0 {
        if dist := int64(notification.number) - int64(f.chainHeight()); dist < -maxUncleDist || dist > maxQueueDist {
            log.Debug("Peer discarded announcement", "peer", notification.origin, "number", notification.number, "hash", notification.hash, "distance", dist)
            propAnnounceDropMeter.Mark(1)
            break
        }
    }

    // All is well, schedule the announce if block's not yet downloading
    // 检查是否已经在下载，已下载则忽略
    if _, ok := f.fetching[notification.hash]; ok {
        break
    }
    if _, ok := f.completing[notification.hash]; ok {
        break
    }
    // 更新peer已经通知给我们的区块数量
    f.announces[notification.origin] = count
    // 把通知信息加入到announced，供调度
    f.announced[notification.hash] = append(f.announced[notification.hash], notification)
    if f.announceChangeHook != nil && len(f.announced[notification.hash]) == 1 {
        f.announceChangeHook(notification.hash, true)
    }
    if len(f.announced) == 1 {
        // 有通知放入到announced，则重设0s定时器，loop的另外一个分支会处理这些通知
        f.rescheduleFetch(fetchTimer)
    }

fetcher获取完整区块

本节介绍fetcher获取完整区块的过程，这也是fetcher最重要的功能，会涉及到fetcher至少80%的代码。单独拉放一大节吧。

Fetcher的大头

Fetcher最主要的功能就是获取完整的区块，然后在合适的实际交给InsertChain去验证和插入到本地区块链。我们还是从宏观入手，看Fetcher是如何工作的，一定要先掌握好宏观，因为代码层面上没有这么清晰。

宏观

首先，看两个节点是如何交互，获取完整区块，使用时序图的方式看一下，见图6，流程很清晰不再文字介绍。

再看下获取区块过程中，fetcher内部的状态转移，它使用状态来记录，要获取的区块在什么阶段，见图7。我稍微解释一下：

1. 收到NewBlockHashesMsg后，相关信息会记录到announced，进入announced状态，代表了本节点接收了消息。
2. announced由fetcher协程处理，经过校验后，会向给他发送消息的Peer发送请求，请求该区块的区块头，然后进入fetching状态。
3. 获取区块头后，如果区块头表示没有交易和uncle，则转移到completing状态，并且使用区块头合成完整的区块，加入到queued优先级队列。
4. 获取区块头后，如果区块头表示该区块有交易和uncle，则转移到fetched状态，然后发送请求，请求交易和uncle，然后转移到completing状态。
5. 收到交易和uncle后，使用头、交易、uncle这3个信息，生成完整的区块，加入到队列queued。

微观

接下来就是从代码角度看如何获取完整区块的流程了，有点多，看不懂的时候，再回顾下上面宏观的介绍图。

首先看Fetcher的定义，它存放了通信数据和状态管理，捡加注释的看，上文提到的状态，里面都有。

// Fetcher is responsible for accumulating block announcements from various peers
// and scheduling them for retrieval.
// 积累块通知，然后调度获取这些块
type Fetcher struct {
    // Various event channels
    // 收到区块hash值的通道
    notify chan *announce
    // 收到完整区块的通道
    inject chan *inject

    blockFilter chan chan []*types.Block
    // 过滤header的通道的通道
    headerFilter chan chan *headerFilterTask
    // 过滤body的通道的通道
    bodyFilter chan chan *bodyFilterTask

    done chan common.Hash
    quit chan struct{}

    // Announce states
    // Peer已经给了本节点多少区块头通知
    announces map[string]int // Per peer announce counts to prevent memory exhaustion
    // 已经announced的区块列表
    announced map[common.Hash][]*announce // Announced blocks, scheduled for fetching
    // 正在fetching区块头的请求
    fetching map[common.Hash]*announce // Announced blocks, currently fetching
    // 已经fetch到区块头，还差body的请求，用来获取body
    fetched map[common.Hash][]*announce // Blocks with headers fetched, scheduled for body retrieval
    // 已经得到区块头的
    completing map[common.Hash]*announce // Blocks with headers, currently body-completing

    // Block cache
    // queue，优先级队列，高度做优先级
    // queues，queued队列中某个peer发来的区块数量
    // queued，等待插入到区块链的区块，实际插入时从queue取，queued就是用来快速判断区块是否在队列的
    queue  *prque.Prque            // Queue containing the import operations (block number sorted)
    queues map[string]int          // Per peer block counts to prevent memory exhaustion
    queued map[common.Hash]*inject // Set of already queued blocks (to dedupe imports)

    // Callbacks
    getBlock       blockRetrievalFn   // Retrieves a block from the local chain
    verifyHeader   headerVerifierFn   // Checks if a block's headers have a valid proof of work，验证区块头，包含了PoW验证
    broadcastBlock blockBroadcasterFn // Broadcasts a block to connected peers，广播给peer
    chainHeight    chainHeightFn      // Retrieves the current chain's height
    insertChain    chainInsertFn      // Injects a batch of blocks into the chain，插入区块到链的函数
    dropPeer       peerDropFn         // Drops a peer for misbehaving

    // Testing hooks
    announceChangeHook func(common.Hash, bool) // Method to call upon adding or deleting a hash from the announce list
    queueChangeHook    func(common.Hash, bool) // Method to call upon adding or deleting a block from the import queue
    fetchingHook       func([]common.Hash)     // Method to call upon starting a block (eth/61) or header (eth/62) fetch
    completingHook     func([]common.Hash)     // Method to call upon starting a block body fetch (eth/62)
    importedHook       func(*types.Block)      // Method to call upon successful block import (both eth/61 and eth/62)
}

NewBlockHashesMsg消息的处理前面的小节已经讲过了，不记得可向前翻看。这里从announced的状态处理说起。loop()中，fetchTimer超时后，代表了收到了消息通知，需要处理，会从announced中选择出需要处理的通知，然后创建请求，请求区块头，由于可能有很多节点都通知了它某个区块的Hash，所以随机的从这些发送消息的Peer中选择一个Peer，发送请求的时候，为每个Peer都创建了单独的协程。

case <-fetchTimer.C:
    // At least one block's timer ran out, check for needing retrieval
    // 有区块通知，去处理
    request := make(map[string][]common.Hash)

    for hash, announces := range f.announced {
        if time.Since(announces[0].time) > arriveTimeout-gatherSlack {
            // Pick a random peer to retrieve from, reset all others
            // 可能有很多peer都发送了这个区块的hash值，随机选择一个peer
            announce := announces[rand.Intn(len(announces))]
            f.forgetHash(hash)

            // If the block still didn't arrive, queue for fetching
            // 本地还没有这个区块，创建获取区块的请求
            if f.getBlock(hash) == nil {
                request[announce.origin] = append(request[announce.origin], hash)
                f.fetching[hash] = announce
            }
        }
    }
    // Send out all block header requests
    // 把所有的request发送出去
    // 为每一个peer都创建一个协程，然后请求所有需要从该peer获取的请求
    for peer, hashes := range request {
        log.Trace("Fetching scheduled headers", "peer", peer, "list", hashes)

        // Create a closure of the fetch and schedule in on a new thread
        fetchHeader, hashes := f.fetching[hashes[0]].fetchHeader, hashes
        go func() {
            if f.fetchingHook != nil {
                f.fetchingHook(hashes)
            }
            for _, hash := range hashes {
                headerFetchMeter.Mark(1)
                fetchHeader(hash) // Suboptimal, but protocol doesn't allow batch header retrievals
            }
        }()
    }
    // Schedule the next fetch if blocks are still pending
    f.rescheduleFetch(fetchTimer)

从Notify的调用中，可以看出，fetcherHeader()的实际函数是RequestOneHeader()，该函数使用的消息是GetBlockHeadersMsg，可以用来请求多个区块头，不过fetcher只请求一个

pm.fetcher.Notify(p.id, block.Hash, block.Number, time.Now(), p.RequestOneHeader, p.RequestBodies)

// RequestOneHeader is a wrapper around the header query functions to fetch a
// single header. It is used solely by the fetcher.
func (p *peer) RequestOneHeader(hash common.Hash) error {
    p.Log().Debug("Fetching single header", "hash", hash)
    return p2p.Send(p.rw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Hash: hash}, Amount: uint64(1), Skip: uint64(0), Reverse: false})
}

GetBlockHeadersMsg的处理如下：因为它是获取多个区块头的，所以处理起来比较“麻烦”，还好，fetcher只获取一个区块头，其处理在20行~33行，获取下一个区块头的处理逻辑，这里就不看了，最后调用SendBlockHeaders()将区块头发送给请求的节点，消息是BlockHeadersMsg。

// handleMsg()
// Block header query, collect the requested headers and reply
case msg.Code == GetBlockHeadersMsg:
    // Decode the complex header query
    var query getBlockHeadersData
    if err := msg.Decode(&query); err != nil {
        return errResp(ErrDecode, "%v: %v", msg, err)
    }
    hashMode := query.Origin.Hash != (common.Hash{})

    // Gather headers until the fetch or network limits is reached
    // 收集区块头，直到达到限制
    var (
        bytes   common.StorageSize
        headers []*types.Header
        unknown bool
    )
    // 自己已知区块 && 少于查询的数量 && 大小小于2MB && 小于能下载的最大数量
    for !unknown && len(headers) < int(query.Amount) && bytes < softResponseLimit && len(headers) < downloader.MaxHeaderFetch {
        // Retrieve the next header satisfying the query
        // 获取区块头
        var origin *types.Header
        if hashMode {
            // fetcher 使用的模式
            origin = pm.blockchain.GetHeaderByHash(query.Origin.Hash)
        } else {
            origin = pm.blockchain.GetHeaderByNumber(query.Origin.Number)
        }
        if origin == nil {
            break
        }
        number := origin.Number.Uint64()
        headers = append(headers, origin)
        bytes += estHeaderRlpSize

        // Advance to the next header of the query
        // 下一个区块头的获取，不同策略，方式不同
        switch {
        case query.Origin.Hash != (common.Hash{}) && query.Reverse:
            // ...
        }
    }
    return p.SendBlockHeaders(headers)

BlockHeadersMsg的处理很有意思，因为GetBlockHeadersMsg并不是fetcher独占的消息，downloader也可以调用，所以，响应消息的处理需要分辨出是fetcher请求的，还是downloader请求的。它的处理逻辑是：fetcher先过滤收到的区块头，如果fetcher不要的，那就是downloader的，在调用fetcher.FilterHeaders的时候，fetcher就将自己要的区块头拿走了。

// handleMsg()
case msg.Code == BlockHeadersMsg:
    // A batch of headers arrived to one of our previous requests
    var headers []*types.Header
    if err := msg.Decode(&headers); err != nil {
        return errResp(ErrDecode, "msg %v: %v", msg, err)
    }
    // If no headers were received, but we're expending a DAO fork check, maybe it's that
    // 检查是不是当前DAO的硬分叉
    if len(headers) == 0 && p.forkDrop != nil {
        // Possibly an empty reply to the fork header checks, sanity check TDs
        verifyDAO := true

        // If we already have a DAO header, we can check the peer's TD against it. If
        // the peer's ahead of this, it too must have a reply to the DAO check
        if daoHeader := pm.blockchain.GetHeaderByNumber(pm.chainconfig.DAOForkBlock.Uint64()); daoHeader != nil {
            if _, td := p.Head(); td.Cmp(pm.blockchain.GetTd(daoHeader.Hash(), daoHeader.Number.Uint64())) >= 0 {
                verifyDAO = false
            }
        }
        // If we're seemingly on the same chain, disable the drop timer
        if verifyDAO {
            p.Log().Debug("Seems to be on the same side of the DAO fork")
            p.forkDrop.Stop()
            p.forkDrop = nil
            return nil
        }
    }
    // Filter out any explicitly requested headers, deliver the rest to the downloader
    // 过滤是不是fetcher请求的区块头，去掉fetcher请求的区块头再交给downloader
    filter := len(headers) == 1
    if filter {
        // If it's a potential DAO fork check, validate against the rules
        // 检查是否硬分叉
        if p.forkDrop != nil && pm.chainconfig.DAOForkBlock.Cmp(headers[0].Number) == 0 {
            // Disable the fork drop timer
            p.forkDrop.Stop()
            p.forkDrop = nil

            // Validate the header and either drop the peer or continue
            if err := misc.VerifyDAOHeaderExtraData(pm.chainconfig, headers[0]); err != nil {
                p.Log().Debug("Verified to be on the other side of the DAO fork, dropping")
                return err
            }
            p.Log().Debug("Verified to be on the same side of the DAO fork")
            return nil
        }
        // Irrelevant of the fork checks, send the header to the fetcher just in case
        // 使用fetcher过滤区块头
        headers = pm.fetcher.FilterHeaders(p.id, headers, time.Now())
    }
    // 剩下的区块头交给downloader
    if len(headers) > 0 || !filter {
        err := pm.downloader.DeliverHeaders(p.id, headers)
        if err != nil {
            log.Debug("Failed to deliver headers", "err", err)
        }
    }

FilterHeaders()是一个很有大智慧的函数，看起来耐人寻味，但实在妙。它要把所有的区块头，都传递给fetcher协程，还要获取fetcher协程处理后的结果。fetcher.headerFilter是存放通道的通道，而filter是存放包含区块头过滤任务的通道。它先把filter传递给了headerFilter，这样fetcher协程就在另外一段等待了，而后将headerFilterTask传入filter，fetcher就能读到数据了，处理后，再将数据写回filter而刚好被FilterHeaders函数处理了，该函数实际运行在handleMsg()的协程中。

每个Peer都会分配一个ProtocolManager然后处理该Peer的消息，但fetcher只有一个事件处理协程，如果不创建一个filter，fetcher哪知道是谁发给它的区块头呢？过滤之后，该如何发回去呢？

// FilterHeaders extracts all the headers that were explicitly requested by the fetcher,
// returning those that should be handled differently.
// 寻找出fetcher请求的区块头
func (f *Fetcher) FilterHeaders(peer string, headers []*types.Header, time time.Time) []*types.Header {
    log.Trace("Filtering headers", "peer", peer, "headers", len(headers))

    // Send the filter channel to the fetcher
    // 任务通道
    filter := make(chan *headerFilterTask)

    select {
    // 任务通道发送到这个通道
    case f.headerFilter <- filter:
    case <-f.quit:
        return nil
    }
    // Request the filtering of the header list
    // 创建过滤任务，发送到任务通道
    select {
    case filter <- &headerFilterTask{peer: peer, headers: headers, time: time}:
    case <-f.quit:
        return nil
    }
    // Retrieve the headers remaining after filtering
    // 从任务通道，获取过滤的结果并返回
    select {
    case task := <-filter:
        return task.headers
    case <-f.quit:
        return nil
    }
}

接下来要看f.headerFilter的处理，这段代码有90行，它做了以下几件事：

1. 从f.headerFilter取出filter，然后取出过滤任务task。
2. 它把区块头分成3类：unknown这不是分是要返回给调用者的，即handleMsg(), incomplete存放还需要获取body的区块头，complete存放只包含区块头的区块。遍历所有的区块头，填到到对应的分类中，具体的判断可看18行的注释，记住宏观中将的状态转移图。
3. 把unknonw中的区块返回给handleMsg()。
4. 把 incomplete的区块头获取状态移动到fetched状态，然后触发定时器，以便去处理complete的区块。
5. 把compelete的区块加入到queued。

// fetcher.loop()
case filter := <-f.headerFilter:
    // Headers arrived from a remote peer. Extract those that were explicitly
    // requested by the fetcher, and return everything else so it's delivered
    // to other parts of the system.
    // 收到从远端节点发送的区块头，过滤出fetcher请求的
    // 从任务通道获取过滤任务
    var task *headerFilterTask
    select {
    case task = <-filter:
    case <-f.quit:
        return
    }
    headerFilterInMeter.Mark(int64(len(task.headers)))

    // Split the batch of headers into unknown ones (to return to the caller),
    // known incomplete ones (requiring body retrievals) and completed blocks.
    // unknown的不是fetcher请求的，complete放没有交易和uncle的区块，有头就够了，incomplete放
    // 还需要获取uncle和交易的区块
    unknown, incomplete, complete := []*types.Header{}, []*announce{}, []*types.Block{}
    // 遍历所有收到的header
    for _, header := range task.headers {
        hash := header.Hash()

        // Filter fetcher-requested headers from other synchronisation algorithms
        // 是正在获取的hash，并且对应请求的peer，并且未fetched，未completing，未queued
        if announce := f.fetching[hash]; announce != nil && announce.origin == task.peer && f.fetched[hash] == nil && f.completing[hash] == nil && f.queued[hash] == nil {
            // If the delivered header does not match the promised number, drop the announcer
            // 高度校验，竟然不匹配，扰乱秩序，peer肯定是坏蛋。
            if header.Number.Uint64() != announce.number {
                log.Trace("Invalid block number fetched", "peer", announce.origin, "hash", header.Hash(), "announced", announce.number, "provided", header.Number)
                f.dropPeer(announce.origin)
                f.forgetHash(hash)
                continue
            }
            // Only keep if not imported by other means
            // 本地链没有当前区块
            if f.getBlock(hash) == nil {
                announce.header = header
                announce.time = task.time

                // If the block is empty (header only), short circuit into the final import queue
                // 如果区块没有交易和uncle，加入到complete
                if header.TxHash == types.DeriveSha(types.Transactions{}) && header.UncleHash == types.CalcUncleHash([]*types.Header{}) {
                    log.Trace("Block empty, skipping body retrieval", "peer", announce.origin, "number", header.Number, "hash", header.Hash())

                    block := types.NewBlockWithHeader(header)
                    block.ReceivedAt = task.time

                    complete = append(complete, block)
                    f.completing[hash] = announce
                    continue
                }
                // Otherwise add to the list of blocks needing completion
                // 否则就是不完整的区块
                incomplete = append(incomplete, announce)
            } else {
                log.Trace("Block already imported, discarding header", "peer", announce.origin, "number", header.Number, "hash", header.Hash())
                f.forgetHash(hash)
            }
        } else {
            // Fetcher doesn't know about it, add to the return list
            // 没请求过的header
            unknown = append(unknown, header)
        }
    }
    // 把未知的区块头，再传递会filter
    headerFilterOutMeter.Mark(int64(len(unknown)))
    select {
    case filter <- &headerFilterTask{headers: unknown, time: task.time}:
    case <-f.quit:
        return
    }
    // Schedule the retrieved headers for body completion
    // 把未完整的区块加入到fetched，跳过已经在completeing中的，然后触发completeTimer定时器
    for _, announce := range incomplete {
        hash := announce.header.Hash()
        if _, ok := f.completing[hash]; ok {
            continue
        }
        f.fetched[hash] = append(f.fetched[hash], announce)
        if len(f.fetched) == 1 {
            f.rescheduleComplete(completeTimer)
        }
    }
    // Schedule the header-only blocks for import
    // 把只有头的区块入队列
    for _, block := range complete {
        if announce := f.completing[block.Hash()]; announce != nil {
            f.enqueue(announce.origin, block)
        }
    }

跟随状态图的转义，剩下的工作是fetched转移到completing，上面的流程已经触发了completeTimer定时器，超时后就会处理，流程与请求Header类似，不再赘述，此时发送的请求消息是GetBlockBodiesMsg，实际调的函数是RequestBodies。

// fetcher.loop()
case <-completeTimer.C:
    // At least one header's timer ran out, retrieve everything
    // 至少有1个header已经获取完了
    request := make(map[string][]common.Hash)

    // 遍历所有待获取body的announce
    for hash, announces := range f.fetched {
        // Pick a random peer to retrieve from, reset all others
        // 随机选一个Peer发送请求，因为可能已经有很多Peer通知它这个区块了
        announce := announces[rand.Intn(len(announces))]
        f.forgetHash(hash)

        // If the block still didn't arrive, queue for completion
        // 如果本地没有这个区块，则放入到completing，创建请求
        if f.getBlock(hash) == nil {
            request[announce.origin] = append(request[announce.origin], hash)
            f.completing[hash] = announce
        }
    }
    // Send out all block body requests
    // 发送所有的请求，获取body，依然是每个peer一个单独协程
    for peer, hashes := range request {
        log.Trace("Fetching scheduled bodies", "peer", peer, "list", hashes)

        // Create a closure of the fetch and schedule in on a new thread
        if f.completingHook != nil {
            f.completingHook(hashes)
        }
        bodyFetchMeter.Mark(int64(len(hashes)))
        go f.completing[hashes[0]].fetchBodies(hashes)
    }
    // Schedule the next fetch if blocks are still pending
    f.rescheduleComplete(completeTimer)

handleMsg()处理该消息也是干净利落，直接获取RLP格式的body，然后发送响应消息。

// handleMsg()
case msg.Code == GetBlockBodiesMsg:
    // Decode the retrieval message
    msgStream := rlp.NewStream(msg.Payload, uint64(msg.Size))
    if _, err := msgStream.List(); err != nil {
        return err
    }
    // Gather blocks until the fetch or network limits is reached
    var (
        hash   common.Hash
        bytes  int
        bodies []rlp.RawValue
    )

    // 遍历所有请求
    for bytes < softResponseLimit && len(bodies) < downloader.MaxBlockFetch {
        // Retrieve the hash of the next block
        if err := msgStream.Decode(&hash); err == rlp.EOL {
            break
        } else if err != nil {
            return errResp(ErrDecode, "msg %v: %v", msg, err)
        }
        // Retrieve the requested block body, stopping if enough was found
        // 获取body，RLP格式
        if data := pm.blockchain.GetBodyRLP(hash); len(data) != 0 {
            bodies = append(bodies, data)
            bytes += len(data)
        }
    }
    return p.SendBlockBodiesRLP(bodies)

响应消息BlockBodiesMsg的处理与处理获取header的处理原理相同，先交给fetcher过滤，然后剩下的才是downloader的。需要注意一点，响应消息里只包含交易列表和叔块列表。

// handleMsg()
case msg.Code == BlockBodiesMsg:
    // A batch of block bodies arrived to one of our previous requests
    var request blockBodiesData
    if err := msg.Decode(&request); err != nil {
        return errResp(ErrDecode, "msg %v: %v", msg, err)
    }
    // Deliver them all to the downloader for queuing
    // 传递给downloader去处理
    transactions := make([][]*types.Transaction, len(request))
    uncles := make([][]*types.Header, len(request))

    for i, body := range request {
        transactions[i] = body.Transactions
        uncles[i] = body.Uncles
    }
    // Filter out any explicitly requested bodies, deliver the rest to the downloader
    // 先让fetcher过滤去fetcher请求的body，剩下的给downloader
    filter := len(transactions) > 0 || len(uncles) > 0
    if filter {
        transactions, uncles = pm.fetcher.FilterBodies(p.id, transactions, uncles, time.Now())
    }

    // 剩下的body交给downloader
    if len(transactions) > 0 || len(uncles) > 0 || !filter {
        err := pm.downloader.DeliverBodies(p.id, transactions, uncles)
        if err != nil {
            log.Debug("Failed to deliver bodies", "err", err)
        }
    }

过滤函数的原理也与Header相同。

// FilterBodies extracts all the block bodies that were explicitly requested by
// the fetcher, returning those that should be handled differently.
// 过去出fetcher请求的body，返回它没有处理的，过程类型header的处理
func (f *Fetcher) FilterBodies(peer string, transactions [][]*types.Transaction, uncles [][]*types.Header, time time.Time) ([][]*types.Transaction, [][]*types.Header) {
    log.Trace("Filtering bodies", "peer", peer, "txs", len(transactions), "uncles", len(uncles))

    // Send the filter channel to the fetcher
    filter := make(chan *bodyFilterTask)

    select {
    case f.bodyFilter <- filter:
    case <-f.quit:
        return nil, nil
    }
    // Request the filtering of the body list
    select {
    case filter <- &bodyFilterTask{peer: peer, transactions: transactions, uncles: uncles, time: time}:
    case <-f.quit:
        return nil, nil
    }
    // Retrieve the bodies remaining after filtering
    select {
    case task := <-filter:
        return task.transactions, task.uncles
    case <-f.quit:
        return nil, nil
    }
}

实际过滤body的处理瞧一下，这和Header的处理是不同的。直接看不点：

1. 它要的区块，单独取出来存到blocks中，它不要的继续留在task中。
2. 判断是不是fetcher请求的方法：如果交易列表和叔块列表计算出的hash值与区块头中的一样，并且消息来自请求的Peer，则就是fetcher请求的。
3. 将blocks中的区块加入到queued，终结。

case filter := <-f.bodyFilter:
    // Block bodies arrived, extract any explicitly requested blocks, return the rest
    var task *bodyFilterTask
    select {
    case task = <-filter:
    case <-f.quit:
        return
    }
    bodyFilterInMeter.Mark(int64(len(task.transactions)))

    blocks := []*types.Block{}
    // 获取的每个body的txs列表和uncle列表
    // 遍历每个区块的txs列表和uncle列表，计算hash后判断是否是当前fetcher请求的body
    for i := 0; i < len(task.transactions) && i < len(task.uncles); i++ {
        // Match up a body to any possible completion request
        matched := false

        // 遍历所有保存的请求，因为tx和uncle，不知道它是属于哪个区块的，只能去遍历所有的请求，通常量不大，所以遍历没有性能影响
        for hash, announce := range f.completing {
            if f.queued[hash] == nil {
                // 把传入的每个块的hash和unclehash和它请求出去的记录进行对比，匹配则说明是fetcher请求的区块body
                txnHash := types.DeriveSha(types.Transactions(task.transactions[i]))
                uncleHash := types.CalcUncleHash(task.uncles[i])

                if txnHash == announce.header.TxHash && uncleHash == announce.header.UncleHash && announce.origin == task.peer {
                    // Mark the body matched, reassemble if still unknown
                    matched = true

                    // 如果当前链还没有这个区块，则收集这个区块，合并成新区块
                    if f.getBlock(hash) == nil {
                        block := types.NewBlockWithHeader(announce.header).WithBody(task.transactions[i], task.uncles[i])
                        block.ReceivedAt = task.time

                        blocks = append(blocks, block)
                    } else {
                        f.forgetHash(hash)
                    }
                }
            }
        }
        // 从task中移除fetcher请求的数据
        if matched {
            task.transactions = append(task.transactions[:i], task.transactions[i+1:]...)
            task.uncles = append(task.uncles[:i], task.uncles[i+1:]...)
            i--
            continue
        }
    }

    // 将剩余的数据返回
    bodyFilterOutMeter.Mark(int64(len(task.transactions)))
    select {
    case filter <- task:
    case <-f.quit:
        return
    }
    // Schedule the retrieved blocks for ordered import
    // 把收集的区块加入到队列
    for _, block := range blocks {
        if announce := f.completing[block.Hash()]; announce != nil {
            f.enqueue(announce.origin, block)
        }
    }
}

至此，fetcher获取完整区块的流程讲完了，fetcher模块中80%的代码也都贴出来了，还有2个值得看看的函数：

1. forgetHash(hash common.Hash) ：用于清空指定hash值的记/状态录信息。
2. forgetBlock(hash common.Hash)：用于从队列中移除一个区块。

最后了，再回到开始看看fetcher模块和新区块的传播流程，有没有豁然开朗。

转载自：https://lessisbetter.site/2018/08/30/ethereum-fetcher-module-and-block-propagate/

过去在学Actor模型的时候，就认为异步消息是相当的重要，在华为的时候，也深扒了一下当时产品用的消息模型，简单实用，支撑起了很多模块和业务，但也有一个缺点是和其他的框架有耦合，最近看到以太坊的事件框架，同样简单简洁，理念很适合初步接触事件框架的同学，写文介绍一下。

以太坊的事件框架是一个单独的基础模块，存在于目录go-ethereum/event中，它有2中独立的事件框架实现，老点的叫TypeMux，已经基本弃用，新的叫Feed，当前正在广泛使用。

TypeMux和Feed还只是简单的事件框架，与Kafka、RocketMQ等消息系统相比，是非常的传统和简单，但是TypeMux和Feed的简单简洁，已经很好的支撑以太坊的上层模块，这是当下最好的选择。

TypeMux和Feed各有优劣，最优秀的共同特点是，他们只依赖于Golang原始的包，完全与以太坊的其他模块隔离开来，也就是说，你完全可以把这两个事件框架用在自己的项目中。

TypeMux的特点是，你把所有的订阅塞给它就好，事件来了它自会通知你，但有可能会阻塞，通知你不是那么及时，甚至过了一段挺长的时间。

Feed的特点是，它通常不存在阻塞的情况，会及时的把事件通知给你，但需要你为每类事件都建立一个Feed，然后不同的事件去不同的Feed上订阅和发送，这其实挺烦人的，如果你用错了Feed，会导致panic。

接下来，介绍下这种简单事件框架的抽象模型，然后再回归到以太坊，介绍下TypeMux和Feed。

事件框架的抽象结构

如上图，轻量级的事件框架会把所有的被订阅的事件收集起来，然后把每个订阅者组合成一个列表，当事件框架收到某个事件的时候，就把订阅该事件的所有订阅者找出来，然后把这个事件发给他们。

它需要具有2个功能：

1. 让订阅者订阅、取消订阅某类事件。
2. 让发布者能够发布某个事件，并且把事件送到每个订阅者。

如果做成完善的消息系统，就还得考虑这些特性：可用性、吞吐量、传输延迟、有序消息、消息存储、过滤、重发，这和事件框架相比就复杂上去了，我们专注的介绍下以太坊的事件模型怎么完成上述3个功能的。

以太坊的事件模型

TypeMux是一个以太坊不太满意的事件框架，所以以太坊就搞了Feed出来，它解决了TypeMux效率低下，延迟交付的问题。接下来就先看下这个TypeMux。

TypeMux：同步事件框架

TypeMux是一个同步事件框架。它的实现和上面讲的事件框架的抽象结构是完全一样的，它维护了一个订阅表，表里维护了每个事件的订阅者列表。它的特点：

1. 采用多对多结构：多个事件对多个订阅者。
2. 采用推模式，把事件/消息推送给订阅者，就像信件一样，会被送到你的信箱，你在信箱里取信就行了。
3. 是一个同步事件框架。这也是它的缺点所在，举个例子就是：邮递员要给小红、小明送信，只有信箱里的信被小红取走后，邮递员才去给小明送信，如果小红旅游去了无法取信，邮递员就一直等在小红家，而小明一直收不到信，小明很无辜无辜啊！

看下它2个功能的实现：

1. 订阅和取消订阅。订阅通过函数TypeMux.Subscribe()，入参为要订阅的事件类型，会返回TypeMuxSubscription给订阅者，订阅者可通过此控制订阅，通过TypeMuxSubscription.Unsubscribe() 可以取消订阅。
2. 发布事件和传递事件。TypeMux.Post()，入参为事件类型，根据订阅表找出该事件的订阅者列表，遍历列表，依次向每个订阅者传递事件，如果前一个没有传递完成进入阻塞，会导致后边的订阅者不能及时收到事件。

TypeMux源码速递

TypeMux的精简组成：

// A TypeMux dispatches events to registered receivers. Receivers can be
// registered to handle events of certain type. Any operation
// called after mux is stopped will return ErrMuxClosed.
//
// The zero value is ready to use.
//
// Deprecated: use Feed
// 本质：哈希列表，每个事件的订阅者都存到对于的列表里
type TypeMux struct {
    mutex   sync.RWMutex // 锁
    subm    map[reflect.Type][]*TypeMuxSubscription // 订阅表：所有事件类型的所有订阅者
    stopped bool
}

订阅：

// Subscribe creates a subscription for events of the given types. The
// subscription's channel is closed when it is unsubscribed
// or the mux is closed.
// 订阅者只传入订阅的事件类型，然后TypeMux会返回给它一个订阅对象
func (mux *TypeMux) Subscribe(types ...interface{}) *TypeMuxSubscription {
    sub := newsub(mux)
    mux.mutex.Lock()
    defer mux.mutex.Unlock()
    if mux.stopped {
        // set the status to closed so that calling Unsubscribe after this
        // call will short circuit.
        sub.closed = true
        close(sub.postC)
    } else {
        if mux.subm == nil {
            mux.subm = make(map[reflect.Type][]*TypeMuxSubscription)
        }
        for _, t := range types {
            rtyp := reflect.TypeOf(t)
            // 在同一次订阅中，不要重复订阅同一个类型的事件
            oldsubs := mux.subm[rtyp]
            if find(oldsubs, sub) != -1 {
                panic(fmt.Sprintf("event: duplicate type %s in Subscribe", rtyp))
            }
            subs := make([]*TypeMuxSubscription, len(oldsubs)+1)
            copy(subs, oldsubs)
            subs[len(oldsubs)] = sub
            mux.subm[rtyp] = subs
        }
    }
    return sub
}

取消订阅:

func (s *TypeMuxSubscription) Unsubscribe() {
    s.mux.del(s)
    s.closewait()
}

发布事件和传递事件：

// Post sends an event to all receivers registered for the given type.
// It returns ErrMuxClosed if the mux has been stopped.
// 遍历map，找到所有订阅的人，向它们传递event，同一个event对象，非拷贝，运行在调用者goroutine
func (mux *TypeMux) Post(ev interface{}) error {
    event := &TypeMuxEvent{
        Time: time.Now(),
        Data: ev,
    }
    rtyp := reflect.TypeOf(ev)
    mux.mutex.RLock()
    if mux.stopped {
        mux.mutex.RUnlock()
        return ErrMuxClosed
    }
    subs := mux.subm[rtyp]
    mux.mutex.RUnlock()
    for _, sub := range subs {
        sub.deliver(event)
    }
    return nil
}

func (s *TypeMuxSubscription) deliver(event *TypeMuxEvent) {
    // Short circuit delivery if stale event
    // 不发送过早（老）的消息
    if s.created.After(event.Time) {
        return
    }
    // Otherwise deliver the event
    s.postMu.RLock()
    defer s.postMu.RUnlock()

    select {
    case s.postC <- event:
    case <-s.closing:
    }
}

我上面指出了发送事件可能阻塞，阻塞在哪？关键就在下面这里：创建TypeMuxSubscription时，通道使用的是无缓存通道，读写是同步的，这里注定了TypeMux是一个同步事件框架，这是以太坊改用Feed的最大原因。

func newsub(mux *TypeMux) *TypeMuxSubscription {
    c := make(chan *TypeMuxEvent) // 无缓冲通道，同步读写
    return &TypeMuxSubscription{
        mux:     mux,
        created: time.Now(),
        readC:   c,
        postC:   c,
        closing: make(chan struct{}),
    }
}

Feed：流式框架

Feed是一个流式事件框架。上文强调了TypeMux是一个同步框架，也正是因为此以太坊丢弃了它，难道Feed就是一个异步框架？不一定是的，这取决于订阅者是否采用有缓存的通道，采用有缓存的通道，则Feed就是异步的，采用无缓存的通道，Feed就是同步的，把同步还是异步的选择交给使用者。

本节强调Feed的流式特点。事件本质是一个数据，连续不断的事件就组成了一个数据流，这些数据流不停的流向它的订阅者那里，并且不会阻塞在任何一个订阅者那里。

举几个不是十分恰当的例子。

1. 公司要放中秋节，HR给所有同事都发了一封邮件，有些同事读了，有些同事没读，要到国庆节了HR又给所有同事发了一封邮件，这些邮件又进入到每个人的邮箱，不会因为任何一个人没有读邮件，导致剩下的同事收不到邮件。
2. 你在朋友圈给朋友旅行的照片点了个赞，每当你们共同朋友点赞或者评论的时候，你都会收到提醒，无论你看没看这些提醒，这些提醒都会不断的发过来。
3. 你微博关注了苍井空，苍井空发了个搞笑的视频，你刷微博的时候就收到了，但也有很多人根本没刷微博，你不会因为别人没有刷，你就收不到苍井空的动态。

Feed和TypeMux相同的是，它们都是推模式，不同的是Feed是异步的，如果有些订阅者阻塞了，没关系，它会继续向后面的订阅者发送事件/消息。

Feed是一个一对多的事件流框架。每个类型的事件都需要一个与之对应的Feed，订阅者通过这个Feed进行订阅事件，发布者通过这个Feed发布事件。
看下Feed是如何实现2个功能的：

1. 订阅和取消订阅：Feed.Subscribe()，入参是一个通道，通常是有缓冲的，就算是无缓存也不会造成Feed阻塞，Feed会校验这个通道的类型和本Feed管理的事件类型是否一致，然后把通道保存下来，返回给订阅者一个Subscription，可以通过它取消订阅和读取通道错误。
2. 发布事件和传递事件。Feed.Send()入参是一个事件，加锁确保本类型事件只有一个发送协程正在进行，然后校验事件类型是否匹配，Feed会尝试给每个订阅者发送事件，如果订阅者阻塞，Feed就继续尝试给下一个订阅者发送，直到给每个订阅者发送事件，返回发送该事件的数量。

Feed源码速递

Feed定义：

// Feed implements one-to-many subscriptions where the carrier of events is a channel.
// Values sent to a Feed are delivered to all subscribed channels simultaneously.
//
// Feeds can only be used with a single type. The type is determined by the first Send or
// Subscribe operation. Subsequent calls to these methods panic if the type does not
// match.
//
// The zero value is ready to use.
// 一对多的事件订阅管理：每个feed对象，当别人调用send的时候，会发送给所有订阅者
// 每种事件类型都有一个自己的feed，一个feed内订阅的是同一种类型的事件，得用某个事件的feed才能订阅该事件
type Feed struct {
    once      sync.Once        // ensures that init only runs once
    sendLock  chan struct{}    // sendLock has a one-element buffer and is empty when held.It protects sendCases. 这个锁确保了只有一个协程在使用go routine
    removeSub chan interface{} // interrupts Send
    sendCases caseList         // the active set of select cases used by Send，订阅的channel列表，这些channel是活跃的

    // The inbox holds newly subscribed channels until they are added to sendCases.
    mu     sync.Mutex
    inbox  caseList // 不活跃的在这里
    etype  reflect.Type
    closed bool
}

订阅事件：

// Subscribe adds a channel to the feed. Future sends will be delivered on the channel
// until the subscription is canceled. All channels added must have the same element type.
//
// The channel should have ample buffer space to avoid blocking other subscribers.
// Slow subscribers are not dropped.
// 订阅者传入接收事件的通道，feed将通道保存为case，然后返回给订阅者订阅对象
func (f *Feed) Subscribe(channel interface{}) Subscription {
    f.once.Do(f.init)

    // 通道和通道类型检查
    chanval := reflect.ValueOf(channel)
    chantyp := chanval.Type()
    if chantyp.Kind() != reflect.Chan || chantyp.ChanDir()&reflect.SendDir == 0 {
        panic(errBadChannel)
    }
    sub := &feedSub{feed: f, channel: chanval, err: make(chan error, 1)}

    f.mu.Lock()
    defer f.mu.Unlock()
    if !f.typecheck(chantyp.Elem()) {
        panic(feedTypeError{op: "Subscribe", got: chantyp, want: reflect.ChanOf(reflect.SendDir, f.etype)})
    }

    // 把通道保存到case
    // Add the select case to the inbox.
    // The next Send will add it to f.sendCases.
    cas := reflect.SelectCase{Dir: reflect.SelectSend, Chan: chanval}
    f.inbox = append(f.inbox, cas)
    return sub
}

发送和传递事件：这个发送是比较绕一点的，要想真正掌握其中的运行，最好写个小程序练习下。

// Send delivers to all subscribed channels simultaneously.
// It returns the number of subscribers that the value was sent to.
// 同时向所有的订阅者发送事件，返回订阅者的数量
func (f *Feed) Send(value interface{}) (nsent int) {
    rvalue := reflect.ValueOf(value)

    f.once.Do(f.init)
    <-f.sendLock // 获取发送锁

    // Add new cases from the inbox after taking the send lock.
    // 从inbox加入到sendCases，不能订阅的时候直接加入到sendCases，因为可能其他协程在调用发送
    f.mu.Lock()
    f.sendCases = append(f.sendCases, f.inbox...)
    f.inbox = nil

    // 类型检查：如果该feed不是要发送的值的类型，释放锁，并且执行panic
    if !f.typecheck(rvalue.Type()) {
        f.sendLock <- struct{}{}
        panic(feedTypeError{op: "Send", got: rvalue.Type(), want: f.etype})
    }
    f.mu.Unlock()

    // Set the sent value on all channels.
    // 把发送的值关联到每个case/channel，每一个事件都有一个feed，所以这里全是同一个事件的
    for i := firstSubSendCase; i < len(f.sendCases); i++ {
        f.sendCases[i].Send = rvalue
    }

    // Send until all channels except removeSub have been chosen. 'cases' tracks a prefix
    // of sendCases. When a send succeeds, the corresponding case moves to the end of
    // 'cases' and it shrinks by one element.
    // 所有case仍然保留在sendCases，只是用过的会移动到最后面
    cases := f.sendCases
    for {
        // Fast path: try sending without blocking before adding to the select set.
        // This should usually succeed if subscribers are fast enough and have free
        // buffer space.
        // 使用非阻塞式发送，如果不能发送就及时返回
        for i := firstSubSendCase; i < len(cases); i++ {
            // 如果发送成功，把这个case移动到末尾，所以i这个位置就是没处理过的，然后大小减1
            if cases[i].Chan.TrySend(rvalue) {
                nsent++
                cases = cases.deactivate(i)
                i--
            }
        }

        // 如果这个地方成立，代表所有订阅者都不阻塞，都发送完了
        if len(cases) == firstSubSendCase {
            break
        }

        // Select on all the receivers, waiting for them to unblock.
        // 返回一个可用的，直到不阻塞。
        chosen, recv, _ := reflect.Select(cases)
        if chosen == 0 /* <-f.removeSub */ {
            // 这个接收方要删除了，删除并缩小sendCases
            index := f.sendCases.find(recv.Interface())
            f.sendCases = f.sendCases.delete(index)
            if index >= 0 && index < len(cases) {
                // Shrink 'cases' too because the removed case was still active.
                cases = f.sendCases[:len(cases)-1]
            }
        } else {
            // reflect已经确保数据已经发送，无需再尝试发送
            cases = cases.deactivate(chosen)
            nsent++
        }
    }

    // 把sendCases中的send都标记为空
    // Forget about the sent value and hand off the send lock.
    for i := firstSubSendCase; i < len(f.sendCases); i++ {
        f.sendCases[i].Send = reflect.Value{}
    }
    f.sendLock <- struct{}{}
    return nsent
}

转载自：https://lessisbetter.site/2018/10/18/ethereum-code-event-framework/