Примеры сценариев использования очереди

Ниже приведены примеры по созданию приложений-клиентов TQE(MQ), которые позволят публиковать сообщения в очередь и подписываться на них по протоколу gRPC с использованием персистентной подписки. Примеры реализованы на языке go.

• Подписка на очередь.
• Публикация сообщений.

Для дальнейшей сборки примеров потребуются следующие установленные и подготовленные компоненты:

• go версии > 1.21.1.
• protoc версии > 26.1.
• protoc-gen-go версии > 1.36.1.
• proto-gen-grpc версии > 1.2.0.
• proto-файлы с описанием gRPC-протокола TQE(MQ).

1. Создайте каталог с go-проектом, в котором в дальнейшем будут размещаться примеры клиентов:

 
mkdir mqee-clientscd mqee-clientsgo mod init example.com/mqee-clients

2. Установите зависимости для сборки примеров клиентов очереди TQE(MQ):

 
go get google.golang.org/grpcgo get google.golang.org/protobuf

Для генерации gRPC-клиента потребуются компилятор protoc и плагины для компиляции go-файлов.

1. Создайте каталог для хранения необходимых для генерации gRPC-клиента исполняемых файлов:

 
mkdir -p bin

2. Скачайте и установите дистрибутив protoc, указав версию, тип операционной системы и архитектуру:

 
PB_REL="https://github.com/protocolbuffers/protobuf/releases"VERSION=26.1OS=linuxARCH=x86_64mkdir -p tmpcurl -o tmp/protoc.zip -L $PB_REL/download/v$VERSION/protoc-$VERSION-$OS-$ARCH.zipunzip -j ./tmp/protoc.zip bin/protoc -d bin

3. Установите плагины для protoc:

 
GOBIN=$(pwd)/bin go install -ldflags '-w -s' google.golang.org/protobuf/cmd/protoc-gen-go@v1.36.0GOBIN=$(pwd)/bin go install -ldflags '-w -s' google.golang.org/grpc/cmd/protoc-gen-go-grpc@v1.2.0

4. Для создания примеров клиентов очереди необходимо получить архив с .proto-файлами, которые поставляются в составе основного пакета TQE(MQ). Полученный архив распакуйте во временный каталог:

 
tar xfzv tmp/protobuf-message-queue-ee-v3.3.0.src.tar.gz -C tmp

5. Предварительно для генерации go-файлов с использованием protoc установите каталог bin в переменную окружения PATH:

 
export PATH=$PWD/bin:$PATH

6. Сгенерируйте go-файлы проекта:

 
mkdir -p protocolprotoc \ --proto_path=$(pwd)/tmp/message-queue-ee/include/tarantool/message_queue_ee/ \ --go_out=protocol \ --go_opt=default_api_level=API_OPAQUE \ --go_opt=module=gitlab.vkteam.ru/tarantool/tqe/message-queue.git/v3/server/protocol \ --go-grpc_out=protocol \ --go-grpc_opt=module=gitlab.vkteam.ru/tarantool/tqe/message-queue.git/v3/server/protocol \ services/consumer.proto services/producer.proto messages/message.proto

1. Подготовьте исходный файл примера клиента сервиса подписки на очередь сообщений:

 
mkdir -p cmd/consumertouch cmd/consumer/main.go

Содержимое файла:

 
package mainfunc main() {}

2. Сначала необходимо создать транспорт для работы gRPC-соединения. Добавьте этот транспорт в функцию main, указав при этом адрес подключения к gRPC-интерфейсу очереди:

 
ctx := context.Background()conn, err := grpc.DialContext(	ctx,	"localhost:18182",	grpc.WithTransportCredentials(insecure.NewCredentials()),)if err != nil {	panic(err)}defer conn.Close()

3. Далее необходимо создать сам клиент сервиса подписки и отправить запрос на подписку на сообщения из очереди queue. Для использования персистентной подписки нужно указать её идентификатор и время жизни:

 
client := protocol.NewConsumerClient(conn)subscribe, err := client.Subscribe(ctx)if err != nil {	panic(err)}subreq := &protocol.SubscriptionRequest{}subreq.SetQueue("queue")subreq.SetConsumeId("example-consume-id")subreq.SetTtl(float32(time.Hour.Seconds()))streamreq := new(protocol.SubscriptionStreamRequest)streamreq.SetSubscribeRequest(subreq)if err := subscribe.Send(streamreq); err != nil {	panic(err)}fmt.Println("Start listening...")

