未验证 提交 d6e66a2d 编写于 作者: J Jerry Tao 提交者: GitHub

[ISSUE #568] Update lastPullTime use atomic.Value as same with lastConsumeTime...

[ISSUE #568] Update lastPullTime use atomic.Value as same with lastConsumeTime and lastLockTime (#613)
上级 05929dac
......@@ -50,7 +50,7 @@ type processQueue struct {
consumeLock sync.Mutex
consumingMsgOrderlyTreeMap *treemap.Map
dropped *uatomic.Bool
lastPullTime time.Time
lastPullTime atomic.Value
lastConsumeTime atomic.Value
locked *uatomic.Bool
lastLockTime atomic.Value
......@@ -69,9 +69,12 @@ func newProcessQueue(order bool) *processQueue {
lastLockTime := atomic.Value{}
lastLockTime.Store(time.Now())
lastPullTime := atomic.Value{}
lastPullTime.Store(time.Now())
pq := &processQueue{
msgCache: treemap.NewWith(utils.Int64Comparator),
lastPullTime: time.Now(),
lastPullTime: lastPullTime,
lastConsumeTime: lastConsumeTime,
lastLockTime: lastLockTime,
msgCh: make(chan []*primitive.MessageExt, 32),
......@@ -157,6 +160,14 @@ func (pq *processQueue) LastLockTime() time.Time {
return pq.lastLockTime.Load().(time.Time)
}
func (pq *processQueue) LastPullTime() time.Time {
return pq.lastPullTime.Load().(time.Time)
}
func (pq *processQueue) UpdateLastPullTime() {
pq.lastPullTime.Store(time.Now())
}
func (pq *processQueue) makeMessageToCosumeAgain(messages ...*primitive.MessageExt) {
pq.mutex.Lock()
for _, msg := range messages {
......@@ -199,7 +210,7 @@ func (pq *processQueue) isLockExpired() bool {
}
func (pq *processQueue) isPullExpired() bool {
return time.Now().Sub(pq.lastPullTime) > _PullMaxIdleTime
return time.Now().Sub(pq.LastPullTime()) > _PullMaxIdleTime
}
func (pq *processQueue) cleanExpiredMsg(consumer defaultConsumer) {
......@@ -360,7 +371,7 @@ func (pq *processQueue) currentInfo() internal.ProcessQueueInfo {
TryUnlockTimes: pq.tryUnlockTimes,
LastLockTimestamp: pq.LastLockTime().UnixNano() / int64(time.Millisecond),
Dropped: pq.dropped.Load(),
LastPullTimestamp: pq.lastPullTime.UnixNano() / int64(time.Millisecond),
LastPullTimestamp: pq.LastPullTime().UnixNano() / int64(time.Millisecond),
LastConsumeTimestamp: pq.LastConsumeTime().UnixNano() / int64(time.Millisecond),
}
......
......@@ -562,7 +562,7 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
}
// reset time
sleepTime = pc.option.PullInterval
pq.lastPullTime = time.Now()
pq.lastPullTime.Store(time.Now())
err := pc.makeSureStateOK()
if err != nil {
rlog.Warning("consumer state error", map[string]interface{}{
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册