メッセージ送受信サンプルコード

Apache Kafkaのメッセージの送信・受信を行うgolangのサンプルコードです。

cf. https://github.com/segmentio/kafka-go

事前準備

• トピックquickstart-eventsを作成してください:

$ bin/kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092
Created topic quickstart-events.

$ bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092
Topic: quickstart-events        TopicId: mvQtvjNfRgWhufRT9kQv0w PartitionCount: 1       ReplicationFactor: 1    Configs: min.insync.replicas=1,segment.bytes=1073741824
        Topic: quickstart-events        Partition: 0    Leader: 1       Replicas: 1     Isr: 1  Elr:    LastKnownElr:

• 下記で必要なパッケージをインストールしてください:

go get github.com/segmentio/kafka-go

送信

package main

import (
	"context"
	"log"

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

func main() {
	w := kafka.NewWriter(kafka.WriterConfig{
		Brokers: []string{"localhost:9092"},
		Topic:   "quickstart-events",
	})
	defer w.Close()

	body := "Hello World!"
	err := w.WriteMessages(context.Background(),
		kafka.Message{
			Key:   []byte("hello"),
			Value: []byte(body),
		},
	)

	if err != nil {
		panic(err)
	}

	log.Printf(" Sent %s", body)
}

実行すると、下記のようなメッセージが表示されます:

$ go run main.go
2026/01/11 14:35:32  Sent Hello World!

受信

package main

import (
	"context"
	"log"

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

func main() {
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers: []string{"localhost:9092"},
		Topic:   "quickstart-events",
	})
	defer r.Close()

	for {
		msg, err := r.ReadMessage(context.Background())
		if err != nil {
			panic(err)
		}

		log.Printf(" Received a message: %s\n", string(msg.Value))
	}
}

メッセージがpushされると下記のようなメッセージが出力されます:

2026/01/11 14:46:00  Received a message: Hello World!