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

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

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

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

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

第一种情况是到达了MaxBatchDelay时间，就会触发Update
第二种情况是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. }