关于kafka的疑惑

Kafka Confusion

1.kafka

(1) consumer与partition是一对多的关系

consumer group里的一个consumer可以消费多个partition，但是一个partition只能被一个consumer消费，因为kafka的设计是不允许在一个partition上并发的。为了提高消费的并发，必须增加partition的数量。

如果consumer比partition多，会导致某些consumer无法消费到partition。所以要让partition的数量大于等于consumer的数量。合理的部署方案是，起初将partition数量尽量分配得多一些(通常是2的整数次幂)，以后为了扩展就直接增加consumer的数量。

当partition数量多于consumer数量时，kafka会尽量均衡每个consumer消费的partition数量，以此达到每个consumer负载均衡的目的。参考以下这句话(来自于kafka官方文档 http://kafka.apache.org/documentation.html)：

The consumers in a group divide up the partitions as fairly as possible, each partition is consumed by exactly one consumer in a consumer group.

以下链接对此做了实验：

http://www.jasongj.com/2015/08/09/KafkaColumn4/

2.saram

github.com/Shopify/sarama 是go语言实现的kafka client库。

(1) SyncProducer是对AsyncProducer作的封装

SyncProducer每发送出去一条消息，就等待返回结果，然后再发下一条。因此SyncProducer不支持批量发送。

(2) AsyncProducer的Input()和Successes()的阻塞问题

AsyncProducer的两个方法：Input()返回用来写入消息的channel，Successes()返回用来收集发送成功的消息的channel(Errors()用来收集发送失败的消息)。

应用程序可以用一个goroutine不断地向Input()写入消息，用另一个goroutine从Successes()和Errors()里读取发送结果，以此实现异步发送。而这两个操作也可以写在一个for-select里：

for {
select {
case producer.Input() <- message:
// do something
case message = <- producer.Successes():
// do something
case err := <- producer.Errors()
// do something
}
}

只有sarama.Config.Producer.Return.Successes设置为true，才可以从producer.Successes()里读取。而且，如果该参数为true，必须读取producer.Successes()，否则producer.successes channel就满，进而导致producer.input channel也满，然后写producer.Input()的时候就阻塞了。

3.consumer group

github.com/wvanbergen/kafka/consumergroup 是基于sarama开发的对consumer group的扩展。sarama目前不支持consumer group功能。

(1) consumergroup包是将offset提交到zookeeper上，而不是kafka上。 它在初始化(join group)时需要zookeeper地址。

(2) 配置消费起始点

配置里有两个参数与此有关：
Offsets.Initial      ： 从最旧还是最新的消息开始消费，只能是sarama.OffsetOldest或sarama.OffsetNewest。
Offsets.ResetOffsets ：　bool类型。如果程序重启，true表示不从上次中断的位置消费，false表示从上次中断的位置消费。
pdos-server里需要设置Offsets.Initial=sarama.OffsetNewest，Offsets.ResetOffsets=true。