开启Geth node 和console

./build/bin/geth --datadir walker/ --miner.etherbase  0x106cdbb5029d186ae6f659b52fb83179fc153e6a --mine --miner.threads 1   --rpc --rpcvhosts "*" --rpcaddr 0.0.0.0 --rpcport 8545 --rpccorsdomain "*" --rpcapi "db,eth,net,web3,personal,debug" 

在console转账

//解锁account
personal.unlockAccount("0x106cdbb5029d186ae6f659b52fb83179fc153e6a","walker")
true
//转账交易
eth.sendTransaction({from:'0x106cdbb5029d186ae6f659b52fb83179fc153e6a',to:'0x3b7bfcffd7a8c1047055edc58e2efa6eb589184f',value:web3.toWei(100,"ether")})
"0x21e5e3e9968d4bcf99402923b168bc172cadcc0fa4b27885d07ff52eac5a4f2b"

node处理流程

通过 console.Interactive() -> c.Evaluate(input)
-> simplechain/internal/ethapi.go

func (s *PublicTransactionPoolAPI) SendTransaction(ctx context.Context, args SendTxArgs) (common.Hash, error) {
   // Look up the wallet containing the requested signer
   account := accounts.Account{Address: args.From}

   wallet, err := s.b.AccountManager().Find(account)
   if err != nil {
       return common.Hash{}, err
   }

   if args.Nonce == nil {
       // Hold the addresse's mutex around signing to prevent concurrent assignment of
       // the same nonce to multiple accounts.
       s.nonceLock.LockAddr(args.From)
       defer s.nonceLock.UnlockAddr(args.From)
   }

   // Set some sanity defaults and terminate on failure
   if err := args.setDefaults(ctx, s.b); err != nil {
       return common.Hash{}, err
   }
   // Assemble the transaction and sign with the wallet
   tx := args.toTransaction()

   var chainID *big.Int
   if config := s.b.ChainConfig(); config.IsEIP155(s.b.CurrentBlock().Number()) {
       chainID = config.ChainID
   }
   signed, err := wallet.SignTx(account, tx, chainID)
   if err != nil {
       return common.Hash{}, err
   }
   return submitTransaction(ctx, s.b, signed)
}

// submitTransaction is a helper function that submits tx to txPool and logs a message.
func submitTransaction(ctx context.Context, b Backend, tx *types.Transaction) (common.Hash, error) {
   if err := b.SendTx(ctx, tx); err != nil {
       return common.Hash{}, err
   }
   if tx.To() == nil {
       signer := types.MakeSigner(b.ChainConfig(), b.CurrentBlock().Number())
       from, err := types.Sender(signer, tx)
       if err != nil {
           return common.Hash{}, err
       }
       addr := crypto.CreateAddress(from, tx.Nonce())
       log.Info("Submitted contract creation", "fullhash", tx.Hash().Hex(), "contract", addr.Hex())
   } else {
       log.Info("Submitted transaction", "fullhash", tx.Hash().Hex(), "recipient", tx.To())
   }
   return tx.Hash(), nil
}

sendTrx() -> simplechain/core/tx_pool.go

// addTx enqueues a single transaction into the pool if it is valid.
func (pool *TxPool) addTx(tx *types.Transaction, local bool) error {
    pool.mu.Lock()
    defer pool.mu.Unlock()

    // Try to inject the transaction and update any state
    replace, err := pool.add(tx, local)
    if err != nil {
        return err
    }
    // If we added a new transaction, run promotion checks and return
    if !replace {
        from, _ := types.Sender(pool.signer, tx) // already validated
        pool.promoteExecutables([]common.Address{from})
    }
    return nil
}

以上代码流程比较清晰。至此node已经将交易放入交易池。

向交易池添加交易

1、添加交易TxPool.add(), add()方法用于将本地或远端的交易加入到交易池，这个方法的基本逻辑是：
1）检查交易是否收到过，重复接受的交易直接丢弃；
2）验证交易是否有效；
3）如果交易池满了，待插入的交易的价值比交易池中任意一个都低，则直接丢弃；
4）如果待插入的交易序号在pending列表中已经存在，且待插入的交易价值大于或等于原交易的110%，则替换原交易；
5）如果待插入的交易序号在pending列表中没有，则直接放入queue列表。如果对应的序号已经有交易了，则如果新交易的价值大于或等于原交易的110%，替换原交易；

注意：这里pool.config.GlobalSlots为所有可执行交易的总数，即pending列表总数，默认4096；pool.config.GlobalQueue为不可执行交易总数，即queue列表总数，默认1024；

