-
Notifications
You must be signed in to change notification settings - Fork 0
/
store.go
144 lines (125 loc) · 3.47 KB
/
store.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
package rocketmq
import (
"errors"
"fmt"
"sync"
"sync/atomic"
)
const (
MemoryFirstThenStore = 0
ReadFromMemory = 1
ReadFromStore = 2
)
type OffsetStore interface {
//load() error
updateOffset(mq *MessageQueue, offset int64, increaseOnly bool)
readOffset(mq *MessageQueue, flag int) int64
//persistAll(mqs []MessageQueue)
//persist(mq MessageQueue)
//removeOffset(mq MessageQueue)
//cloneOffsetTable(topic string) map[MessageQueue]int64
}
type RemoteOffsetStore struct {
groupName string
mqClient *MqClient
offsetTable map[MessageQueue]int64
offsetTableLock sync.RWMutex
}
func (r *RemoteOffsetStore) readOffset(mq *MessageQueue, readType int) int64 {
switch readType {
case MemoryFirstThenStore:
case ReadFromMemory:
r.offsetTableLock.RLock()
offset, ok := r.offsetTable[*mq]
r.offsetTableLock.RUnlock()
if ok {
return offset
} else if readType == ReadFromMemory {
return -1
}
case ReadFromStore:
offset, err := r.fetchConsumeOffsetFromBroker(mq)
if err != nil {
fmt.Println(err)
return -1
}
r.updateOffset(mq, offset, false)
return offset
}
return -1
}
func (r *RemoteOffsetStore) fetchConsumeOffsetFromBroker(mq *MessageQueue) (int64, error) {
brokerAddr, _, found := r.mqClient.findBrokerAddressInSubscribe(mq.brokerName, 0, false)
if !found {
if _, err := r.mqClient.updateTopicRouteInfoFromNameServerKernel(mq.topic, false, DefaultProducer{}); err != nil {
return 0, err
}
brokerAddr, _, found = r.mqClient.findBrokerAddressInSubscribe(mq.brokerName, 0, false)
}
if found {
requestHeader := &QueryConsumerOffsetRequestHeader{}
requestHeader.Topic = mq.topic
requestHeader.QueueId = mq.queueId
requestHeader.ConsumerGroup = r.groupName
return r.mqClient.queryConsumerOffset(brokerAddr, requestHeader, 3000)
}
return 0, errors.New("fetch consumer offset error")
}
func (r *RemoteOffsetStore) persist(mq *MessageQueue) {
offset, ok := r.offsetTable[*mq]
if ok {
err := r.updateConsumeOffsetToBroker(mq, offset)
if err != nil {
fmt.Println(err)
}
}
}
type UpdateConsumerOffsetRequestHeader struct {
consumerGroup string
topic string
queueId int32
commitOffset int64
}
func (r *RemoteOffsetStore) updateConsumeOffsetToBroker(mq *MessageQueue, offset int64) error {
addr, found, _ := r.mqClient.findBrokerAddressInSubscribe(mq.brokerName, 0, false)
if !found {
if _, err := r.mqClient.updateTopicRouteInfoFromNameServerKernel(mq.topic, false, DefaultProducer{}); err != nil {
return err
}
addr, found, _ = r.mqClient.findBrokerAddressInSubscribe(mq.brokerName, 0, false)
}
if found {
requestHeader := &UpdateConsumerOffsetRequestHeader{
consumerGroup: r.groupName,
topic: mq.topic,
queueId: mq.queueId,
commitOffset: offset,
}
r.mqClient.updateConsumerOffsetOneway(addr, requestHeader, 5*1000)
return nil
}
return errors.New("not found broker")
}
func (r *RemoteOffsetStore) updateOffset(mq *MessageQueue, offset int64, increaseOnly bool) {
if mq != nil {
r.offsetTableLock.RLock()
offsetOld, ok := r.offsetTable[*mq]
r.offsetTableLock.RUnlock()
if !ok {
r.offsetTableLock.Lock()
r.offsetTable[*mq] = offset
r.offsetTableLock.Unlock()
} else {
if increaseOnly {
atomic.AddInt64(&offsetOld, offset)
r.offsetTableLock.Lock()
r.offsetTable[*mq] = offsetOld
r.offsetTableLock.Unlock()
} else {
r.offsetTableLock.Lock()
r.offsetTable[*mq] = offset
r.offsetTableLock.Unlock()
}
}
}
}