Go에서 Kafka 생산자와 소비자 구축

Apache Kafka는 실시간 데이터 파이프라인 및 스트리밍 애플리케이션을 구축하는 데 사용되는 강력한 분산 스트리밍 플랫폼입니다. 이 블로그 게시물에서는 Golang을 사용하여 Kafka 생산자와 소비자를 설정하는 과정을 안내합니다.

전제조건

시작하기 전에 컴퓨터에 다음이 설치되어 있는지 확인하세요.

• 고(1.16 이상)

• Docker(Kafka를 로컬에서 실행하는 데 사용)

• 카프카

Docker로 Kafka 설정

Kafka를 빠르게 설정하기 위해 Docker를 사용하겠습니다. 프로젝트 디렉터리에 docker-compose.yml 파일을 만듭니다.

yamlCopy codeversion: '3.7'
services:
  zookeeper:
    image: wurstmeister/zookeeper:3.4.6
    ports:
      - "2181:2181"
  kafka:
    image: wurstmeister/kafka:2.13-2.7.0
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    depends_on:
      - zookeeper

Kafka와 Zookeeper를 시작하려면 다음 명령을 실행하세요.

docker-compose up -d

Go에서 Kafka 프로듀서 만들기

먼저 새 Go 모듈을 초기화합니다.

go mod init kafka-example

kafka-go 라이브러리 설치:

go get github.com/segmentio/kafka-go

이제 producer.go 파일을 생성하고 다음 코드를 추가하세요.

package main

import (
    "context"
    "fmt"
    "github.com/segmentio/kafka-go"
    "log"
    "time"
)

func main() {
    writer := kafka.Writer{
        Addr:     kafka.TCP("localhost:9092"),
        Topic:    "example-topic",
        Balancer: &kafka.LeastBytes{},
    }

    defer writer.Close()

    for i := 0; i < 10; i++ {
        msg := kafka.Message{
            Key:   []byte(fmt.Sprintf("Key-%d", i)),
            Value: []byte(fmt.Sprintf("Hello Kafka %d", i)),
        }

        err := writer.WriteMessages(context.Background(), msg)
        if err != nil {
            log.Fatal("could not write message " + err.Error())
        }

        time.Sleep(1 * time.Second)
        fmt.Printf("Produced message: %s\n", msg.Value)
    }
}

이 코드는 예제 주제 주제에 10개의 메시지를 보내는 Kafka 생산자를 설정합니다.

프로듀서 실행:

go run producer.go

메시지가 생성되었음을 나타내는 출력이 표시됩니다.

Go에서 Kafka 소비자 만들기

consumer.go 파일을 만들고 다음 코드를 추가합니다.

package main

import (
    "context"
    "fmt"
    "github.com/segmentio/kafka-go"
    "log"
)

func main() {
    reader := kafka.NewReader(kafka.ReaderConfig{
        Brokers: []string{"localhost:9092"},
        Topic:   "example-topic",
        GroupID: "example-group",
    })

    defer reader.Close()

    for {
        msg, err := reader.ReadMessage(context.Background())
        if err != nil {
            log.Fatal("could not read message " + err.Error())
        }
        fmt.Printf("Consumed message: %s\n", msg.Value)
    }
}

이 소비자는 예제 주제 주제의 메시지를 읽고 이를 콘솔에 인쇄합니다.

소비자 실행:

go run consumer.go

메시지가 소비되었음을 나타내는 출력이 표시됩니다.

결론

이 블로그 게시물에서는 Golang을 사용하여 Kafka 생산자와 소비자를 설정하는 방법을 시연했습니다. 이 간단한 예는 메시지 생성 및 소비의 기본 사항을 보여 주지만 Kafka의 기능은 이보다 훨씬 뛰어납니다. Kafka를 사용하면 강력하고 확장 가능한 실시간 데이터 처리 시스템을 구축할 수 있습니다.

메시지 분할, 키 기반 메시지 배포, 다른 시스템과의 통합과 같은 고급 기능을 자유롭게 살펴보세요. 즐거운 코딩하세요!

---

그렇습니다! 이 블로그 게시물은 실시간 데이터 처리를 시작하려는 개발자에게 적합한 Go와 함께 Kafka를 사용하는 방법을 간결하게 소개합니다.

위 내용은 Go에서 Kafka 생산자와 소비자 구축의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!