Go Language

How to Create Async Producers with Kafka & Go Sarama

It's often useful to not wait for a Kafka producer, you can use an async producer

Golang Community Moderator. Software Engineer - Open Source Enthusiast. Goper Lover

Published · Jul 28 2019

It's often useful to not wait for a Kafka producer to complete before moving on to the next task. In cases like this, you can use an async producer. These producers take Sarama messages on a channel and have methods to return a success/error channel that can be checked separately.

  
In this recipe, we'll create a go routine that will handle success and failure messages while we allow a handler to queue messages to send regardless of the result.

Create a file called producer.go:
Create a file called handler.go:
Create a file called main.go:
Run the go build command.
Navigate up a directory.
Run go run main.go.
In a separate terminal from the same directory,  go run producer.go
In a third terminal, run the following commands:
$ curl "http://localhost:3333/?msg=this" 
$ curl "http://localhost:3333/?msg=is" 
$ curl "http://localhost:3333/?msg=an" 
$ curl "http://localhost:3333/?msg=example"

In the producer terminal, you should see the following:
$ go run producer.go
Listening on port :3333
2017/05/07 13:52:54 > message: "this" sent to partition 0 at 
offset 0
2017/05/07 13:53:25 > message: "is" sent to partition 0 at offset 
1
2017/05/07 13:53:27 > message: "an" sent to partition 0 at offset 
2
2017/05/07 13:53:29 > message: "example" sent to partition 0 at 
offset 3


How it works...
Our modifications in this chapter were all made to the producer. This time, we created a separate go routine to handle successes and errors. If these are left unhandled, your application will deadlock. Next, we attached our producer to a handler and we emit messages on it whenever a message is received via a GET call to the handler.

The handler will immediately return success upon sending the message regardless of its response. If this is not acceptable, a synchronous approach should be used instead. In our case, we're okay with later processing success and errors separately.

Lastly, we curl our endpoint with a few different messages and you can see them flow from the handler to where they're eventually printed by the Kafka consumer we wrote in the previous section.