add files

src/collector.go

@@ -0,0 +1,50 @@
+package main
+
+import (
+	"sync"
+	"github.com/Shopify/sarama"
+	"encoding/json"
+	"log"
+)
+
+var (
+	wg sync.WaitGroup
+)
+
+func start(pool *ResourcePool) {
+	consumer, err := sarama.NewConsumer([]string{"kafka_node1:9092", "kafka_node2:9093", "kafka_node3:9094"}, nil)
+	if err != nil {
+		panic(err)
+	}
+
+	partitionList, err := consumer.Partitions("yao")
+	if err != nil {
+		panic(err)
+	}
+
+	for partition := range partitionList {
+		pc, err := consumer.ConsumePartition("yao", int32(partition), sarama.OffsetNewest)
+		if err != nil {
+			panic(err)
+		}
+		defer pc.AsyncClose()
+
+		wg.Add(1)
+
+		go func(sarama.PartitionConsumer) {
+			defer wg.Done()
+			for msg := range pc.Messages() {
+				var msgAgent MsgAgent
+				err = json.Unmarshal([]byte(string(msg.Value)), &msgAgent)
+				if err != nil {
+					log.Println(err)
+					continue
+				}
+				pool.update(msgAgent)
+			}
+
+		}(pc)
+	}
+	wg.Wait()
+	consumer.Close()
+}