Skip to content

Commit 3bacb20

Browse files
UnderTreeTechUnderTreeTech
committed
[feat] support rocketmq pull consumer mode
1 parent 32a8fda commit 3bacb20

4 files changed

Lines changed: 131 additions & 30 deletions

File tree

‎go.mod‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,13 +7,13 @@ require (
77
github.com/Shopify/sarama v1.32.0
88
github.com/StackExchange/wmi v0.0.0-20190523213315-cbe66965904d // indirect
99
github.com/alibaba/sentinel-golang v1.0.2
10-
github.com/apache/rocketmq-client-go/v2 v2.1.1
10+
github.com/apache/rocketmq-client-go/v2 v2.1.2
1111
github.com/codahale/hdrhistogram v0.0.0-20161010025455-3a0bb77429bd // indirect
1212
github.com/gin-contrib/cors v1.3.1
1313
github.com/gin-gonic/gin v1.7.7
1414
github.com/go-ole/go-ole v1.2.4 // indirect
1515
github.com/go-playground/validator/v10 v10.4.1
16-
github.com/go-redis/redis/v8 v8.11.0
16+
github.com/go-redis/redis/v8 v8.11.5
1717
github.com/go-resty/resty/v2 v2.7.0
1818
github.com/go-sql-driver/mysql v1.6.0
1919
github.com/golang/mock v1.4.4 // indirect
@@ -25,7 +25,6 @@ require (
2525
github.com/mitchellh/mapstructure v1.4.1
2626
github.com/natefinch/lumberjack v0.0.0-20201021141957-47ffae23317c
2727
github.com/olivere/elastic/v7 v7.0.25
28-
github.com/onsi/gomega v1.15.0 // indirect
2928
github.com/opentracing/opentracing-go v1.2.0
3029
github.com/prometheus/client_golang v1.11.1
3130
github.com/qiniu/qmgo v1.0.6

‎go.sum‎

Lines changed: 19 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -18,8 +18,8 @@ github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk5
1818
github.com/alibaba/sentinel-golang v1.0.2 h1:Acopq74hOtZN4MV1v811MQ6QcqPFLDSczTrRXv9zpIg=
1919
github.com/alibaba/sentinel-golang v1.0.2/go.mod h1:QsB99f/z35D2AiMrAWwgWE85kDTkBUIkcmPrRt+61NI=
2020
github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
21-
github.com/apache/rocketmq-client-go/v2 v2.1.1 h1:WY/LkOYSQaVyV+HOqdiIgF4LE3beZ/jwdSLKZlzpabw=
22-
github.com/apache/rocketmq-client-go/v2 v2.1.1/go.mod h1:GZzExtXY9zpI6FfiVJYAhw2IXQtgnHUuWpULo7nr5lw=
21+
github.com/apache/rocketmq-client-go/v2 v2.1.2 h1:yt73olKe5N6894Dbm+ojRf/JPiP0cxfDNNffKwhpJVg=
22+
github.com/apache/rocketmq-client-go/v2 v2.1.2/go.mod h1:6I6vgxHR3hzrvn+6n/4mrhS+UTulzK/X9LB2Vk1U5gE=
2323
github.com/aws/aws-sdk-go v1.38.17/go.mod h1:hcU610XS61/+aQV88ixoOzUoG7v3b31pl2zKMmprdro=
2424
github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
2525
github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
@@ -30,8 +30,12 @@ github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6r
3030
github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
3131
github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
3232
github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
33-
github.com/cespare/xxhash/v2 v2.1.1 h1:6MnRN8NT7+YBpUIWxHtefFZOKTAPgGjpQSxqLNn0+qY=
3433
github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
34+
github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
35+
github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
36+
github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
37+
github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
38+
github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
3539
github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
3640
github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
3741
github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
@@ -99,8 +103,8 @@ github.com/go-playground/universal-translator v0.17.0 h1:icxd5fm+REJzpZx7ZfpaD87
99103
github.com/go-playground/universal-translator v0.17.0/go.mod h1:UkSxE5sNxxRwHyU+Scu5vgOQjsIJAF8j9muTVoKLVtA=
100104
github.com/go-playground/validator/v10 v10.4.1 h1:pH2c5ADXtd66mxoE0Zm9SUhxE20r7aM3F26W0hOn+GE=
101105
github.com/go-playground/validator/v10 v10.4.1/go.mod h1:nlOn6nFhuKACm19sB/8EGNn9GlaMV7XkbRSipzJ0Ii4=
102-
github.com/go-redis/redis/v8 v8.11.0 h1:O1Td0mQ8UFChQ3N9zFQqo6kTU2cJ+/it88gDB+zg0wo=
103-
github.com/go-redis/redis/v8 v8.11.0/go.mod h1:DLomh7y2e3ggQXQLd1YgmvIfecPJoFl7WU5SOQ/r06M=
106+
github.com/go-redis/redis/v8 v8.11.5 h1:AcZZR7igkdvfVmQTPnu9WE37LRrO/YrBH5zWyjDC0oI=
107+
github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo=
104108
github.com/go-resty/resty/v2 v2.7.0 h1:me+K9p3uhSmXtrBZ4k9jcEAfJmuC8IivWHwaLZwPrFY=
105109
github.com/go-resty/resty/v2 v2.7.0/go.mod h1:9PWDzw47qPphMRFfhsyk0NnSgvluHcljSMVIq3w7q0I=
106110
github.com/go-sql-driver/mysql v1.6.0 h1:BCTh4TKNUYmOmMUcQ3IipzF5prigylS7XXjEkfCHuOE=
@@ -145,10 +149,10 @@ github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
145149
github.com/google/go-cmp v0.5.3/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
146150
github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
147151
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
148-
github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
149152
github.com/google/go-cmp v0.5.7 h1:81/ik6ipDQS2aGcBfIN5dHDB36BwrStyeAQquSYCV4o=
150153
github.com/google/go-cmp v0.5.7/go.mod h1:n+brtR0CgQNWTVd5ZUFpTBC8YFBDLK/h/bpaJ8/DtOE=
151154
github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
155+
github.com/google/pprof v0.0.0-20210407192527-94a9f03dee38/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
152156
github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
153157
github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
154158
github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
@@ -168,6 +172,7 @@ github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFb
168172
github.com/hashicorp/go-uuid v1.0.2 h1:cfejS+Tpcp13yd5nYHWDI6qVCny6wyX2Mt5SGur2IGE=
169173
github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
170174
github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
175+
github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
171176
github.com/jcmturner/aescts/v2 v2.0.0 h1:9YKLH6ey7H4eDBXW8khjYslgyqG2xZikXP0EQFKrle8=
172177
github.com/jcmturner/aescts/v2 v2.0.0/go.mod h1:AiaICIRyfYg35RUkr8yESTqvSy7csK90qZ5xfvvsoNs=
173178
github.com/jcmturner/dnsutils/v2 v2.0.0 h1:lltnkeZGL0wILNvrNiVCR6Ro5PGU/SeBvVO/8c/iPbo=
@@ -256,14 +261,16 @@ github.com/olivere/elastic/v7 v7.0.25 h1:q3ef8PqC4PyT3b8BAcjDVo48KNzr0HVKosMqMsF
256261
github.com/olivere/elastic/v7 v7.0.25/go.mod h1:ySKeM+7yrE9HmsUi6+vSp0anvWiDOuPa9kpuknxjKbU=
257262
github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
258263
github.com/onsi/ginkgo v1.12.1/go.mod h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108oapk=
259-
github.com/onsi/ginkgo v1.15.0/go.mod h1:hF8qUzuuC8DJGygJH3726JnCZX4MYbRB8yFfISqnKUg=
260-
github.com/onsi/ginkgo v1.16.4 h1:29JGrr5oVBm5ulCWet69zQkzWipVXIol6ygQUe/EzNc=
261264
github.com/onsi/ginkgo v1.16.4/go.mod h1:dX+/inL/fNMqNlz0e9LfyB9TswhZpCVdJM/Z6Vvnwo0=
265+
github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
266+
github.com/onsi/ginkgo v1.16.5/go.mod h1:+E8gABHa3K6zRBolWtd+ROzc/U5bkGt0FwiG042wbpU=
267+
github.com/onsi/ginkgo/v2 v2.0.0 h1:CcuG/HvWNkkaqCUpJifQY8z7qEMBJya6aLPx6ftGyjQ=
268+
github.com/onsi/ginkgo/v2 v2.0.0/go.mod h1:vw5CSIxN1JObi/U8gcbwft7ZxR2dgaR70JSE3/PpL4c=
262269
github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY=
263270
github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
264-
github.com/onsi/gomega v1.10.5/go.mod h1:gza4q3jKQJijlu05nKWRCW/GavJumGt8aNRxWg7mt48=
265-
github.com/onsi/gomega v1.15.0 h1:WjP/FQ/sk43MRmnEcT+MlDw2TFvkrXlprrPST/IudjU=
266-
github.com/onsi/gomega v1.15.0/go.mod h1:cIuvLEne0aoVhAgh/O6ac0Op8WWw9H6eYCriF+tEHG0=
271+
github.com/onsi/gomega v1.17.0/go.mod h1:HnhC7FXeEQY45zxNK3PPoIUhzk/80Xly9PcubAlGdZY=
272+
github.com/onsi/gomega v1.18.1 h1:M1GfJqGRrBrrGGsbxzV5dqM2U2ApXefZCQpkukxYRLE=
273+
github.com/onsi/gomega v1.18.1/go.mod h1:0q+aL8jAiMXy9hbwj2mr5GziHiwhAIQpFmmtT5hitRs=
267274
github.com/opentracing/opentracing-go v1.2.0 h1:uEJPy/1a5RIPAJ0Ov+OIO8OxWu77jEv+1B0VhjKrZUs=
268275
github.com/opentracing/opentracing-go v1.2.0/go.mod h1:GxEUsuufX4nBwe+T+Wl9TAgYrxe9dPLANfrWvHYVTgc=
269276
github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
@@ -307,8 +314,6 @@ github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTE
307314
github.com/rs/xid v1.2.1 h1:mhH9Nq+C1fY2l1XIpgxIiUOfNpRBYH1kKcr+qfKgjRc=
308315
github.com/rs/xid v1.2.1/go.mod h1:+uKXf+4Djp6Md1KODXJxgGQPKngRmWyn10oCKFzNHOQ=
309316
github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
310-
github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww=
311-
github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
312317
github.com/shirou/gopsutil v3.20.11+incompatible h1:LJr4ZQK4mPpIV5gOa4jCOKOGb4ty4DZO54I4FGqIpto=
313318
github.com/shirou/gopsutil v3.20.11+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA=
314319
github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
@@ -328,8 +333,6 @@ github.com/smartystreets/gunit v1.4.2/go.mod h1:ZjM1ozSIMJlAz/ay4SG8PeKF00ckUp+z
328333
github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
329334
github.com/spf13/cast v1.4.1 h1:s0hze+J0196ZfEMTs80N7UlFt0BDuQ7Q+JDnHiMWKdA=
330335
github.com/spf13/cast v1.4.1/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
331-
github.com/stathat/consistent v1.0.0 h1:ZFJ1QTRn8npNBKW065raSZ8xfOqhpb8vLOkfp4CcL/U=
332-
github.com/stathat/consistent v1.0.0/go.mod h1:uajTPbgSygZBJ+V+0mY7meZ8i0XAcZs7AQ6V121XSxw=
333336
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
334337
github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
335338
github.com/stretchr/objx v0.2.0 h1:Hbg2NidpLE8veEBkEZTL3CvlkUIVzuU9jDplZO54c48=
@@ -434,7 +437,6 @@ golang.org/x/net v0.0.0-20200707034311-ab3426394381/go.mod h1:/O7V0waA8r7cgGh81R
434437
golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
435438
golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
436439
golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
437-
golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
438440
golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
439441
golang.org/x/net v0.0.0-20210428140749-89ef3d95e781/go.mod h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk=
440442
golang.org/x/net v0.0.0-20211029224645-99673261e6eb/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
@@ -465,6 +467,7 @@ golang.org/x/sys v0.0.0-20190904154756-749cb33beabd/go.mod h1:h1NjWce9XRLGQEsW7w
465467
golang.org/x/sys v0.0.0-20191005200804-aed5e4c7ecf9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
466468
golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
467469
golang.org/x/sys v0.0.0-20191120155948-bd437916bb0e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
470+
golang.org/x/sys v0.0.0-20191204072324-ce4227a45e2e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
468471
golang.org/x/sys v0.0.0-20200106162015-b016eb3dc98e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
469472
golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
470473
golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

‎pkg/broker/rocketmq/consumer.go‎

Lines changed: 104 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -21,11 +21,11 @@ package rocketmq
2121
import (
2222
"context"
2323
"fmt"
24-
"strings"
25-
2624
rocketmq "github.com/apache/rocketmq-client-go/v2"
2725
"github.com/apache/rocketmq-client-go/v2/consumer"
2826
"github.com/apache/rocketmq-client-go/v2/primitive"
27+
"strings"
28+
"time"
2929
)
3030

3131
// ConsumerConfig RocketMQ consumer config
@@ -41,6 +41,9 @@ type ConsumerConfig struct {
4141

4242
Orderly bool
4343
interceptors []primitive.Interceptor
44+
45+
PullTimeout time.Duration
46+
PullBatchSize int32
4447
}
4548

4649
// PushConsumer push consumer mode
@@ -55,17 +58,21 @@ func NewPushConsumer(config *ConsumerConfig) *PushConsumer {
5558
AccessKey: config.AccessKey,
5659
SecretKey: config.SecretKey,
5760
}
58-
59-
consumer, err := rocketmq.NewPushConsumer(
61+
opts := []consumer.Option{
6062
consumer.WithNsResolver(primitive.NewPassthroughResolver(config.Endpoint)),
6163
consumer.WithCredentials(credentials),
6264
consumer.WithGroupName(config.Gid),
6365
consumer.WithConsumerOrder(config.Orderly),
6466
consumer.WithMaxReconsumeTimes(config.Retry),
65-
consumer.WithInterceptor(pushConsumerMetricInterceptor(config)),
67+
consumer.WithInterceptor(consumerMetricInterceptor(config)),
6668
consumer.WithInterceptor(config.interceptors...),
67-
)
68-
69+
}
70+
if config.PullBatchSize > 0 && config.PullBatchSize <= 1024 {
71+
opts = append(opts, consumer.WithPullBatchSize(config.PullBatchSize))
72+
} else {
73+
opts = append(opts, consumer.WithPullBatchSize(32))
74+
}
75+
consumer, err := rocketmq.NewPushConsumer(opts...)
6976
if err != nil {
7077
panic(fmt.Sprintf("init consumer fail, err msg: %s", err.Error()))
7178
}
@@ -101,7 +108,6 @@ func (pc *PushConsumer) Subscribe(cb func(context.Context, *primitive.MessageExt
101108
return consumer.ConsumeRetryLater, err
102109
}
103110
}
104-
105111
return consumer.ConsumeSuccess, nil
106112
})
107113

