goKafka/encode-consume.go

package main

import (
	"encoding/json"
	"github.com/Shopify/sarama"
	"log"
	"os"
	"os/signal"
	"time"
)

type Event struct {
	Name   string    `json:"name"`
	Type   string    `json:"type"`
	Source string    `json:"source"`
	Target string    `json:"target"`
	Time   time.Time `json:"time"`
}

func main() {
	// 一，获取消费者
	consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, nil)
	if err != nil {
		log.Fatalln(err)
	}
	defer func() {
		if err := consumer.Close(); err != nil {
			log.Fatalln(err)
		}
	}()

	// 二，获取topic具体某个partition的的消费者
	// topic 1:N partition
	partitionConsumer, err := consumer.ConsumePartition("event_topic", 0, sarama.OffsetNewest)
	if err != nil {
		log.Fatalln(err)
	}
	defer func() {
		if err := partitionConsumer.Close(); err != nil {
			log.Fatalln(err)
		}
	}()

	// 三，消费
	signals := make(chan os.Signal, 1)
	signal.Notify(signals, os.Interrupt)

	consumeCounter := 0
loop:
	for {
		select {
		case msg := <-partitionConsumer.Messages():
			log.Printf("Consumed message, partition: %d, offset: %d, topic: %s, value: %s, type:%T\n", msg.Partition, msg.Offset, msg.Topic, msg.Value, msg.Value)
			evt := Event{}
			err := json.Unmarshal(msg.Value, &evt)
			if err != nil {
				log.Fatalln(err)
			}
			log.Println(evt)
			log.Printf("%T\n", evt)

			consumeCounter++

		case <-signals:
			break loop
		}
	}

	log.Printf("Consumed: %d\n", consumeCounter)
}