Pub/Sub with Go-Micro

Here’s a small tutorial on how to do pub/sub with RabbitMQ in golang using go-micro which is a pluggable RPC framework for microservices.

We’ll be creating a service which creates two subroutines. One will be listening on a routing key on a rabbitmq exchange, while another writes to the exchange every second.

Here’s your main.go

package main

import (
  "fmt"
  "log"
  "time"

  "github.com/micro/go-micro/broker"
  "github.com/micro/go-micro/cmd"
  _ "github.com/micro/go-plugins/broker/rabbitmq"
)

func pub() {
  cmd.Init()

  if err := broker.Init(); err != nil {
    log.Fatalf("Broker Init error: %v", err)
  }
  if err := broker.Connect(); err != nil {
    log.Fatalf("Broker Connect error: %v", err)
  }

  tick := time.NewTicker(time.Second)
  i := 0
  for _ = range tick.C {
    msg := &broker.Message{
      Header: map[string]string{
        "id": fmt.Sprintf("%d", i),
      },
      Body: []byte(fmt.Sprintf("%d: %s", i, time.Now().String())),
    }
    if err := broker.Publish("service.topic", msg); err != nil {
      log.Printf("[pub] failed: %v", err)
    } else {
      fmt.Println("[pub] pubbed message:", string(msg.Body))
    }
    i++
  }
}

func sub() {
  cmd.Init()

  if err := broker.Init(); err != nil {
    log.Fatalf("Broker Init error: %v", err)
  }
  if err := broker.Connect(); err != nil {
    log.Fatalf("Broker Connect error: %v", err)
  }

  _, err := broker.Subscribe("service.topic", func(p broker.Publication) error {
    fmt.Println("[sub] received message:", string(p.Message().Body), "header", p.Message().Header)
    return nil
  })
  if err != nil {
    fmt.Println(err)
  }
}

func main() {
  forever := make(chan struct{})

  go pub()
  go sub()

  <-forever
}

Let’s use docker to create the service and dependent servers like rabbitmq and consul which is need by go-micro.

Here’s a simple Dockerfile

FROM golang

COPY . /go/src/github.com/my-repo/pubsub
WORKDIR /go/src/github.com/my-repo/pubsub

RUN go get ./...
RUN go build
CMD pubsub

Here’s the docker-compose.yml file I used where we can use environment variables to set the rabbitmq connection details.

version: "3"
services:
  services:
    build: .
    ports:
      - "8080:8080"
    environment:
      - MICRO_REGISTRY_ADDRESS=consul:8500
      - MICRO_BROKER_ADDRESS=amqp://admin:password@rabbit:5672/
      - MICRO_BROKER=rabbitmq
    depends_on:
      - consul
      - rabbit

  consul:
    image: progrium/consul:latest
    command: -server -bootstrap -rejoin
    ports:
      - "8500"

  rabbit:
    image: rabbitmq:3-management
    ports:
      - "5672"
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: password

If you prefer to not use environment variables for setting the rabbitmq address or exchange value, you can do it on the code by instantiating a new rabbitmq broker like this

r := rabbitmq.NewBroker(micro.Addrs("amqp://admin:password@rabbit:5672/"))
//For setting exhange do
r := rabbitmq.NewBroker(
    rabbitmq.Exchange("foo"),
)
//Or
rabbitmq.DefaultExchange = "foo"