@@ -110,3 +116,93 @@ func (pc *PushConsumer) Subscribe(cb func(context.Context, *primitive.MessageExt
110116
}
111117
return pc
112118
}
119+
120+
// PullConsumer pull consumer mode
121+
type PullConsumer struct {
122+
consumer rocketmq.PullConsumer
123+
config *ConsumerConfig
124+
cb func(context.Context, *primitive.MessageExt) error
125+
}
126+
127+
// NewPullConsumer returns a PullConsumer instance
128+
func NewPullConsumer(config *ConsumerConfig) *PullConsumer {
129+
if config.PullTimeout == 0 {
130+
config.PullTimeout = time.Minute // default poll every 1min
131+
}
132+
var credentials = primitive.Credentials{
133+
AccessKey: config.AccessKey,
134+
SecretKey: config.SecretKey,
135+
}
136+
137+
opts := []consumer.Option{
138+
consumer.WithNsResolver(primitive.NewPassthroughResolver(config.Endpoint)),
139+
consumer.WithCredentials(credentials),
140+
consumer.WithGroupName(config.Gid),
141+
consumer.WithConsumerOrder(config.Orderly),
142+
consumer.WithMaxReconsumeTimes(config.Retry),
143+
consumer.WithInterceptor(consumerMetricInterceptor(config)),
144+
consumer.WithInterceptor(config.interceptors...),
145+
}
146+
if config.PullBatchSize > 0 && config.PullBatchSize <= 1024 {
147+
opts = append(opts, consumer.WithPullBatchSize(config.PullBatchSize))
148+
} else {
149+
opts = append(opts, consumer.WithPullBatchSize(32))
150+
}
151+
consumer, err := rocketmq.NewPullConsumer(opts...)
152+
if err != nil {
153+
panic(fmt.Sprintf("init consumer fail, err msg: %s", err.Error()))
154+
}
155+
156+
pc := &PullConsumer{
157+
consumer: consumer,
158+
config: config,
159+
}
160+
return pc
161+
}
162+
163+
// Start start consumer
164+
func (pc *PullConsumer) Start() error {
165+
err := pc.consumer.Start()
166+
if err != nil {
167+
return err
168+
}
169+
go pc.poll(context.Background())
170+
return nil
171+
}
172+
173+
// Shutdown consumer
174+
func (pc *PullConsumer) Shutdown() error {
175+
return pc.consumer.Shutdown()
176+
}
177+
178+
// Subscribe subscribe topic
179+
func (pc *PullConsumer) Subscribe(cb func(context.Context, *primitive.MessageExt) error) error {
180+
selector := consumer.MessageSelector{
181+
Type: consumer.TAG,
182+
Expression: strings.Join(pc.config.Tags, "||"),
183+
}
184+
pc.cb = cb
185+
return pc.consumer.Subscribe(pc.config.Topic, selector)
186+
}
187+
188+
// poll messages with timeout
189+
func (pc *PullConsumer) poll(ctx context.Context) {
190+
for {
191+
cr, err := pc.consumer.Poll(ctx, pc.config.PullTimeout)
192+
if err != nil {
193+
continue
194+
}
195+
196+
for _, msg := range cr.GetMsgList() {
197+
err = pc.cb(ctx, msg)
198+
if err != nil {
199+
break
200+
}
201+
}
202+
if err != nil {
203+
pc.consumer.ACK(context.Background(), cr, consumer.ConsumeRetryLater)
204+
continue
205+
}
206+
pc.consumer.ACK(context.Background(), cr, consumer.ConsumeSuccess)
207+
}
208+
}

