@@ -35,11 +35,11 @@ type RocketQueue struct {
3535 producerOnce sync.Once
3636}
3737
38- func (rq * RocketQueue ) initProducer (ctx context.Context ) error {
38+ func (m * RocketQueue ) initProducer (ctx context.Context ) error {
3939 var err error
40- rq .producerOnce .Do (
40+ m .producerOnce .Do (
4141 func () {
42- err = rq .Producer .Start ()
42+ err = m .Producer .Start ()
4343 if err != nil {
4444 logger .Fatal (ctx , "RocketQueue:Producer:Start" , err .Error ())
4545 return
@@ -48,11 +48,11 @@ func (rq *RocketQueue) initProducer(ctx context.Context) error {
4848 return err
4949}
5050
51- func (rq * RocketQueue ) initConsumer (ctx context.Context , topic , messageTag string , num int ) error {
51+ func (m * RocketQueue ) initConsumer (ctx context.Context , topic , messageTag string , num int ) error {
5252 var err error
53- rq .consumerOnce .Do (
53+ m .consumerOnce .Do (
5454 func () {
55- rq .consumerMessageChan = make (chan * primitive.MessageExt , num )
55+ m .consumerMessageChan = make (chan * primitive.MessageExt , num )
5656
5757 var selector consumer.MessageSelector
5858 if len (messageTag ) > 0 {
@@ -61,10 +61,10 @@ func (rq *RocketQueue) initConsumer(ctx context.Context, topic, messageTag strin
6161 Expression : messageTag ,
6262 }
6363 }
64- err = rq .Consumer .Subscribe (topic , selector , func (ctx context.Context , messages ... * primitive.MessageExt ) (consumer.ConsumeResult , error ) {
64+ err = m .Consumer .Subscribe (topic , selector , func (ctx context.Context , messages ... * primitive.MessageExt ) (consumer.ConsumeResult , error ) {
6565 // 取到的消息放入管道，交给下游处理
6666 for _ , msg := range messages {
67- rq .consumerMessageChan <- msg
67+ m .consumerMessageChan <- msg
6868 }
6969
7070 return consumer .ConsumeSuccess , nil
@@ -74,7 +74,7 @@ func (rq *RocketQueue) initConsumer(ctx context.Context, topic, messageTag strin
7474 return
7575 }
7676
77- err = rq .Consumer .Start ()
77+ err = m .Consumer .Start ()
7878 if err != nil {
7979 logger .Fatal (ctx , "RocketQueue:Start" , err .Error ())
8080 return
@@ -95,8 +95,8 @@ func (rq *RocketQueue) initConsumer(ctx context.Context, topic, messageTag strin
9595 sig := <- c //blocked
9696 switch sig {
9797 case syscall .SIGINT , syscall .SIGTERM :
98- close (rq .consumerMessageChan )
99- err = rq .Consumer .Shutdown ()
98+ close (m .consumerMessageChan )
99+ err = m .Consumer .Shutdown ()
100100 if err != nil {
101101 logger .Error (ctx , "Shutdown.Failure" , err .Error ())
102102 return
@@ -116,7 +116,7 @@ func (rq *RocketQueue) initConsumer(ctx context.Context, topic, messageTag strin
116116 return nil
117117}
118118
119- //new实例
119+ // new实例
120120func newRocketQueue (diName string ) queue.Queue {
121121 m := new (RocketQueue )
122122 client := rkmq .GetRocketMq (diName )
@@ -127,7 +127,9 @@ func newRocketQueue(diName string) queue.Queue {
127127 return m
128128}
129129
130- //单例模式
130+ // GetRocketQueue
131+ //
132+ // 单例模式
131133func GetRocketQueue (diName string ) queue.Queue {
132134 key := diName
133135 mu .RLock ()
@@ -145,18 +147,15 @@ func GetRocketQueue(diName string) queue.Queue {
145147 return q
146148}
147149
148- /**
149- * 队列消息入队
150- * args[0] instanceId
151- */
150+ // Enqueue 队列消息入队
151+ //
152+ // args[0] instanceId
152153func (m * RocketQueue ) Enqueue (ctx context.Context , key string , message string , args ... interface {}) (bool , error ) {
153-
154154 err := m .initProducer (ctx )
155155 if err != nil {
156156 return false , err
157157 }
158-
159- _ , _ , messageTag := getOption (args ... )
158+ _ , _ , messageTag , timeLevel := getOption (args ... )
160159 log .Printf ("messageTag: %v" , messageTag )
161160 if len (messageTag ) > 0 {
162161 tags := strings .Split (messageTag , "||" )
@@ -167,6 +166,10 @@ func (m *RocketQueue) Enqueue(ctx context.Context, key string, message string, a
167166 Body : []byte (message ),
168167 }
169168 msg .WithTag (tag )
169+ // https://rocketmq.apache.org/docs/4.x/producer/04message3/
170+ if timeLevel > 0 && timeLevel <= 18 {
171+ msg .WithDelayTimeLevel (timeLevel )
172+ }
170173 log .Printf ("send for tag: %v" , tag )
171174 res , err := m .Producer .SendSync (context .Background (), msg )
172175 if err != nil {
@@ -180,6 +183,10 @@ func (m *RocketQueue) Enqueue(ctx context.Context, key string, message string, a
180183 Topic : key ,
181184 Body : []byte (message ),
182185 }
186+ // https://rocketmq.apache.org/docs/4.x/producer/04message3/
187+ if timeLevel > 0 && timeLevel <= 18 {
188+ msg .WithDelayTimeLevel (timeLevel )
189+ }
183190 res , err := m .Producer .SendSync (ctx , msg )
184191 if err != nil {
185192 return false , err
@@ -191,13 +198,13 @@ func (m *RocketQueue) Enqueue(ctx context.Context, key string, message string, a
191198 return true , nil
192199}
193200
194- /**
195- * 队列消息出队
196- * param 第二个参数是队列名称，args[0]是instanceId，args[1]是groupId，目前只有rocketmq需要groupId
197- * return 第一个参数是消息 第二个参数是aliyunmq的ReceiptHandle命名为token，通过token确定消息是否从队列删除，第三个参数为消费次数
198- */
201+ // Dequeue 队列消息出队
202+ //
203+ // param 第二个参数是队列名称，args[0]是instanceId，args[1]是groupId，目前只有rocketmq需要groupId
204+ //
205+ // return 第一个参数是消息 第二个参数是aliyunmq的ReceiptHandle命名为token，通过token确定消息是否从队列删除，第三个参数为消费次数
199206func (m * RocketQueue ) Dequeue (ctx context.Context , key string , args ... interface {}) (message string , tag string , token string , dequeueCount int64 , err error ) {
200- _ , _ , messageTag := getOption (args ... )
207+ _ , _ , messageTag , _ := getOption (args ... )
201208
202209 err = m .initConsumer (ctx , key , messageTag , 5 )
203210 if err != nil {
@@ -213,11 +220,9 @@ func (m *RocketQueue) Dequeue(ctx context.Context, key string, args ...interface
213220 }
214221}
215222
216- /**
217- * 队列消息批量入队
218- * args[0] instanceId
219- * 注：rocket其实没有批量函数，所以循环调用publishMsg方法
220- */
223+ // BatchEnqueue 队列消息批量入队
224+ // args[0] instanceId
225+ // 注：rocket其实没有批量函数，所以循环调用publishMsg方法
221226func (m * RocketQueue ) BatchEnqueue (ctx context.Context , key string , messageList []string , args ... interface {}) (bool , error ) {
222227 if len (messageList ) == 0 {
223228 return false , errors .New ("messageList is empty" )
@@ -233,21 +238,20 @@ func (m *RocketQueue) BatchEnqueue(ctx context.Context, key string, messageList
233238 return true , nil
234239}
235240
236- /**
237- * 确认消息接收
238- * args[0]是instanceId，args[1]是groupId，args[2]是messageTag
239- */
241+ // AckMsg 确认消息接收
242+ // args[0]是instanceId，args[1]是groupId，args[2]是messageTag, args[2]是delayTimeLevel
240243func (m * RocketQueue ) AckMsg (ctx context.Context , key string , token string , args ... interface {}) (bool , error ) {
241244 return true , nil
242245}
243246
244- // 缺省参数统一获取
245- // args[0]是instanceId，args[1]是groupId，args[2]是messageTag
246- func getOption (args ... interface {}) (instanceId , groupId , messageTag string ) {
247+ // getOption 缺省参数统一获取
248+ //
249+ // args[0]是instanceId，args[1]是groupId，args[2]是messageTag, args[3]是delayTimeLevel
250+ func getOption (args ... interface {}) (instanceId , groupId , messageTag string , delayTimeLevel int ) {
247251 instanceId = ""
248252 groupId = ""
249253 messageTag = ""
250-
254+ delayTimeLevel = 0
251255 l := len (args )
252256 if l > 0 {
253257 tempInstance , ok := args [0 ].(string )
@@ -266,6 +270,12 @@ func getOption(args ...interface{}) (instanceId, groupId, messageTag string) {
266270 messageTag = tempTag
267271 }
268272 }
273+ if l > 3 {
274+ tempDelayTimeLevel , ok := args [3 ].(int )
275+ if ok {
276+ delayTimeLevel = tempDelayTimeLevel
277+ }
278+ }
269279 }
270280 return
271281}
0 commit comments