1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
|
package main
import (
"context"
"fmt"
"os"
"time"
"github.com/google/uuid"
kafka "github.com/segmentio/kafka-go"
)
func newKafkaWriter(kafkaURL, topic string) *kafka.Writer {
return &kafka.Writer{
Addr: kafka.TCP(kafkaURL),
Topic: topic,
Balancer: &kafka.LeastBytes{},
}
}
func main() {
// get kafka writer using environment variables.
kafkaURL := os.Getenv("kafkaURL")
topic := os.Getenv("topic")
writer := newKafkaWriter(kafkaURL, topic)
defer writer.Close()
fmt.Println("start producing ... !!")
for i := 0; ; i++ {
key := fmt.Sprintf("Key-%d", i)
msg := kafka.Message{
Key: []byte(key),
Value: []byte(fmt.Sprint(uuid.New())),
}
err := writer.WriteMessages(context.Background(), msg)
if err != nil {
fmt.Println(err)
} else {
fmt.Println("produced", key)
}
time.Sleep(1 * time.Second)
}
}
|