‎pkg/broker/rocketmq/interceptor.go‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -78,22 +78,25 @@ func producerMetricInterceptor(pc *ProducerConfig) primitive.Interceptor {
7878
}
7979
}
8080

81-
// pushConsumerMetricInterceptor push consumer metric
82-
func pushConsumerMetricInterceptor(pc *ConsumerConfig) primitive.Interceptor {
81+
// consumerMetricInterceptor consumer metric
82+
func consumerMetricInterceptor(pc *ConsumerConfig) primitive.Interceptor {
8383
return func(ctx context.Context, req, reply interface{}, next primitive.Invoker) error {
8484
now := time.Now()
8585
msgs := req.([]*primitive.MessageExt)
8686

8787
err := next(ctx, msgs, reply)
8888

89+
if reply == nil {
90+
return err
91+
}
92+
8993
var errmsg string
9094
if err != nil {
9195
errmsg = err.Error()
9296
}
9397
holder := reply.(*consumer.ConsumeResultHolder)
9498
replyCode := strconv.Itoa(int(holder.ConsumeResult))
9599
duration := time.Since(now).Seconds()
96-
97100
for _, msg := range msgs {
98101
metric.RocketMQClientHandleCounter.Inc(msg.StoreHost, "rocketmq", pc.Topic, "consume", replyCode)
99102
metric.RocketMQClientReqDuration.Observe(duration, msg.StoreHost, "rocketmq", pc.Topic, "consume")

0 commit comments

Comments
 (0)