// add validates a transaction and inserts it into the non-executable queue for
// later pending promotion and execution. If the transaction is a replacement for
// an already pending or queued one, it overwrites the previous and returns this
// so outer code doesn't uselessly call promote.
//
// If a newly added transaction is marked as local, its sending account will be
// whitelisted, preventing any associated transaction from being dropped out of
// the pool due to pricing constraints.
func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
    // If the transaction is already known, discard it
    hash := tx.Hash()
    if pool.all.Get(hash) != nil {
        log.Trace("Discarding already known transaction", "hash", hash)
        return false, fmt.Errorf("known transaction: %x", hash)
    }
    // If the transaction fails basic validation, discard it
    if err := pool.validateTx(tx, local); err != nil {
        log.Trace("Discarding invalid transaction", "hash", hash, "err", err)
        invalidTxCounter.Inc(1)
        return false, err
    }
    // If the transaction pool is full, discard underpriced transactions
    if uint64(pool.all.Count()) >= pool.config.GlobalSlots+pool.config.GlobalQueue {
        // If the new transaction is underpriced, don't accept it
        if !local && pool.priced.Underpriced(tx, pool.locals) {
            log.Trace("Discarding underpriced transaction", "hash", hash, "price", tx.GasPrice())
            underpricedTxCounter.Inc(1)
            return false, ErrUnderpriced
        }
        // New transaction is better than our worse ones, make room for it
        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)
        }
    }
    // If the transaction is replacing an already pending one, do directly
    //如果插入pending已有的交易，必须交易价值大于或等于原交易的110%，方可替换
    from, _ := types.Sender(pool.signer, tx) // already validated
    if list := pool.pending[from]; list != nil && list.Overlaps(tx) {
        // Nonce already pending, check if required price bump is met
        inserted, old := list.Add(tx, pool.config.PriceBump)
        if !inserted {
            pendingDiscardCounter.Inc(1)
            return false, ErrReplaceUnderpriced
        }
        // New transaction is better, replace old one
        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)

        log.Trace("Pooled new executable transaction", "hash", hash, "from", from, "to", tx.To())

        // We've directly injected a replacement transaction, notify subsystems
        //通知其他对交易池增加交易感兴趣的子系统：广播和矿工
        go pool.txFeed.Send(NewTxsEvent{types.Transactions{tx}})

        return old != nil, nil
    }
    // New transaction isn't replacing a pending one, push into queue
    replace, err := pool.enqueueTx(hash, tx)
    if err != nil {
        return false, err
    }
    // Mark local addresses and journal local transactions
    if local {
        if !pool.locals.contains(from) {
            log.Info("Setting new local account", "address", from)
            pool.locals.add(from)
        }
    }
    pool.journalTx(from, tx)

    log.Trace("Pooled new future transaction", "hash", hash, "from", from, "to", tx.To())
    return replace, nil
}

TxPool.add()的调用时机：

1）命令行发送交易

EthAPIBackend.SendTx
      ==> TxPool.AddLocal
          ==>Txpool.addTx
              ==>Txpool.add 

2）收到远程节点广播的交易时

AddRemotes
    ==> TxPool.addRemotes
        ==> TxPool.addTxs
            ==> TxPool.addTxLocked
                ==> TxPool.add

3）交易池重新整理的过程

TxPool.reset
    ==> TxPool.addTxLocked
         ==> TxPool.add

这里的addLocal和addRemote的区别，其中有第二个参数来设定该交易是local还是remote，local的交易在打包时有优先权，在删除时有豁免权，还会以文件的形式保存在磁盘上。

func (pool *TxPool) AddLocal(tx *types.Transaction) error {
    return pool.addTx(tx, !pool.config.NoLocals)
}

func (pool *TxPool) AddRemote(tx *types.Transaction) error {
    return pool.addTx(tx, false)
}

交易加入queue列表：TxPool.enqueueTx

主要流程：
1）将交易插入queue中，如果待插入的交易序号在queue列表中已经有一个交易，那么待插入的交易价值大于原交易价值的110%，则替换原交易；
2）如果新交易替换成功，则从all列表中删除这个被替换的交易
3）更新all列表

// enqueueTx inserts a new transaction into the non-executable transaction queue.
//
// Note, this method assumes the pool lock is held!
func (pool *TxPool) enqueueTx(hash common.Hash, tx *types.Transaction) (bool, error) {
    // Try to insert the transaction into the future queue
    from, _ := types.Sender(pool.signer, tx) // already validated
    if pool.queue[from] == nil {
        pool.queue[from] = newTxList(false)
    }
    inserted, old := pool.queue[from].Add(tx, pool.config.PriceBump)
    if !inserted {
        // An older transaction was better, discard this
        queuedDiscardCounter.Inc(1)
        return false, ErrReplaceUnderpriced
    }
    // Discard any previous transaction and mark this
    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
}