Tutorial을 위한 Consumer 구현

• 코드

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/segmentio/kafka-go"
)

func main() {

	fmt.Println("Kafka Consumer Example")
	kafkaServer := "192.168.20.121:9092"
	conn, err := kafka.DialLeader(context.Background(), "tcp", kafkaServer, "topic_test", 0)
	if err != nil {
		log.Fatal("failed to dial leader: ", err)
	}

	conn.SetReadDeadline(time.Now().Add(time.Second * 3))

	batch := conn.ReadBatch(1e3, 1e9)
	bytes := make([]byte, 1e3)

	for {
		_, err := batch.Read(bytes)
		if err != nil {
			log.Fatal("batch : ", err)
			break
		}
		fmt.Println(string(bytes))
	}

}