1
package main
2

3
import (
4
  "fmt"
5

6
  "github.com/Shopify/sarama"
7
)
8

9
// kafka consumer
10

11
func main() {
12
  consumer, err := sarama.NewConsumer([]string{"127.0.0.1:9092"}, nil)
13
  if err != nil {
14
    fmt.Printf("fail to start consumer, err:%v\n", err)
15
    return
16
  }
17
  partitionList, err := consumer.Partitions("web_log") // 根据topic取到所有的分区
18
  if err != nil {
19
    fmt.Printf("fail to get list of partition:err%v\n", err)
20
    return
21
  }
22
  fmt.Println(partitionList)
23
  for partition := range partitionList { // 遍历所有的分区
24
    // 针对每个分区创建一个对应的分区消费者
25
    pc, err := consumer.ConsumePartition("web_log", int32(partition), sarama.OffsetNewest)
26
    if err != nil {
27
      fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err)
28
      return
29
    }
30
    defer pc.AsyncClose()
31
    // 异步从每个分区消费信息
32
    go func(sarama.PartitionConsumer) {
33
      for msg := range pc.Messages() {
34
        fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v", msg.Partition, msg.Offset, msg.Key, msg.Value)
35
      }
36
    }(pc)
37
  }
38
}

1
// 0. 加载配置文件
2
var cfg = new(conf.LogTransferCfg)
3
err := ini.MapTo(cfg, "./conf/cfg.ini")
4
if err != nil {
5
  fmt.Printf("init config failed, err:%v\n", err)
6
  return
7
}
8
fmt.Printf("cfg:%v\n", cfg)