Pub/Sub is a pattern where the publisher is not programmed to send a message (payload) to a specific receiver. These messages are sent by publishers to specific channels, and receivers can subscribe to one or more channels to consume those same messages.

Imagine that you have a monolithic backend, however you want to add a new feature to that backend, such as sending emails. Instead of this backend being responsible for sending the emails, you can make it a publisher that sends the emails to a channel to be consumed by another backend (receiver) that will be responsible for sending the emails (like newsletters).

As you may have already understood, we are going to have two backends. One of the backends we will call a server , which will be our message sender. The other backend will be the worker , which will be our small microservice.

First and foremost, let's install our dependencies:

go get github.com/gofiber/fiber/v2
go get github.com/pebbe/zmq4
    app.Get("/", func(c *fiber.Ctx) error {
        return c.SendString("Hello there 👋")
    app.Listen(":3000")
    app.Post("/", func(c *fiber.Ctx) error {
        input := new(Input)
        if err := c.BodyParser(input); err != nil {
            panic(err)
        // ...
    app.Listen(":3000")
    app.Post("/", func(c *fiber.Ctx) error {
        input := new(Input)
        if err := c.BodyParser(input); err != nil {
            panic(err)
        if buffer, err := json.Marshal(input); err != nil {
            panic(err)
        } else {
            // ...
        // ...
    app.Listen(":3000")
    app.Post("/", func(c *fiber.Ctx) error {
        input := new(Input)
        if err := c.BodyParser(input); err != nil {
            panic(err)
        if buffer, err := json.Marshal(input); err != nil {
            panic(err)
        } else {
            s.Send("dev.to", zmq.SNDMORE)
            // ...
        // ...
    app.Listen(":3000")
    app.Post("/", func(c *fiber.Ctx) error {
        input := new(Input)
        if err := c.BodyParser(input); err != nil {
            panic(err)
        if buffer, err := json.Marshal(input); err != nil {
            panic(err)
        } else {
            s.Send("dev.to", zmq.SNDMORE)
            s.Send(string(buffer), 0)
        // ...
    app.Listen(":3000")
    app.Post("/", func(c *fiber.Ctx) error {
        input := new(Input)
        if err := c.BodyParser(input); err != nil {
            panic(err)
        if buffer, err := json.Marshal(input); err != nil {
            panic(err)
        } else {
            s.Send("dev.to", zmq.SNDMORE)
            s.Send(string(buffer), 0)
        return c.SendString("Sent to the subscriber/worker.")
    app.Listen(":3000")
    
Then we will create our ZeroMQ socket of the Subscriber type and we will accept connections through the address that we defined before.
And with an instance of our client created and the connection established, we can subscribe to our channel to receive messages from it.

// @/worker/worker.go
package main
import (
    zmq "github.com/pebbe/zmq4"
func main() {
    zctx, _ := zmq.NewContext()
    s, _ := zctx.NewSocket(zmq.SUB)
    s.Connect("tcp://localhost:5757")
    s.SetSubscribe("dev.to")
    // ...
    s, _ := zctx.NewSocket(zmq.SUB)
    s.Connect("tcp://localhost:5757")
    s.SetSubscribe("dev.to")
    for {
        address, err := s.Recv(0)
        if err != nil {
            panic(err)
        // ...
    s, _ := zctx.NewSocket(zmq.SUB)
    s.Connect("tcp://localhost:5757")
    s.SetSubscribe("dev.to")
    for {
        address, err := s.Recv(0)
        if err != nil {
            panic(err)
        if msg, err := s.Recv(0); err != nil {
            panic(err)
        } else {
            contents := Message{}
            if err := json.Unmarshal([]byte(msg), &contents); err != nil {
                panic(err)
            // ...
    s, _ := zctx.NewSocket(zmq.SUB)
    s.Connect("tcp://localhost:5757")
    s.SetSubscribe("dev.to")
    for {
        address, err := s.Recv(0)
        if err != nil {
            panic(err)
        if msg, err := s.Recv(0); err != nil {
            panic(err)
        } else {
            contents := Message{}
            if err := json.Unmarshal([]byte(msg), &contents); err != nil {
                panic(err)
            fmt.Println("Received message from " + address + " channel.")
            fmt.Printf("%+v\n", contents)
    
As always, I hope you found it interesting. If you noticed any errors in this article, please mention them in the comments. 🧑🏻‍💻
Hope you have a great day! 🤪
      
Built on Forem — the open source software that powers DEV and other inclusive communities.
      
Made with love and Ruby on Rails. DEV Community © 2016 - 2025.