Przesyłanie danych pomiędzy serwisami może odbywać się zarówno synchronicznie, jak i asynchronicznie. Możemy wysyłać dane także w wielu różnych formatach, spośród których najpopularniejszymi są Protocol Buffers oraz JSON (ten pierwszy jest aktualnie bardziej powszechny, jeśli wiemy, że żaden serwis komunikujący się z naszym nie jest napisany w języku, który jeszcze nie ma wsparcia dla tego formatu).

Protocol Buffers

Protocol Buffers, tak samo jak język Go, jest rozwijany przez firmę Google, jednak nie jest on ściśle związany z Golang. Jest on językowo–neutralny, mimo że sposób definiowania schematów danych może wydać się nam znajomy.

Głównymi atutami Protobuf jest jego mały rozmiar, a także wsteczna kompatybilność, jeśli wiemy jak go używać – jeśli nie, łatwo pozbawić się tej dogodności. Spójrzmy na przykład definiowania struktury, która będzie dla nas przechowywać informacje o koncie:

message Account {
    string id = 1;
    string username = 2;
    string email = 3;
    Type type = 10;

    enum Type {
        TYPE_UNSPECIFIED = 0;
        TYPE_WEB_ACCOUNT = 1;
        TYPE_MOBILE_ACCOUNT = 2;
    }
}

Jak widzimy, takie zdefiniowane wiadomości wygląda bardzo prosto. Numerowanie pól dostępnych w wiadomości zaczynamy od numeru 1 i dodajemy kolejne. Dodawanie cyfr przy definicji pola pozwala zmniejszyć znacznie rozmiar wiadomości oraz utrzymać wspomnianą wcześniej wsteczną kompatybilność.

Zastosowanie wstecznej kompatybilności najłatwiej pokazać przez kolejny przykład zaktualizowanej wiadomości.

message Account {
    string id = 1;
    string name = 2;
    string email = 3;
    string phone_number = 4;
    Type type = 10;

    enum Type {
        TYPE_UNSPECIFIED = 0;
        TYPE_WEB_ACCOUNT = 1;
        TYPE_MOBILE_ACCOUNT = 2;
    }
}

Zmieniliśmy nazwę pola numer 2 z username na name oraz dodaliśmy nowe pole numer 4, co ma również związek z tym, że nasze pole type miało cały czas cyfrę 10. W tym momencie, jeśli na naszej kolejce mamy wiadomość, która cały czas przechowuje stary format wiadomości, będziemy mogli mimo wszystko odczytać ją z sukcesem (mimo że na przykład serwis wysyłający wiadomość dalej użył pola nazwanego username, a nasz konsumujący wiadomość użył pola name). Wszystko dzięki temu, że w wysłanej wiadomości mamy informacje tylko o numerze pola, a nie o jego nazwie.

Dzięki temu, że podaliśmy numer 10 przy polu type, możemy grupować nasze pola, choć ma to wyłącznie aspekt wizualny i nie ma większego wpływu na sposób działania. W naszym przykładzie wszystkie podstawowe informacje o koncie będą ponumerowane od 1 do 9, natomiast dodatkowe informacje konfiguracyjne będą zaczynały się od 10.

Generowanie kodu w Go

Aby nasz stworzony Protobuf działał w jakimkolwiek języku, musimy go najpierw w nim wygenerować. Do dyspozycji mamy dużą liczbę generatorów  – do Javy, PHP i oczywiście do Go, który omawiamy, w tym artykule.

