Working with Kafka and NATS
Producing and consuming messages from Go with kafka-go and nats.go, and how consumer groups split a topic between processes.
func (nc *nats.Conn) QueueSubscribe(subj, queue string, cb nats.MsgHandler) (*nats.Subscription, error)Kafka and NATS are the two brokers you will meet most often. A broker is a server that holds messages between programs. Go talks to it over the network through a client library you add with go get.
Which library
For Kafka, github.com/segmentio/kafka-go is written in Go and needs nothing else. github.com/confluentinc/confluent-kafka-go wraps a C library, so it needs cgo and a C compiler wherever you build. For NATS, github.com/nats-io/nats.go is the official client. The example below needs a nats-server running on localhost; start one, then go get github.com/nats-io/nats.go and go run ..
Producing and consuming
In Kafka you write to a topic, a named stream of messages. Each topic is split into numbered partitions, and a partition is a list that only grows at the end. A message's place in its partition is its offset. A consumer reads forward and saves how far it got by committing an offset. In NATS the name is a subject, and orders.* subscribes to every subject one level below orders.
Consumer groups
Give several Kafka consumers the same group name, GroupID in kafka-go, and the broker gives each partition to exactly one member. Start a second copy of your service and the work splits. When a member dies its partitions move to the survivors, which is called a rebalance. Two different group names on one topic each receive every message.
kafka-go's ReadMessage is FetchMessage plus, when GroupID is set, an immediate CommitMessages on that message, so the offset is committed before your handler has run; the docs recommend FetchMessage with CommitMessages when you need to commit after the work instead. NATS has no groups, but a queue name passed to QueueSubscribe makes a set of subscribers share a subject instead of all getting copies.
Example
package main
import (
"fmt"
"log"
"time"
"github.com/nats-io/nats.go"
)
func main() {
// nats.DefaultURL is "nats://127.0.0.1:4222".
nc, err := nats.Connect(nats.DefaultURL)
if err != nil {
log.Fatal(err)
}
defer nc.Close()
// Subscribe first: NATS core keeps nothing for a subject nobody is on.
sub, err := nc.SubscribeSync("orders.paid")
if err != nil {
log.Fatal(err)
}
if err := nc.Publish("orders.paid", []byte("order-7")); err != nil {
log.Fatal(err)
}
// Publish buffers; Flush waits for the server to have it.
if err := nc.Flush(); err != nil {
log.Fatal(err)
}
msg, err := sub.NextMsg(2 * time.Second) // wait up to two seconds
if err != nil {
log.Fatal(err)
}
fmt.Printf("%s: %s\n", msg.Subject, msg.Data)
}
// Output:
// orders.paid: order-7When to use it
- Write an order event to Kafka from your HTTP handler and let a separate worker send the email.
- Run three copies of a worker and have Kafka split the topic's partitions between them.
- Scale a NATS worker pool horizontally with a queue group so each job is handled once.
- Fan a cache-invalidation message out to every running instance over a NATS subject.
More examples
Kafka producer with kafka-go
One Writer is safe to share across goroutines and should be built once at startup, next to your database pool, not per request. RequiredAcks is the dial that trades speed for safety: RequireNone returns without waiting and can lose data, RequireAll is slower and survives a broker dying. Set the Balancer explicitly whenever the key matters.
package main
import (
"context"
"log"
"github.com/segmentio/kafka-go"
)
func main() {
w := &kafka.Writer{
Addr: kafka.TCP("localhost:9092"),
Topic: "orders",
// Without a Balancer, kafka-go spreads messages round-robin and
// ignores the key.
Balancer: &kafka.Hash{},
// RequireAll waits for every replica to store the message.
RequiredAcks: kafka.RequireAll,
}
defer w.Close()
err := w.WriteMessages(context.Background(),
kafka.Message{Key: []byte("order-7"), Value: []byte(`{"event":"created"}`)},
kafka.Message{Key: []byte("order-7"), Value: []byte(`{"event":"paid"}`)},
)
if err != nil {
log.Fatal(err)
}
log.Print("two messages written")
}Kafka consumer group with manual commits
This is the shape of almost every Kafka worker: fetch, do the work, commit. Run two copies of this program and the group splits the partitions between them with no code change. Replace log.Fatal on the fetch error with a check for a cancelled context when you wire this into graceful shutdown.
package main
import (
"context"
"log"
"github.com/segmentio/kafka-go"
)
func main() {
r := kafka.NewReader(kafka.ReaderConfig{
Brokers: []string{"localhost:9092"},
GroupID: "order-emailer",
Topic: "orders",
MinBytes: 1,
MaxBytes: 10e6,
})
defer r.Close()
ctx := context.Background()
for {
// FetchMessage reads without committing.
m, err := r.FetchMessage(ctx)
if err != nil {
log.Fatal(err)
}
log.Printf("partition %d offset %d key %s", m.Partition, m.Offset, m.Key)
// Commit only once the work is done.
if err := r.CommitMessages(ctx, m); err != nil {
log.Fatal(err)
}
}
}NATS queue group: queue behaviour on a subject
Subscribe and QueueSubscribe are the same call with one extra argument, and that argument is the whole difference between pub/sub and a queue. Use QueueSubscribe for work that must happen once however many copies of the service are running, and plain Subscribe for notifications every copy needs, such as clearing a local cache.
package main
import (
"log"
"os"
"os/signal"
"github.com/nats-io/nats.go"
)
func main() {
nc, err := nats.Connect(nats.DefaultURL)
if err != nil {
log.Fatal(err)
}
// Drain delivers what has already arrived, then closes the connection.
defer nc.Drain()
// Every process using the queue name "workers" shares the messages:
// each message on orders.* reaches exactly one of them. Plain
// nc.Subscribe would instead give every process a copy.
_, err = nc.QueueSubscribe("orders.*", "workers", func(m *nats.Msg) {
log.Printf("%s: %s", m.Subject, m.Data)
})
if err != nil {
log.Fatal(err)
}
stop := make(chan os.Signal, 1)
signal.Notify(stop, os.Interrupt)
<-stop // block until Ctrl-C
}
Discussion