Comments (6)
我看了代码,认为也会是这样,会导致丢消息和重复消费
from go-queue.
感谢对
kafka-go
进行服务化封装,的确用起来更简单了!我对消费代码有个疑惑,望解答:
for i := 0; i < q.c.Processors; i++ { q.consumerRoutines.Run(func() { for msg := range q.channel { if err := q.consumeOne(string(msg.Key), string(msg.Value)); err != nil { logx.Errorf("Error on consuming: %s, error: %v", string(msg.Value), err) } q.consumer.CommitMessages(context.Background(), msg) } }) }
多个goroutine并行提交位移是否有问题?如提前把大位移提交导致丢消息,或滞后把小位移提交导致重复消费。
不会,因为channel是无容量的,多个goroutine一起消费也只会按顺序提交
from go-queue.
感谢对
kafka-go
进行服务化封装,的确用起来更简单了!
我对消费代码有个疑惑,望解答:for i := 0; i < q.c.Processors; i++ { q.consumerRoutines.Run(func() { for msg := range q.channel { if err := q.consumeOne(string(msg.Key), string(msg.Value)); err != nil { logx.Errorf("Error on consuming: %s, error: %v", string(msg.Value), err) } q.consumer.CommitMessages(context.Background(), msg) } }) }
多个goroutine并行提交位移是否有问题?如提前把大位移提交导致丢消息,或滞后把小位移提交导致重复消费。
不会,因为channel是无容量的,多个goroutine一起消费也只会按顺序提交
为啥,consumeOne时间不确定,即使消费者按顺序,提交也不一定按顺序
from go-queue.
感谢对
kafka-go
进行服务化封装,的确用起来更简单了!
我对消费代码有个疑惑,望解答:for i := 0; i < q.c.Processors; i++ { q.consumerRoutines.Run(func() { for msg := range q.channel { if err := q.consumeOne(string(msg.Key), string(msg.Value)); err != nil { logx.Errorf("Error on consuming: %s, error: %v", string(msg.Value), err) } q.consumer.CommitMessages(context.Background(), msg) } }) }
多个goroutine并行提交位移是否有问题?如提前把大位移提交导致丢消息,或滞后把小位移提交导致重复消费。
不会,因为channel是无容量的,多个goroutine一起消费也只会按顺序提交
为啥,consumeOne时间不确定,即使消费者按顺序,提交也不一定按顺序
kafka-go 会merge commit,这里面会按照message的offset排序的
from go-queue.
感谢对
kafka-go
进行服务化封装,的确用起来更简单了!
我对消费代码有个疑惑,望解答:for i := 0; i < q.c.Processors; i++ { q.consumerRoutines.Run(func() { for msg := range q.channel { if err := q.consumeOne(string(msg.Key), string(msg.Value)); err != nil { logx.Errorf("Error on consuming: %s, error: %v", string(msg.Value), err) } q.consumer.CommitMessages(context.Background(), msg) } }) }
多个goroutine并行提交位移是否有问题?如提前把大位移提交导致丢消息,或滞后把小位移提交导致重复消费。
不会,因为channel是无容量的,多个goroutine一起消费也只会按顺序提交
为啥,consumeOne时间不确定,即使消费者按顺序,提交也不一定按顺序
kafka-go 会merge commit,这里面会按照message的offset排序的
看了一下源码,是有merge操作
func (o offsetStash) merge(commits []commit) { for _, c := range commits { offsetsByPartition, ok := o[c.topic] if !ok { offsetsByPartition = map[int]int64{} o[c.topic] = offsetsByPartition } if offset, ok := offsetsByPartition[c.partition]; !ok || c.offset > offset { offsetsByPartition[c.partition] = c.offset } } }
但是上面只要大于就更新,不是连续的啊,比如现在有offset的2,3,4个消息处理,4先提交,那2,3都没提交,万一2,3失败了,是不是也相当于他们成功,下一次获取到5的消息
from go-queue.
我也有同样的疑问,看起来是不是只能定义一个processer才能保证提交正常呀,是不是应该设计成像spring-kafka那样,每个consumer对应着一个消费线程,这样就可以保证提交不错乱了
from go-queue.
Related Issues (20)
- kq如何订阅多个主题 HOT 3
- customized kafka message key HOT 2
- pusher.go:47 kafka write errors (1/1) HOT 3
- dq example 疑问 HOT 1
- kq的报错机制似乎并不完善?
- bug: go-queue v1.1.6 an error occurred [NotEnoughReplicas]
- How to manually confirm ACK?
- push.close() 后程序退出丢失消息
- kq how to with SASL ?
- RabbitListener can not Manual ack calls are supported
- 为什么要将收到的消息转换为string HOT 2
- Support SASL/SCRAM authentication method HOT 1
- kq.KqConf.Consumers 感觉没什么意义啊 HOT 6
- 我go get的包是1.1.8版本,但是实际和仓库的最新代码不一样呢 HOT 1
- 生产消息 Push(v string) 使用纳秒并不能保证key唯一 HOT 1
- 这个介绍也写的太寒酸了点吧
- consumer example typo HOT 1
- go-queue 有计划支持pulsar消息队列吗? HOT 3
- rabbitmq 消息手动应答与死信队列
Recommend Projects
-
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
-
Vue.js
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
-
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
-
TensorFlow
An Open Source Machine Learning Framework for Everyone
-
Django
The Web framework for perfectionists with deadlines.
-
Laravel
A PHP framework for web artisans
-
D3
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
-
Recommend Topics
-
javascript
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
-
web
Some thing interesting about web. New door for the world.
-
server
A server is a program made to process requests and deliver data to clients.
-
Machine learning
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
-
Visualization
Some thing interesting about visualization, use data art
-
Game
Some thing interesting about game, make everyone happy.
Recommend Org
-
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
-
Microsoft
Open source projects and samples from Microsoft.
-
Google
Google ❤️ Open Source for everyone.
-
Alibaba
Alibaba Open Source for everyone
-
D3
Data-Driven Documents codes.
-
Tencent
China tencent open source team.
from go-queue.