お題

Cloud Pub/Sub使ってますか？
大量のメッセージを捌きたい時に使いますね。
これ、メッセージ投入時の順序、保証したい時、ないですか？
AWSの場合であればSQSのFIFOキューを使うのかなと思います。
（過去に使おうとしたことがあったものの、まだ東京リージョンなかったのと１秒に最大3000メッセージという制約に尻込みして、結局使わなかった記憶が。。。）
SQSに相当するといったらGCPではCloud Pub/Subだと思いますが、残念ながらFIFOキューは無い。
と思っていたら、
（Beta版ですが）「メッセージの順序指定」という機能が出来ていたようで。
これで、メッセージがPublishした順にSubscribeできるのかと思い、試してみました。

とはいえ、だいぶアバウトな試し方なので、とりあえず、こういう試し方をしたところ、こういう結果になったというレベルの捉え方をしてもらえれば。

想定する読者

GCPについては知っている。
Cloud Pub/Subを使ったことがある。ないし、おおよそどういうものかは知っている。
Golangもそれなりに書ける。

前提

ローカルにGoの開発環境構築済み。
GCP契約済み。
ローカルでCloud SDKのセットアップ済み。
ローカルの環境変数GOOGLE_APPLICATION_CREDENTIALSに（必要な権限を全て有したサービスアカウントの）鍵JSONファイルパス設定済み。

開発環境

# OS - Linux(Ubuntu)

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

# バックエンド

# 言語 - 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

今回の全ソース

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

ソース抜粋解説

go-publisher

５つのエンドポイントを用意。
topic.EnableMessageOrdering = trueとOrderingKey: operationSequence,の部分は、実は「メッセージの順序指定」のために必要なコード。
ただし、このソースでメッセージをPublishするTopic（"my-normal-topic"）は「メッセージの順序指定」用ではないSubscriptionと対応しているので、「メッセージの順序指定」は機能しない。

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
}

５つのエンドポイントを順繰り叩くシェル

かなり雑なつくり。

#!/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