第六节 db.Batch()实现分析

现在对Batch()方法稍作分析，在DB定义的那一节中我们可以看到，一个DB对象拥有一个batch对象，该对象是全局的。当我们使用Batch()方法时，内部会对将传递进去的fn缓存在calls中。

其内部也是调用了Update，只不过是在Update内部遍历之前缓存的calls。

有两种情况会触发调用Update。

  1. 第一种情况是到达了MaxBatchDelay时间，就会触发Update
  2. 第二种情况是len(db.batch.calls) >= db.MaxBatchSize，即缓存的calls个数大于等于MaxBatchSize时，也会触发Update。

Batch的本质是： 将每次写、每次刷盘的操作转变成了多次写、一次刷盘，从而提升性能。

  1. // Batch calls fn as part of a batch. It behaves similar to Update,
  2. // except:
  3. //
  4. // 1. concurrent Batch calls can be combined into a single Bolt
  5. // transaction.
  6. //
  7. // 2. the function passed to Batch may be called multiple times,
  8. // regardless of whether it returns error or not.
  9. //
  10. // This means that Batch function side effects must be idempotent and
  11. // take permanent effect only after a successful return is seen in
  12. // caller.
  13. // 幂等
  14. // The maximum batch size and delay can be adjusted with DB.MaxBatchSize
  15. // and DB.MaxBatchDelay, respectively.
  16. //
  17. // Batch is only useful when there are multiple goroutines calling it.
  18. func (db *DB) Batch(fn func(*Tx) error) error {
  19.     errCh := make(chan error, 1)
  21.     db.batchMu.Lock()
  22.     if (db.batch == nil) || (db.batch != nil && len(db.batch.calls) >= db.MaxBatchSize) {
  23.         // There is no existing batch, or the existing batch is full; start a new one.
  24.         db.batch = &batch{
  25.             db: db,
  26.         }
  27.         db.batch.timer = time.AfterFunc(db.MaxBatchDelay, db.batch.trigger)
  28.     }
  29.     db.batch.calls = append(db.batch.calls, call{fn: fn, err: errCh})
  30.     if len(db.batch.calls) >= db.MaxBatchSize {
  31.         // wake up batch, it's ready to run
  32.         go db.batch.trigger()
  33.     }
  34.     db.batchMu.Unlock()
  36.     err := <-errCh
  37.     if err == trySolo {
  38.         err = db.Update(fn)
  39.     }
  40.     return err
  41. }
  43. type call struct {
  44.     fn  func(*Tx) error
  45.     err chan<- error
  46. }
  48. type batch struct {
  49.     db    *DB
  50.     timer *time.Timer
  51.     start sync.Once
  52.     calls []call
  53. }
  55. // trigger runs the batch if it hasn't already been run.
  56. func (b *batch) trigger() {
  57.     b.start.Do(b.run)
  58. }
  60. // run performs the transactions in the batch and communicates results
  61. // back to DB.Batch.
  62. func (b *batch) run() {
  63.     b.db.batchMu.Lock()
  64.     b.timer.Stop()
  65.     // Make sure no new work is added to this batch, but don't break
  66.     // other batches.
  67.     if b.db.batch == b {
  68.         b.db.batch = nil
  69.     }
  70.     b.db.batchMu.Unlock()
  72. retry:
  73.     // 内部多次调用Update，最后一次Commit刷盘，提升性能
  74.     for len(b.calls) > 0 {
  75.         var failIdx = -1
  76.         err := b.db.Update(func(tx *Tx) error {
  77.             遍历calls中的函数c，多次调用，最后一次提交刷盘
  78.             for i, c := range b.calls {
  79.                 // safelyCall里面捕获了panic
  80.                 if err := safelyCall(c.fn, tx); err != nil {
  81.                     failIdx = i
  82.                     //只要又失败，事务就不提交
  83.                     return err
  84.                 }
  85.             }
  86.             return nil
  87.         })
  89.         if failIdx >= 0 {
  90.             // take the failing transaction out of the batch. it's
  91.             // safe to shorten b.calls here because db.batch no longer
  92.             // points to us, and we hold the mutex anyway.
  93.             c := b.calls[failIdx]
  94.             //这儿只是把失败的事务给踢出去了，然后其他的事务会重新执行
  95.             b.calls[failIdx], b.calls = b.calls[len(b.calls)-1], b.calls[:len(b.calls)-1]
  96.             // tell the submitter re-run it solo, continue with the rest of the batch
  97.             c.err <- trySolo
  98.             continue retry
  99.         }
  101.         // pass success, or bolt internal errors, to all callers
  102.         for _, c := range b.calls {
  103.             c.err <- err
  104.         }
  105.         break retry
  106.     }
  107. }
  109. // trySolo is a special sentinel error value used for signaling that a
  110. // transaction function should be re-run. It should never be seen by
  111. // callers.
  112. var trySolo = errors.New("batch function returned an error and should be re-run solo")
  114. type panicked struct {
  115.     reason interface{}
  116. }
  118. func (p panicked) Error() string {
  119.     if err, ok := p.reason.(error); ok {
  120.         return err.Error()
  121.     }
  122.     return fmt.Sprintf("panic: %v", p.reason)
  123. }
  125. func safelyCall(fn func(*Tx) error, tx *Tx) (err error) {
  126.     defer func() {
  127.         if p := recover(); p != nil {
  128.             err = panicked{p}
  129.         }
  130.     }()
  131.     return fn(tx)
  132. }