package mainimport (
"fmt"
"log""github.com/confluentinc/confluent-kafka-go/kafka"
)func main() {
// Konfigurasi producer
config := &kafka.ConfigMap{
"bootstrap.servers": "localhost:9092",
}// Buat producer
producer, err := kafka.NewProducer(config)
if err != nil {
log.Fatalf("Failed to create producer: %s", err)
}
defer producer.Close()// Topic yang akan digunakan
topic := "my-logs"// Pesan yang akan dikirim
message := &kafka.Message{
Value: []byte("Hello from Golang!"),
}// Kirim pesan
err = producer.Produce(message, nil)
if err != nil {
log.Printf("Failed to produce message: %v", err)
} else {
fmt.Printf("Produced message to topic %s\n", topic)
}
}
2) Consumer, untuk meng-konsumsi pesan dari topic kafka.
package mainimport (
"fmt"
"log""github.com/confluentinc/confluent-kafka-go/kafka"
)func main() {
// Konfigurasi consumer
config := &kafka.ConfigMap{
"bootstrap.servers": "localhost:9092",
"group.id": "my-group",
"auto.offset.reset": "earliest",
}// Buat consumer
c, err := kafka.NewConsumer(config)
if err != nil {
log.Fatalf("Failed to create consumer: %s", err)
}
defer c.Close()// Subscrib ke topic
err = c.Subscribe([]string{"my-logs"}, nil)
if err != nil {
log.Fatalf("Failed to subscribe: %s", err)
}for {
msg, err := c.ReadMessage(100)
if err == nil {
fmt.Printf("Message on %s: %s\n", msg.TopicPartition, string(msg.Value))
} else {
// Handle errors
}
}
}