[Sarama] Sarama 로 Kafka Producer 만들어보기 ( Golang )

 

- 목차

 

들어가며.

이번 글에서는 Golang 의 Kafka Clients 라이브러리인 Sarama 에 대해서 알아보려고 합니다.

무엇보다도 구체적인 예시와 시각적인 자료가 이해하는데에 있어서 큰 도움이 된다고 생각해서

Sarama 를 활용한 Kafka Producer 를 만들어보겠습니다.

 

현 시점의 가장 최신 버전인 Golang 1.19 와 Sarama 1.43.0 버전을 사용하도록 하겠습니다.

Sarama 와 Golang 의 버전 호환성은 아래와 링크를 통해서 확인하실 수 있습니다.

 

https://github.com/IBM/sarama/releases

Releases · IBM/sarama

 

 

go.mod 구성하기.

먼저 sarama 모듈을 설치합니다.

go get 명령어를 통해서 sarama 1.43 버전의 모듈을 설치하였고, 결과는 아래와 같습니다.

go get github.com/IBM/sarama@v1.43.0

 

< go.mod >

module golang

go 1.19

require github.com/IBM/sarama v1.43.0

require (
	github.com/davecgh/go-spew v1.1.1 // indirect
	github.com/eapache/go-resiliency v1.6.0 // indirect
	github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect
	github.com/eapache/queue v1.1.0 // indirect
	github.com/golang/snappy v0.0.4 // indirect
	github.com/hashicorp/errwrap v1.0.0 // indirect
	github.com/hashicorp/go-multierror v1.1.1 // indirect
	github.com/hashicorp/go-uuid v1.0.3 // indirect
	github.com/jcmturner/aescts/v2 v2.0.0 // indirect
	github.com/jcmturner/dnsutils/v2 v2.0.0 // indirect
	github.com/jcmturner/gofork v1.7.6 // indirect
	github.com/jcmturner/gokrb5/v8 v8.4.4 // indirect
	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
	github.com/klauspost/compress v1.17.7 // indirect
	github.com/pierrec/lz4/v4 v4.1.21 // indirect
	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
	golang.org/x/crypto v0.19.0 // indirect
	golang.org/x/net v0.21.0 // indirect
)

 

 

Kafka Cluster 생성하기.

간단히 Docker 를 통해서 Kafka Cluster 를 생성하도록 하겠습니다.

아래 링크는 Docker 를 활용하여 Kafka Cluster 를 생성하는 간단한 방법이 적힌 페이지의 주소입니다.

 

https://westlife0615.tistory.com/474

Docker 로 Kafka Cluster 구축해보기.

 

저는 test-topic 이라는 이름의 토픽을 생성하였고,

advertised url 은 localhost:29091, localhost:29092, localhost:29093 로 설정하였습니다.

 

Kafka Producer 만들어보기.

 

아래 예시 코드는 SyncProducer 를 생성하여 0 부터 999 까지의 값을 test-topic 으로 Push 합니다.

총 1000 개의 메시지를 생성합니다.

 

package main

import (
	"fmt"
	"github.com/IBM/sarama"
)

func getProducer(bootstrapServers []string) sarama.SyncProducer {
	var config *sarama.Config = sarama.NewConfig()
	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true
	producer, err := sarama.NewSyncProducer(bootstrapServers, config)
	if err != nil {
		fmt.Println("failed to create Producer", err)
		return nil
	}

	return producer
}

func sendMessage(producer sarama.SyncProducer, topic string, value string) {
	kafkaMessage := &sarama.ProducerMessage{
		Topic:     topic,
		Partition: -1,
		Value:     sarama.StringEncoder(value),
	}

	partition, offset, err := producer.SendMessage(kafkaMessage)
	if err != nil {
		fmt.Printf("%s error occured.", err.Error())
	} else {
		fmt.Printf("Message was saved to partion: %d.\nMessage offset is: %d.\n", partition, offset)
	}
}

func main() {
	var topic string = "test-topic"
	producer := getProducer([]string{"localhost:29091", "localhost:29092", "localhost:29093"})
	for index := 0; index < 1000; index++ {
		var value string = fmt.Sprintf("%s%d", "message", index)
		sendMessage(producer, topic, value)
	}
}