kq下多Processors位移提交的问题

[wanzirong]

感谢对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并行提交位移是否有问题？如提前把大位移提交导致丢消息，或滞后把小位移提交导致重复消费。

[colinrs]

我看了代码，认为也会是这样，会导致丢消息和重复消费

[dinofei]

感谢对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一起消费也只会按顺序提交

[1426919587]

感谢对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时间不确定，即使消费者按顺序，提交也不一定按顺序

[dinofei]

感谢对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排序的

[1426919587]

感谢对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的消息