Aby móc wygenerować nasz kod pobieramy i zgodnie z instrukcją instalujemy grpc-go (https://github.com/grpc/grpc-go).

Uruchomienie samego generatora jest bardzo proste. Wystarczy, że uruchomimy komendę:

“protoc -I /usr/include/google/protobuf -I . –go_out=plugins=grpc:output ./our_service.proto” 

I jeśli tylko nie popełniliśmy żadnego błędu w definicjach, to otrzymamy gotowy plik z końcówką .go, który możemy zaimportować do naszego kodu. Najlepszym rozwiązaniem jest udostępnienie tej paczki w naszym prywatnym repozytorium, tak aby każdy serwis mógł z łatwością załadować nową wersję paczki. Sam Protobuf też może mieć mniejsze paczki, dzięki temu możemy wydzielić na przykład wspólne ustawienia paginacji, których będziemy potem używać w osobnych definicjach dla naszych innych projektów.

Porównanie z JSON

Poza Protocol Buffers w naszych serwisach używamy też formatu JSON. Ma to głównie związek z zewnętrznymi serwisami i jesteśmy głównie konsumentem tego formatu, poza sytuacjami, w których musimy wysłać wiadomość zwrotną lub też symulować zewnętrzną wiadomość. Wtedy wysyłamy wiadomości w formacje JSON, co za tym idzie, nie zrezygnowaliśmy całkiem z tego formatu, chociaż wewnętrznie wolimy jednak Protobuf.

Poza wspomnianą wcześniej kompatybilnością wsteczną Protobuf wygrywa też rozmiarem wiadomości. Rezultaty, które można uzyskać to nawet 25% rozmiaru takiej samej wiadomości w JSON – ma to oczywiście związek z niewysyłaniem nagłówków pól w wiadomości. Rozmiar takiej wiadomości jest o wiele większy, ma to szczególnie znaczenie w komunikacji przez kolejki, w której (między innymi w AWS) mamy limit rozmiaru wiadomości.

Komunikacja asynchroniczna

Komunikacja asynchroniczna to coś, czego jestem zazwyczaj ogromnym zwolennikiem, co prawda raczej unikam jeszcze stosowania jej przy przekazywaniu danych pomiędzy frontendem a backendem, jednak pomiędzy serwisami w backendzie pozwala ona zaoszczędzić sporo czasu (oraz nerwów!) na debugowaniu. Oczywiście wszystko jak zawsze jest kwestią implementacji.

Subskrypcje i kolejki

W naszych serwisach wewnętrznie używamy kolejek oraz subskrypcji AWS, gdzie sprawa jest dosyć prosta – tworzy się kolejkę (Queue) i wysyła do niej wiadomości za pomocą AWS SDK dla Go lub też tworzy się subskrypcję (Subscription) dla tematu (Topic) i wiąże ją z kolejką. Takie rozwiązanie ma sporo plusów, niezależnie czy wysyła się wiadomość do jednej kolejki, czy do kilku na raz – konfiguracja wygląda tak samo.

Inaczej sprawa ma się z Azure Service Bus. Subskrypcja jest już sama w sobie kolejką, nie trzeba jej wiązać z istniejącymi kolejkami. Jednakże już zmiany na konsumowanie lub produkowanie wiadomości z subskrypcji na kolejkę w ASB nie są takie proste i wymagają kilku zmian w kodzie, bądź też kilku strategii, które będzie można sobie przełączyć za pomocą jakiegoś parametru. ASB używany jest przez nas głównie do komunikacji z zewnętrznymi serwisami. W ramach ciekawostki wspomnę, że jeden z naszych serwisów w PHP (ze względu na słabe wówczas wsparcie dla ASB w PHP) odczytywał wiadomości z kolejki AWS, które trafiały tam uprzednio przekazane przez nasz serwis w Go, który z kolei wcześniej odczytywał je z kolejki w Azure. Trochę zagmatwane, prawda? 🙂

Rozmiar wiadomości

Przy używaniu kolejek trzeba pamiętać o rozmiarze. Maksymalny rozmiar wiadomości w AWS SQS wynosi 256 KB i aktualnie tylko Java ma dodatkowo rozszerzonego klienta, który pozwala na wysłanie nawet 2GB wiadomości. Jednak jest do tego używany także serwis AWS S3, gdzie zawartość tej wiadomości jest przechowywana.

W Go też jest bardzo prosta możliwość wysyłania większych wiadomości, również z użyciem S3. Jednakże w przypadku dużej liczby serwisów, które mogą wysyłać duże wiadomości, warto pomyśleć o dodatkowym serwisie, który będzie w tym pośredniczył. Cierpliwości, więcej powiem o tym za chwilę 🙂 

Kolejność wiadomości

Kolejną bardzo ważną rzeczą, o której trzeba pamiętać, jest kolejność wiadomości. To problem istniejący bardziej w samej warstwie aplikacji niż w usłudze kolejkowania, jednak w większości przypadków może nam przysporzyć niemało kłopotów.

Zależnie od systemu, którego używamy do kolejkowania, możemy mieć różne ustawienia. Ja w naszym przypadku posłużę się AWS SQS z domieszką SNS, który znam najlepiej. Domyślnie SQS próbuje przekazywać nam wiadomości z kolejki zgodnie z zasadą first–in–first–out, czyli wiadomość, która wpadła pierwsza na kolejkę, jest z niej odczytywana również jako pierwsza. W praktyce zgodnie z dokumentacją usługi, wiadomości mogą się nam trochę pomieszać, szczególnie jeśli nasz serwis konsumujący wiadomości ma jednocześnie uruchomione kilka procesów. Rozwiązaniem tego problemu jest włączenie FIFO. Odbywa się to poprzez dodanie .fifo na końcu nazwy kolejki i to tak naprawdę tyle. Od teraz możemy być spokojniejsi o kolejność wiadomości. Co ciekawe, do niedawna FIFO nie działało dla połączenia SNS z SQS, ale od niespełna roku takie połączenie jest już dostępne.

Pytanie tylko, czy samo włączenie FIFO zawsze wystarczy? W przypadku mniej skomplikowanych serwisów być może tak, jednak przy bardziej złożonych systemach dobrze zadbać o to, żeby (o ile to możliwe) kontrolować przepływ wiadomości i jeśli coś wydaje się nam podejrzane wyrzucać błąd. Można to osiągnąć między innymi przez implementacje statusów oraz uprawnień do tego jakie statusy mogą być zastępowane. Jeszcze bardziej złożone systemy mogą od nas wymagać wdrożenia maszyny stanów i konkretnego określenia co może dziać się w konkretnych momentach. Do stworzenia takiej maszyny stanów możemy użyć paczki fsm.

Dodatkowo może nam się przydać możliwość zablokowania zmiany konkretnego zasobu. W przypadku naszych serwisów poza użyciem samych transakcji w MariaDB zdecydowaliśmy się także na blokowanie konkretnego zasobu przez jego zapis w DynamoDB. Użyliśmy do tego paczki dynamolock, którą nieco zapakowaliśmy i dostosowaliśmy do dodatkowych wymagań. Dzięki temu widzimy na przykład dodatkowe zabezpieczenie przed utworzeniem dwóch takich samych rekordów płatności w tej samej sekundzie i obniżeniem wartości faktury dwukrotnie.

Rozwiązania dla Go

Jeśli chodzi o implementację jakiegokolwiek systemu kolejek w Go to nie znam paczki, która mógłbym polecić i która obsługiwała by większość popularnych systemów. W projekcie mamy napisaną własną paczkę, której używamy w naszych serwisach i jeśli zachodzi taka potrzeba, implementujemy w niej nowe usługi kolejkowania. W tej chwili wspieramy SQS, ASB, RabbitMQ oraz mamy kolejkę, która przechowuje wiadomości w pamięci. Jest to kolejka używana wyłącznie w niektórych testach automatycznych.

Przykładowo implementacja SQS w Go nie powinna być skomplikowana dla osoby, która miała do czynienia ze współbieżnością. Szczególnie z kanałami (chan) i rutynami (goroutines), ponieważ sama komunikacja z AWS odbywa się za pomocą SDK to po wczytaniu wiadomości za pomocą funkcji ReceiveMessage, pozostaje nam tylko zrzucenie jej na kanał i obsłużenie. Dobre przykłady można znaleźć w oficjalnej dokumentacji AWS.

Uproszczony przykład pętli odczytującej wiadomości z wykorzystaniem przykładu z dokumentacji: 

outgoingChannel := make(chanMessage)

go func() {
       for {
           msgResult, err := svc.ReceiveMessage(&sqs.ReceiveMessageInput{
              AttributeNames: []*string{
                 aws.String(sqs.MessageSystemAttributeNameSentTimestamp),
              },
              MessageAttributeNames: []*string{
                 aws.String(sqs.QueueAttributeNameAll),
              },
              QueueUrl:            queueURL,
              MaxNumberOfMessages: aws.Int64(1),
              VisibilityTimeout:   timeout,
           })

              for _, message := range output.Messages {
                 outgoingChannel <- messageAdapter(message)
              }
         }
}()

return (<-chan Message)(outgoingChannel)

To, o czym warto pamiętać przy implementacji, to także ustawienie odpowiedniego czasu odczytu wiadomości. Najważniejsze, aby nie był on zbyt krótki w przypadku używania kilku procesów współbieżnie, ponieważ może nam to spowodować odczytanie tej samej wiadomości dwa razy. Standardowo na odczyt mamy 30 sekund i powinno nam to wystarczyć, jeśli tylko nie blokujemy z jakiegoś powodu przetwarzania wiadomości. 

Inną ważną rzeczą jest też wysłanie potwierdzenia odbioru wiadomości do kolejki, dzięki czemu zostanie ona z niej usunięta. W przypadku SQS jest to po prostu DeleteMessage.

Case study: wysyłanie maili z załącznikami

Na zakończenie przedstawię krótkie case study z jednego z naszych projektów, Jest to temat, z którym możemy się spotkać przy wdrażaniu komunikacji przez kolejki w wielu różnych sytuacjach.

Jak już wspominałem wcześniej, systemy kolejkowania wiadomości mają limit rozmiaru. W przypadku AWS SQS 256 KB, jest to oczywiście rozsądne podejście, w końcu odczytanie wiadomości ważącej 1GB mogłoby zająć o wiele za długo. Jeśli mamy jeden serwis, który tworzy faktury i drugi, który wysyła wiadomości e–mail, jeśli będziemy chcieli wysłać dość duży plik PDF z fakturą, możemy natrafić na problem.

Co może być rozwiązaniem? Oczywiście dodatkowy serwis. Ok… może nie zawsze 😉 Jeśli mamy mały zespół to duża liczba serwisów może w końcu być przytłaczająca. Dlatego zawsze warto poszukać innych rozwiązań, takich jak wrzucanie pliku PDF bezpośrednio na AWS S3, znaleźć inne miejsce do przechowywania plików, lub też dodatkową funkcję w naszym serwisie wysyłającym maile do wrzucania załączników przez gRPC, lub inne API.

W naszym przypadku jest to osobny serwis. Dodatkowym argumentem “za” była też możliwość zastosowania go w innych serwisach, niekoniecznie tylko do wysyłania załączników. Przewaga takiego rozwiązania nad bezpośrednim połączeniem z AWS S3 jest ogromna. Głównie przez to, że nie musimy w każdym serwisie podawać informacji o miejscu przechowywania plików oraz dbać o aktualną wersję SDK. Poza tym w przypadku zmiany usługodawcy z AWS S3 na Azure Blob Storage musimy to zmienić tylko w jednym serwisie, pozostałe serwisy po prostu używają funkcji Download i Upload w wygenerowanym przez nas Protobuf i gRPC, o czym napiszę więcej w kolejnym artykule o komunikacji synchronicznej.

Podsumowując, pomiędzy naszym serwisem generującym faktury a wysyłającym maile podajemy tylko informacje, jaki szablon maila ma być użyty, do kogo ma być wysłany oraz ID i hash załącznika, po którym nasz drugi serwis może pobrać sobie załącznik.

Nasuwa się jednak pytanie – co jeśli taki załącznik jest duży, czy to nie opóźni za bardzo odczytywania wiadomości z kolejki? Oczywiście opóźni 🙂 Dlatego warto rozdzielić konsumenta wiadomości z kolejki i wysłanie żądania do AWS SES na dwie osobne komendy. Konsument wrzuca informacje o wiadomości z kolejki do bazy danych, natomiast kolejna komenda samodzielnie je sobie wyciąga, zmienia status na processing i zależnie od późniejszego statusu wiadomości zmienia na wysłaną lub odrzuconą.

Jak widać, tak prosty temat, jak komunikacja asynchroniczna w Go pomiędzy serwisami jest dość obszerny, implementacje mogą też być różne. Zależnie od wybranego systemu kolejkowania, mogą one różnić się między sobą sposobem połączenia z kolejką czy odczytywania wiadomości. Jednak po stronie Go możemy wszystko schować w implementacji za interfejsem za pomocą kanałów (chan). 

W kolejnym artykule poruszymy temat komunikacji synchronicznej, który uzupełni informacje zawarte w krótkim case study z powyższego artykułu.