4. После подключения необходимо запустить обработку входящих сообщений. Для этого создается среда выполнения Go, которая будет, которая будет получать сообщения из gRPC-потока, выводить краткое уведомление и обновлять состояние подписки:

 
go func() {	for {		recv, err := subscribe.Recv()		if err != nil {			panic(err)		}		switch recv.WhichResponse() {		case protocol.SubscriptionStreamResponse_Notifications_case:			nresp := recv.GetNotifications()			for _, notification := range notifResp.GetNotifications() {				cursor := notification.GetCursor()				message := notification.GetMessage()				fmt.Printf("Notification:\n\tQueue: %s\n\tCursor: %s\n\tId: %d\n\tPayload: %s\n\n",					message.GetQueue(), cursor, message.GetId(), message.GetPayload())								comreq := new(protocol.CommitRequest)				comreq.SetCursor(cursor)				streamreq := new(protocol.SubscriptionStreamRequest)				streamreq.SetCommitRequest(comreq)				if err := stream.Send(streamreq); err != nil {					panic(err)				}			}			fmt.Println("Messages received")		case protocol.SubscriptionStreamResponse_CommitResponse_case:			resp := recv.GetCommitResponse()			cursor := resp.GetCursor()			fmt.Printf("Consumer state has updated by %s cursor\n", cursor)		}	}}()

5. Добавьте обработку системных сигналов:

 
notifyCtx, _ := signal.NotifyContext(ctx, syscall.SIGINT, syscall.SIGTERM)<-notifyCtx.Done()fmt.Println("Exit")

Клиент сервиса подписки на сообщения создан.

6. Выполните запуск клиента сервиса подписки на сообщения:

 
$ go run example.com/mqee-clients/cmd/consumerStart listening...

После этого клиент станет ожидать появления новых сообщений в очереди.

1. Создайте файл, который будет содержать исходный код клиента сервиса для публикации сообщений в очередь:

 
mkdir -p cmd/producertouch cmd/producer/main.go

Первоначальное содержимое файла:

 
package mainfunc main() {}

2. Создайте транспорт для работы gRPC-соединения. Добавьте этот транспорт в функцию main, указав при этом адрес подключения к gRPC-интерфейсу очереди:

 
func main() {	// ...	ctx := context.Background()	conn, err := grpc.DialContext(		ctx,		"localhost:18182",		grpc.WithTransportCredentials(insecure.NewCredentials()),	)	if err != nil {		panic(err)	}	defer conn.Close()	// ...}

3. Создайте сам клиент сервиса публикации для дальнейшей отправки сообщений в очередь:

 
client := protocol.NewProducerClient(conn)

4. Отправьте запрос на публикацию сообщения с помощью вызова метода Produce:

 
responseBatch, err := client.Produce(ctx, &protocol.ProduceRequest{	Queue: "queue",	Messages: []*protocol.ProduceMessage{		{Payload: []byte("First message from batch")},		{Payload: []byte("Second message from batch")},	},})if err != nil {	panic(err)}fmt.Printf("Messages published with ids: %v\n", responseBatch.Ids)

Теперь при запуске клиента будет отправлено два сообщения с помощью вызова метода Produce:

 
$ go run example.com/mqee-clients/cmd/producerMessages published with ids: [115720 115721]

Если в другом окне терминала запущен клиент-подписчик (cmd/consumer/main.go), то вы получите уведомление о новых сообщениях:

 
Notification:	Queue: queue	Cursor: DX8EAQL/gAABDAEGAAAS/4AAAQlzdG9yYWdlLTH9AcQR	Id: 115729	Payload: First message from batchConsumer state has updatedNotification:	Queue: queue	Cursor: DX8EAQL/gAABDAEGAAAS/4AAAQlzdG9yYWdlLTH9AcQS	Id: 115730	Payload: Second message from batchConsumer state has updated