A quick introduction to how Apache Kafka works and differs from other messaging systems using an example application. In this video I explain partitioning, consumer offsets, replication and many other concepts found in Kafka.
Example
package main
import (
"encoding/json"
"fmt"
k "github.com/segmentio/kafka-go"
"github.com/sohamkamani/golang-kafka-example/kafka"
)
func main() {
k := kafka.New([]string{"localhost:9092"}, "test",
kafka.WithPartitions(3),
kafka.WithReplications(1),
kafka.WithBalancer(&k.Hash{}),
)
//producers
k.Producer(produce("producer_1"))
fmt.Println("--------------------------------------------")
//consumers
go k.Consumer("mobile", consume("mobile_consumer_1"))
go k.Consumer("mobile", consume("mobile_consumer_2"))
go k.Consumer("mobile", consume("mobile_consumer_3"))
go k.Consumer("computer", consume("computer_consumer_1"))
go k.Consumer("computer", consume("computer_consumer_2"))
go k.Consumer("computer", consume("computer_consumer_3"))
exit()
}
type Event struct {
Event string
Key string
Team1 string
Team2 string
}
func produce(id string) func(producer *kafka.Producer) {
return func(producer *kafka.Producer) {
events := []Event{{
Event: "Querter Start",
Key: "America_Canada",
Team1: "America",
Team2: "Canada",
}, {
Event: "Foul",
Key: "Malta_Portugal",
Team1: "Malta",
Team2: "Portugal",
}, {
Event: "Score 39-46",
Key: "America_Canada",
Team1: "America",
Team2: "Canada",
}, {
Event: "Free Throw",
Key: "Brazil_Australia",
Team1: "Brazil",
Team2: "Australia",
}, {
Event: "Score 41-46",
Key: "America_Canada",
Team1: "America",
Team2: "Canada",
}, {
Event: "Querter End",
Key: "Brazil_Australia",
Team1: "Brazil",
Team2: "Australia",
}}
for _, event := range events {
value, _ := json.Marshal(event)
err := producer.Produce(event.Key, value)
if err != nil {
fmt.Println("could not write message " + err.Error())
}
fmt.Printf("%s -> %s %s : %s\n", id, event.Team1, event.Team2, event.Event)
}
}
}
func consume(id string) func(consumer *kafka.Consumer) {
return func(consumer *kafka.Consumer) {
for {
msg, err := consumer.ReadMessage()
if err != nil {
fmt.Println("could not read message " + err.Error())
}
var event Event
json.Unmarshal(msg.Value, &event)
fmt.Printf("%s -> %s %s : %s\n", id, event.Team1, event.Team2, event.Event)
}
}
}
func exit() {
exit := make(chan int)
<-exit
}Output
producer_1 -> America Canada : Querter Start
producer_1 -> Malta Portugal : Foul
producer_1 -> America Canada : Score 39-46
producer_1 -> Brazil Australia : Free Throw
producer_1 -> America Canada : Score 41-46
producer_1 -> Brazil Australia : Querter End
--------------------------------------------
mobile_consumer_2 -> America Canada : Querter Start
mobile_consumer_3 -> Malta Portugal : Foul
mobile_consumer_1 -> America Canada : Score 39-46
mobile_consumer_2 -> Brazil Australia : Free Throw
mobile_consumer_1 -> Brazil Australia : Querter End
mobile_consumer_3 -> America Canada : Score 41-46
computer_consumer_2 -> America Canada : Querter Start
computer_consumer_2 -> Brazil Australia : Free Throw
computer_consumer_3 -> America Canada : Score 39-46
computer_consumer_1 -> Malta Portugal : Foul
computer_consumer_3 -> Brazil Australia : Querter End
computer_consumer_1 -> America Canada : Score 41-46