Can FIFO queues be realized by "Specifying message order" in Cloud Pub / Sub?

theme

Do you use Cloud Pub / Sub? It is used when you want to handle a large number of messages. Is this the order in which messages are sent, when you want to guarantee it, or not? In the case of AWS, I think you should use SQS's FIFO Queue. (I had tried to use it in the past, but I remember that I didn't use it after all because I didn't have the Tokyo region yet and I was squeezed by the limit of 3000 messages per second. I think that it is Cloud Pub / Sub in GCP if it corresponds to SQS, but unfortunately there is no FIFO queue. If you think (Beta version) It seems that the function "Specify message ordering" was created. I wondered if I could subscribe to the messages in the order they were published, so I tried it.

However, it's a fairly about trial method, so for the time being, if you try this kind of trial, you can get a level of understanding that this is the result.

Assumed reader

――I know about GCP. --I have used Cloud Pub / Sub. I don't know what it is. --Golang can be written as it is.

Premise

--Go development environment has been built locally. --GCP contract completed. --Cloud SDK has been set up locally. --The key JSON file path (of the service account with all the required permissions) has been set in the local environment variable GOOGLE_APPLICATION_CREDENTIALS.

Development environment

OS - Linux(Ubuntu)

$ cat /etc/os-release 
NAME="Ubuntu"
VERSION="18.04.5 LTS (Bionic Beaver)"

#Backend

#Language --Golang

$ go version
go version go1.15.2 linux/amd64

IDE - Goland

GoLand 2020.2.3
Build #GO-202.7319.61, built on September 16, 2020

All sources this time

https://github.com/sky0621/go-publisher/tree/v0.1.0 https://github.com/sky0621/go-subscriber/tree/v0.1.0

Source excerpt commentary

go-publisher

Prepare 5 endpoints. The parts of topic.EnableMessageOrdering = true and ʻOrderingKey: operationSequence, are actually the codes required for "ordering messages". However, Topic (" my-normal-topic "`), which publishes messages from this source, corresponds to Subscriptions that are not for "message ordering", so "message ordering" does not work.

main.go

package main

import (
	"fmt"
	"log"
	"net/http"
	"os"
	"time"

	"cloud.google.com/go/pubsub"
	"github.com/labstack/echo/v4"
	"github.com/labstack/echo/v4/middleware"
)

func main() {
	project := os.Getenv("PUB_PROJECT")

	e := echo.New()
	e.Use(middleware.Logger())
	e.Use(middleware.Recover())

	e.GET("/order01", handler(project, "order01"))
	e.GET("/order02", handler(project, "order02"))
	e.GET("/order03", handler(project, "order03"))
	e.GET("/order04", handler(project, "order04"))
	e.GET("/order05", handler(project, "order05"))

	e.Logger.Fatal(e.Start(":8080"))
}

func handler(project, path string) func(c echo.Context) error {
	return func(c echo.Context) error {
		ctx := c.Request().Context()
		operationSequence := createOperationSequence()

		client, err := pubsub.NewClient(ctx, project)
		if err != nil {
			log.Fatal(err)
		}

		topic := client.Topic("my-normal-topic")
		defer topic.Stop()

		topic.EnableMessageOrdering = true

		message := &pubsub.Message{
			OrderingKey: operationSequence,
			Data:        []byte(path + ":" + operationSequence),
		}
		r := topic.Publish(ctx, message)
		if r == nil {
			log.Fatal("failed to topic.Publish!")
		}
		log.Printf("%+v", r)

		return c.String(http.StatusOK, path+":"+operationSequence)
	}
}

func createOperationSequence() string {
	return fmt.Sprintf("%d", time.Now().UnixNano())
}

go-subscriber

main.go

package main

import (
	"encoding/json"
	"io"
	"io/ioutil"
	"log"
	"net/http"

	"github.com/labstack/echo/v4"
	"github.com/labstack/echo/v4/middleware"
)

func main() {
	e := echo.New()
	e.Use(middleware.Logger())
	e.Use(middleware.Recover())

	e.POST("/", func(c echo.Context) error {
		m, err := unmarshal(c.Request().Body)
		if err != nil {
			return c.String(http.StatusBadRequest, err.Error())
		}
		data := string(m.Message.Data)
		log.Printf("fs_Received__Data:%s", data)
		return c.String(http.StatusOK, "OK")
	})

	e.Logger.Fatal(e.Start(":8080"))
}

type PubSubMessage struct {
	Message struct {
		Data []byte `json:"data,omitempty"`
		ID   string `json:"id"`
	} `json:"message"`
	Subscription string `json:"subscription"`
}

func unmarshal(r io.ReadCloser) (*PubSubMessage, error) {
	var m PubSubMessage
	body, err := ioutil.ReadAll(r)
	if err != nil {
		log.Printf("ioutil.ReadAll: %v", err)
		return nil, err
	}
	if err := json.Unmarshal(body, &m); err != nil {
		log.Printf("json.Unmarshal: %v", err)
		return nil, err
	}
	return &m, nil
}

A shell that hits 5 endpoints in sequence

It's pretty rough.

#!/usr/bin/env bash
set -euox pipefail

for i in {0..15}
do
	for j in {1..5}
	do
		curl "https://go-publisher-xxxxxxxxxxxxxx/order0${j}"
		sleepenh 0.005
	done
	sleepenh 0.005
done

Practice

image

Put a service with a REST mouth on Cloud Run, publish it to Cloud Pub / Sub when access comes, and the Subscription prepared with Push type is also a REST mouth put on Cloud Run Skip the request to the service with.

For regular Topic and Subscription

Topic and Subscription settings

Message input result

The messages are repeatedly hitting the endpoints in the following order, and although they are in that order on the Publisher log, they are in no particular order on the Subscriber log. /order01 /order02 /order03 /order04 /order05

Of course, it is in order on the log issued by the shell that hits the endpoint.

For Topic and Subscription that support "Specify message order"

Topic and Subscription settings

Since there is a condition that exists in the same region, specify the Topic region.

Of course, "Specify message order" should be "Enabled".

Message input result

After trying it several times, there were some cases where the order of subscribing was changed at the start, but after that, the order was correct. Compared to the normal Topic / Subscription combination, which was rather out of order, the order is stable.

Summary

Even in the "message order specification" mode, the order was changed, so I couldn't say "it can be realized" for the subject. .. .. Whether this is a matter of how to try it, or maybe it's the Beta version, if you feel like it, let's dig into it. .. ..