개발자의 개발괴발

아래와 같이 코드를 짜고 consume을 했는데 데이터가 나오질 않는다.에러는 ReadMessage에서 나온 err는 timeout 에러를 내보내고 있다.package mainimport ( "fmt" "time" "github.com/confluentinc/confluent-kafka-go/v2/kafka")func Consumer(consumerID int, topicName string) { group_name := fmt.Sprintf("myGroup_%d", consumerID) c, err := kafka.NewConsumer(&kafka.ConfigMap{ "group.instance.id": fmt.Sprintf("myGroup_%d", consumerID), "bootstrap.se..

consumer group은 마치 하나의 consumer가 동작하는 것 처럼 보이게 한다.consumer group이 하나가 있다. 이 consumer group에는 두 개의 consumer가 있다. 이때 이 두 consumer가 하나의 topic(파티션이 3개)에서 데이터를 가져온다면 아래 그림처럼 동작할 것이다. Topic에 데이터가 저장이 된 다는 것은 정확하게는 Topic 안의 partition에 저장이 된다. consumer는 이 토픽의 각 partition들을 나눠서 담당한다. consumer는 여러개의 partition을 담당할 수 있다. 그러나 partition은 단 하나의 consumer와 일을 한다. 위 그림처럼 partition의 수가 더 많으면 consumer가 두개 이상의 part..

이전 편에서 topic에 message를 produce해보았다.이제 topic에서 message를 consume해보자. package mainimport ( "fmt" "time" "github.com/confluentinc/confluent-kafka-go/v2/kafka")func Consumer(consumerID int) { c, err := kafka.NewConsumer(&kafka.ConfigMap{ "group.instance.id": fmt.Sprintf("myGroup_%d", consumerID), "bootstrap.servers": "localhost:9095", "group.id": "myGroup", "auto.offset.reset": "earl..

※ kafka 설치된 것에 따라 명령어의 위치가 다를 수 있음(/opt/bitnami/kafka/bin/kafka-topics.sh 여기에 있을 수도 있고 /bin에 있을 수도 있고...) ※ 쉘에 .sh이 붙고 안붙고 차이가 있을 수 있음 Group 리스트 : kafka-consumer-groups.sh --bootstrap-server localhost:9092 --listGroup 정보 조회 :kafka-consumer-groups.sh --bootstrap-server localhost:9092 \ --group $group --describeGroup Consumer 정보 조회 : kafka-consumer-groups.sh --bootstrap-server localhost:9092 \ ..