Nsq – systém pro doručování zpráv bez centrálního message brokera

29. 10. 2019

Doba čtení: 34 minut

Líbí se vám článek?
Podpořte redakci

V další části seriálu (nejenom) o message brokerech se seznámíme se systémem zpráv nazvaným NSQ. Je mimochodem vyvinut v jazyce Go a je založen na myšlence decentralizace a navíc tak, aby neobsahoval centrální uzel.

Obsah

1. Systém nsq se představuje

2. Architektura systému nsq

3. Doručení zprávy konzumentům

4. Překlad a instalace systému nsq

5. Instalace knihoven používaných klienty

6. Spuštění jednotlivých uzlů systému nsq

7. Používané porty

8. Poslání zprávy přes REST API

9. Komunikace s uzlem nsqlookupd přes REST API

10. Jednoduchý producent zpráv naprogramovaný v Pythonu

11. Konzument zpráv naprogramovaný v Pythonu

12. Použití systému nsq klienty naprogramovanými v Go

13. Producent zpráv naprogramovaný v jazyce Go

14. Využití bloku defer pro ukončení producenta zpráv

15. Jednoduchý konzument zpráv používající synchronizaci pro ukončení gorutiny s příjemcem

16. Použití kanálu ve funkci synchronizační struktury

17. Připojení klienta ke službě nsqlookupd

18. Obsah následující části seriálu

19. Repositář s demonstračními příklady

20. Odkazy na Internetu

1. Systém nsq se představuje

V seriálu o message brokerech jsme se seznámili již s mnoha typy message brokerů, implementací front zpráv, komunikačních strategií typu PUB-SUB atd. Již dříve popsané systémy byly vytvořeny v různých programovacích jazycích, například populární message broker RabbitMQ v Erlangu, RQ (Redis Queue) v Pythonu (samotný Redis v jazyku C), ZeroMQ v C++, nanomsg v C, AMQ a Artemis v Javě a konečně systém NATS v programovacím jazyce Go. V Go je vytvořen i dnes popisovaný nástroj nsq, který se skládá z několika samostatně spouštěných uzlů, které mezi sebou vhodným způsobem komunikují. Výhodou a jednou ze základních vlastností nsq je fakt, že způsob jeho návrhu počítá s tím, že jednotlivé uzly mohou být po nějakou dobu nedostupné a přitom systém jako celek by měl být stále použitelný. Jedná se tedy o architekturu vhodnou pro nasazení do dnešního hybridního prostředí se systémy běžícími v cloudu, lokálně v kontejneru, na dedikovaných serverech atd. Základní vlastnosti nsq naleznete na stránce https://nsq.io/overview/features_and_guarantees.html, my se s jednotlivými body seznámíme postupně.

Obrázek 1: Logo systému nsq.

Za vznikem a vývojem systému nsq stojí programátoři z bitly (https://bitly.com/), kteří mají s podobnými systémy dlouholeté zkušenosti. Za zmínku stojí především jejich projekt simplehttp, v němž jsou fronty zpráv i komunikační strategie PUB-SUB implementovány, konkrétně ve zdrojových souborech simplequeue.c a pubsub.c. Zkušenosti s programováním v Go mají poměrně dlouhodobé, například známý blog post o použití tohoto jazyka Go Go Gadget byl publikován již před pěti lety.

2. Architektura systému nsq

Základním úkolem systému nsq je zajištění doručení zpráv od producentů ke konzumentům, a to pochopitelně bez toho, aby spolu producenti a konzumenti museli komunikovat přímo (popř. aniž by o sobě vůbec věděli – takové pevné spojení částí systému bývá příliš rigidní a více náchylné k poruchám). Systém má taktéž umožnit škálování, tj. typicky použití většího množství konzumentů pro určité typy zpráv (rozlišených podle svého tématu – topic). Navíc musí systém umožnit, aby byly zprávy uchovány (ve frontách) ve chvíli, kdy není žádný konzument schopný zprávy přečíst a zpracovat (ideálně se zajištěním perzistence zpráv). Tradičně jsou tyto požadavky vyřešeny klasickým message brokerem, což bývá (jediný) proces s několika frontami zpráv, přičemž každá fronta je určena pro přeposílání zpráv patřících k jednomu tématu. Message broker navíc bývá doplněn i o možnost posílat zprávy s využitím komunikační strategie PUB-SUB, tedy všem konzumentům, kteří dané téma odebírají a současně jsou v daný okamžik připojeni a schopni přijmu zpráv.

Obrázek 2: Interní konfigurovatelná struktura systému RabbitMQ, který se v některých ohledech podobá dnes popisovanému systému nsq.

Systém nsq je oproti klasickým message brokerům navržen nepatrně odlišným způsobem, protože obsahuje tři základní typy uzlů, přičemž každý uzel je představován samostatným procesem (ten typicky běží na pozadí). Tyto procesy mohou běžet na jediném počítači nebo mohou být distribuovány a škálovány na větší množství strojů, a to pochopitelně i (v případě potřeby) geograficky oddělených. Jedná se o tyto uzly:

nsqlookupd – jedná se o takzvanou directory service, tj. o službu, do které se registrují všechny ostatní uzly a která tak má přehled o umístění (adresa+číslo portu) i o stavu jednotlivých uzlů. Konzumenti typicky nejprve kontaktují právě službu nsqlookupd, aby zjistili, které brokery jim jsou schopny dodávat zprávy požadovaných témat (topic). Aby byla zaručena funkcionalita celého systému, je možné, aby bylo současně spuštěno více těchto služeb. Klienti (konzumenti) se pak mohou pokusit připojit k více službám.
nsqd – tyto uzly, jichž může běžet libovolné množství, zajišťují vlastní příjem zpráv, jejich ukládání do front a následné doručení konzumentům. Zprávy jsou perzistentní, tj. ve chvíli, kdy je uzel nsqd z nějakého důvodu zastavován, jsou zprávy uloženy do souboru a po znovuspuštění uzlu jsou ze souboru načteny do paměti a popř. zaslány konzumentům. Je možné mít spuštěno větší množství nsqd, které mohou být nastaveny takovým způsobem, že budou uchovávat a poskytovat stejné zprávy (téma+kanál). V takovém případě se konzument může pokusit získat zprávy z libovolného (právě dostupného) nsqd, popř. se připojit k jinému nsqd v případě nějakého výpadku.
nsqadmin – tato služba poskytuje (webové) uživatelské rozhraní, ze kterého je možné sledovat činnost celého systému. Samotné sledování je primárně založeno na komunikaci s prvním typem uzlu – nsqlookupd a sekundárně s uzly nsqd.

Poznámka: celá architektura sice může na první pohled vypadat zbytečně složitě, ovšem v té nejjednodušší možné (a současně i prakticky využitelné) konfiguraci postačuje na stejném počítači pustit jedinou instanci každé výše zmíněné služby; dokonce je možné použít implicitní přednastavené porty, o nichž se zmíníme v navazujících kapitolách. Pokud naopak budete vyžadovat vysokou dostupnost, je možné jednotlivé služby spustit na různých počítačích/clusterech, což do značné míry udělá systém odolný vůči výpadku jednotlivých uzlů (podrobnosti si řekneme příště, protože existuje hned několik způsobů, jak architekturu navrhnout a použít).

Obrázek 3: Webové uživatelské rozhraní vytvářené uzlem typu nsqadmin.

3. Doručení zprávy konzumentům

Podívejme se nyní na způsob, jakým je zpráva doručena od producenta ke konzumentovi (či při správném nastavení k více konzumentům). Zprávou je v systému nsq myšlena prakticky libovolně dlouhá sekvence bajtů označená časovým razítkem – může se tedy jednat jak o zprávu textovou, tak i o obecná binární data, JSON, YAML, XML atd. Každá zpráva má navíc přiřazeno téma (topic), což je libovolný řetězec. Samotná témata není zapotřebí dopředu vytvářet, protože první poslaná zpráva s daným tématem toto téma založí automaticky. Zpráva je poslána určitému nsqd, popř. většímu množství nsqd – záleží, které nsqd dokážou zpracovat správu se zvoleným tématem. Po přijetí zprávy systémem je tato zpráva rozeslána do všech kanálů (topic), které pracují se zvoleným tématem. Toto je důležité chování – díky rozeslání do více kanálů je umožněn multicast, ovšem navíc to znamená, že jedna zpráva může být zpracována vícekrát, s čímž musí konzumenti počítat (měli by být idempotentní – https://en.wikipedia.org/wiki/Idempotence).

Konzumenti se připojují k určitým kanálům (a pochopitelně odebírají jen zprávy ke zvoleným tématům), přičemž je možné, aby se k jednomu kanálu připojilo větší množství konzumentů. V takovém případě jsou zprávy distribuovány mezi připojené konzumenty na základě jejich dostupnosti a rychlosti, s níž jsou schopni zpracování zpráv.

Poznámka: na kanál (channel) se můžeme dívat jako na samostatně pracujícího message brokera v klasickém slova smyslu. Důležité je, že producenti zpráv s kanály vůbec nepracují a ani o nich nemusí vědět. Producent pouze musí zjistit, které nsqd akceptují zprávy s daným tématem, což zajišťuje nsqlookupd, který tyto informace dostává od jednotlivých nsqd (ti mu je posílají přes dlouhodobé TCP). Naproti tomu konzument si volí jak kanál, tak i téma akceptovaných zpráv (což ostatně uvidíme ze zdrojových kódů demonstračních příkladů). Navíc může být konzument připojen k většímu množství nsqd, což vede k tomu, že může jednu zprávu dostat vícekrát.

Toto schéma do značné míry odstraňuje problém SPOF – Single Point of Failure. Sice může nastat situace, kdy je nsqlookupd jakožto centrální prvek nedostupný, ovšem stávající klienty to neovlivní (ti jsou již připojení ke konkrétním nsqd) a po znovuzprovoznění této služby se po chvíli obnoví kýžený stav celého systému.

V dalších kapitolách si ukážeme, jak celé nsq funguje prakticky.

4. Překlad a instalace systému nsq

Před odzkoušením nsq pochopitelně musíme celý systém nainstalovat, nakonfigurovat a spustit. Nejprve je nutné nainstalovat samotný systém nsq. Máme přitom dvě možnosti – buď se přímo ze stránky https://nsq.io/deployment/installing.html stáhnou předpřipravené binární obrazy všech utilit, nebo je umožněno překlad provést lokálně. První možnost – stažení tarballu se všemi potřebnými utilitami – je (nebo alespoň v mém případě z nějakého důvodu byla) velmi pomalá, takže si vyzkoušíme druhou možnost, tj. překlad nsq ze zdrojových kódů. Potřebovat budeme standardní nástroje programovacího jazyka Go, ideálně ve verzi podporující práci s moduly (tj. mělo by se jednat o Go od verze 1.11, což by dnes neměl být na všech majoritních distribucích problém). Druhým potřebným nástrojem je make.

Nejprve naklonujeme repositář s nsq z GitHubu:

$ git clone https://github.com/nsqio/nsq
 
Cloning into 'nsq'...
remote: Enumerating objects: 18, done.
remote: Counting objects: 100% (18/18), done.
remote: Compressing objects: 100% (17/17), done.
remote: Total 11117 (delta 2), reused 7 (delta 1), pack-reused 11099
Receiving objects: 100% (11117/11117), 12.18 MiB | 505.00 KiB/s, done.
Resolving deltas: 100% (6658/6658), done.
Checking connectivity... done.

Dále přejdeme do naklonovaného repositáře:

$ cd nsq

Samotný překlad spustíme jediným příkazem make:

$ make
 
go build  -o build/nsqd ./apps/nsqd
go: finding github.com/nsqio/go-nsq v1.0.7
go: finding github.com/bitly/timer_metrics v0.0.0-20170606164300-b1c65ca7ae62
go: finding github.com/pmezard/go-difflib v1.0.0
go: finding github.com/stretchr/testify v1.2.2
go: finding github.com/julienschmidt/httprouter v1.2.0
go: finding github.com/BurntSushi/toml v0.3.1
go: finding github.com/bitly/go-hostpool v0.0.0-20171023180738-a3a6125de932
go: finding github.com/davecgh/go-spew v1.1.1
go: finding github.com/bmizerany/perks v0.0.0-20141205001514-d9a9656a3a4b
go: finding golang.org/x/sys v0.0.0-20181221143128-b4a75ba826a6
go: finding github.com/nsqio/go-diskqueue v0.0.0-20180306152900-74cfbc9de839
go: finding github.com/judwhite/go-svc v1.0.0
go: finding github.com/mreiferson/go-options v0.0.0-20190302015348-0c63f026bcd6
go: finding github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db
go: finding github.com/blang/semver v3.5.1+incompatible
go: downloading github.com/BurntSushi/toml v0.3.1
go: downloading github.com/judwhite/go-svc v1.0.0
go: downloading github.com/nsqio/go-nsq v1.0.7
go: downloading github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db
go: downloading github.com/julienschmidt/httprouter v1.2.0
go: downloading github.com/mreiferson/go-options v0.0.0-20190302015348-0c63f026bcd6
go: downloading github.com/bmizerany/perks v0.0.0-20141205001514-d9a9656a3a4b
go: downloading github.com/nsqio/go-diskqueue v0.0.0-20180306152900-74cfbc9de839
go: downloading github.com/blang/semver v3.5.1+incompatible
go build  -o build/nsqlookupd ./apps/nsqlookupd
go build  -o build/nsqadmin ./apps/nsqadmin
go build  -o build/nsq_to_nsq ./apps/nsq_to_nsq
go: downloading github.com/bitly/go-hostpool v0.0.0-20171023180738-a3a6125de932
go: downloading github.com/bitly/timer_metrics v0.0.0-20170606164300-b1c65ca7ae62
go build  -o build/nsq_to_file ./apps/nsq_to_file
go build  -o build/nsq_to_http ./apps/nsq_to_http
go build  -o build/nsq_tail ./apps/nsq_tail
go build  -o build/nsq_stat ./apps/nsq_stat
go build  -o build/to_nsq ./apps/to_nsq

Po relativně krátkém překladu by měl být výsledek tohoto procesu uložen do podadresáře build/. Povšimněte si, že vzniklo hned několik binárních spustitelných souborů, z nichž některé budou popsány a použity v navazujících kapitolách:

-rwxr-xr-x 1 tester tester 10725519 říj 25 22:07 nsqadmin
-rwxr-xr-x 1 tester tester 11209429 říj 25 22:06 nsqd
-rwxr-xr-x 1 tester tester 10303290 říj 25 22:06 nsqlookupd
-rwxr-xr-x 1 tester tester  6922106 říj 25 22:07 nsq_stat
-rwxr-xr-x 1 tester tester  7297068 říj 25 22:07 nsq_tail
-rwxr-xr-x 1 tester tester  7506571 říj 25 22:07 nsq_to_file
-rwxr-xr-x 1 tester tester  7434625 říj 25 22:07 nsq_to_http
-rwxr-xr-x 1 tester tester  7409874 říj 25 22:07 nsq_to_nsq
-rwxr-xr-x 1 tester tester  7080549 říj 25 22:07 to_nsq

Zkontrolujeme, zda jsou soubory spustitelné:

$ cd build
 
$ ./nsqlookupd --version
nsqlookupd v1.2.1-alpha (built w/go1.11.2)
 
$ ./nsqd --version
nsqd v1.2.1-alpha (built w/go1.11.2)

Poznámka: podadresář build je vhodné přidat na PATH, protože právě zde umístěné soubory budeme spouštět v rámci pokusů prováděných v dalších kapitolách.

5. Instalace knihoven používaných klienty

Ve druhém kroku nainstalujeme knihovny, které budou používány klienty systému nsq, tj. jak producenty, tak i konzumenty. Seznam oficiálních i neoficiálních knihoven určených pro různé programovací jazyky je možné nalézt na adrese https://nsq.io/clients/client_libraries.html.

Poznámka: nsqd na prvním řádku tabulky je uveden z toho důvodu, že umožňuje posílání zpráv přes REST API.

Pro účely dnešního článku nám budou postačovat knihovny určené pro jazyk Go (https://github.com/nsqio/go-nsq) a taktéž pro jazyk Python (https://github.com/nsqio/pynsq).

Instalace knihovny (či možná lépe řečeno balíčku) pro jazyk Go je ve skutečnosti triviální, protože nám postačuje použít tento příkaz:

$ go get github.com/nsqio/go-nsq

Knihoven určených pro jazyk Python existuje několik (jedna je založena na Tornadu, druhá na knihovně gevent), my ovšem použijeme oficiálně podporovanou knihovnu pynsq určenou právě pro Tornado. Tuto knihovnu je možné nainstalovat nástrojem pip popř. pip3, a to následujícím způsobem:

$ pip3 install --user pynsq
 
Downloading/unpacking pynsq
  Downloading pynsq-0.8.3.tar.gz
  Running setup.py (path:/tmp/ramdisk/pip_build_tester/pynsq/setup.py) egg_info for package pynsq
 
Downloading/unpacking tornado<5.0 (from pynsq)
  Downloading tornado-4.5.3.tar.gz (484kB): 484kB downloaded
  Running setup.py (path:/tmp/ramdisk/pip_build_tester/tornado/setup.py) egg_info for package tornado
 
    no previously-included directories found matching 'docs/build'
    warning: no files found matching 'tornado/test/README'
Downloading/unpacking backports-abc>=0.4 (from tornado<5.0->pynsq)
  Downloading backports_abc-0.5-py2.py3-none-any.whl
Installing collected packages: pynsq, tornado, backports-abc
  Running setup.py install for pynsq
 
  Running setup.py install for tornado
    building 'tornado.speedups' extension
    x86_64-linux-gnu-gcc -pthread -DNDEBUG -g -fwrapv -O2 -Wall -Wstrict-prototy
    ...
    ...
    ...
Successfully installed pynsq tornado backports-abc
Cleaning up...

Poznámka: pokud ještě nemáte nainstalován balíček Tornado, dojde k pokusu o jeho překlad. Budete tedy potřebovat jak GCC, tak i balíček python-dev či python-devel z vaší Linuxové distribuce.

6. Spuštění jednotlivých uzlů systému nsq

Nyní, když již máme nainstalován jak samotný systém nsq, tak i knihovny používané na straně klientů, si můžeme vyzkoušet jeho základní chování. Nejdříve je nutné spustit jednotlivé uzly, z nichž se nsq skládá (viz též druhou kapitolu s popisem celkové architektury systému).

Nejprve spustíme nsqlookupd, což je služba, které není (v našem značně zjednodušeném případě) nutné předávat žádné další parametry. Tato služba pochopitelně může běžet na pozadí, ovšem pro sledování její činnosti je lepší ji spustit v samostatném terminálu:

$ ./nsqlookupd
 
[nsqlookupd] 2019/10/26 12:14:11.480329 INFO: nsqlookupd v1.2.1-alpha (built w/go1.11.2)
[nsqlookupd] 2019/10/26 12:14:11.481027 INFO: TCP: listening on [::]:4160
[nsqlookupd] 2019/10/26 12:14:11.481947 INFO: HTTP: listening on [::]:4161

Pro jistotu odzkoušíme, jestli služba skutečně běží:

$ curl -v localhost:4161/ping
 
* Hostname was NOT found in DNS cache
*   Trying 127.0.0.1...
* Connected to localhost (127.0.0.1) port 4161 (#0)
> GET /ping HTTP/1.1
> User-Agent: curl/7.35.0
> Host: localhost:4161
> Accept: */*
>
< HTTP/1.1 200 OK
< Date: Sun, 27 Oct 2019 18:07:41 GMT
< Content-Length: 2
< Content-Type: text/plain; charset=utf-8
<
* Connection #0 to host localhost left intact
OK

V dalším kroku spustíme jednu instanci nsqd, které předáme adresu a port již běžící služby nsqlookupd. Tyto informace je nutné znát, protože nsqd bude službu nsqlookupd informovat o svém stavu (kanály, témata, počty a stavy zpráv). Komunikace probíhá po TCP, jehož socket je otevřen na portu 4160, tedy:

$ ./nsqd --lookupd-tcp-address=127.0.0.1:4160
 
[nsqd] 2019/10/26 12:15:09.617256 INFO: nsqd v1.2.1-alpha (built w/go1.11.2)
[nsqd] 2019/10/26 12:15:09.617365 INFO: ID: 961
[nsqd] 2019/10/26 12:15:09.617669 INFO: NSQ: persisting topic/channel metadata to nsqd.dat
[nsqd] 2019/10/26 12:15:09.623347 INFO: LOOKUP(127.0.0.1:4160): adding peer
[nsqd] 2019/10/26 12:15:09.623403 INFO: LOOKUP connecting to 127.0.0.1:4160
[nsqd] 2019/10/26 12:15:09.623756 INFO: HTTP: listening on [::]:4151
[nsqd] 2019/10/26 12:15:09.623410 INFO: TCP: listening on [::]:4150
[nsqd] 2019/10/26 12:15:09.625281 INFO: LOOKUPD(127.0.0.1:4160): peer info {TCPPort:4160 HTTPPort:4161 Version:1.2.1-alpha BroadcastAddress:tester-ThinkPad-T410}

I tuto službu lze otestovat přes její REST API:

$ curl -v localhost:4151/ping
 
* Hostname was NOT found in DNS cache
*   Trying 127.0.0.1...
* Connected to localhost (127.0.0.1) port 4151 (#0)
> GET /ping HTTP/1.1
> User-Agent: curl/7.35.0
> Host: localhost:4151
> Accept: */*
>
< HTTP/1.1 200 OK
< Date: Sun, 27 Oct 2019 18:09:05 GMT
< Content-Length: 2
< Content-Type: text/plain; charset=utf-8
<
* Connection #0 to host localhost left intact
OK

Nakonec spustíme službu nsqadmin, která poskytuje uživatelské rozhraní, k němuž je možné se připojit z webového prohlížeče:

$ ./nsqadmin --lookupd-http-address=127.0.0.1:4161
 
[nsqadmin] 2019/10/26 12:15:43.095179 INFO: nsqadmin v1.2.1-alpha (built w/go1.11.2)
[nsqadmin] 2019/10/26 12:15:43.096054 INFO: HTTP: listening on [::]:4171

Opět můžeme použít REST API (ping) pro otestování základní činnosti serveru, a to na portu 4117:

$ curl -v localhost:4171/ping
 
* Hostname was NOT found in DNS cache
*   Trying 127.0.0.1...
* Connected to localhost (127.0.0.1) port 4171 (#0)
> GET /ping HTTP/1.1
> User-Agent: curl/7.35.0
> Host: localhost:4171
> Accept: */*
>
< HTTP/1.1 200 OK
< Date: Sun, 27 Oct 2019 18:10:37 GMT
< Content-Length: 2
< Content-Type: text/plain; charset=utf-8
<
* Connection #0 to host localhost left intact
OK

7. Používané porty

Jen pro větší přehlednost si vypišme TCP porty, s nimiž budeme buď přímo či nepřímo pracovat:

Port	Uzel	Typ	Význam
4150	nsqd	TCP	použit producenty pro posílání zpráv
4151	nsqd	HTTP	použit producenty pro posílání zpráv
4160	nsqlookupd	TCP	použit nsqd, kteří informují o svém stavu
4161	nsqlookupd	HTTP	REST API rozhraní pro konzumenty a další nástroje
4171	nsqadmin	HTTP	webové UI administrátorské konzole

Poznámka: producenti mohou zprávy posílat jak na port 4150, tak i 4151 v závislosti na jejich konfiguraci, použité knihovně atd. Použití portu 4151 je jednodušší (klasické REST API), ovšem pomalejší, než když si klient zajistí dlouhodobě otevřený socket na port 4150. Navíc, pokud se nakonfiguruje i použití HTTPS, se používá port 4152. V případě, že je nutné spustit více nsqd na stejném počítači, bude pochopitelně nutné zvolit odlišná čísla portů – viz též příkaz:

./nsqd --help
 
  ...
  ...
  ...
  -http-address string
        <addr>:<port> to listen on for HTTP clients (default "0.0.0.0:4151")
  ...
  ...
  ...
  -https-address string
        <addr>:<port> to listen on for HTTPS clients (default "0.0.0.0:4152")
  ...
  ...
  ...
  -tcp-address string
        <addr>:<port> to listen on for TCP clients (default "0.0.0.0:4150")

8. Poslání zprávy přes REST API

Pro poslání zprávy není ve skutečnosti nutné používat žádné komplikované mechanismy, protože si vystačíme s běžným REST API běžícím (přesněji řečeno naslouchajícím) na portu 4151 a tedy i s nástrojem typu curl, httpie atd. Zpráva se konkrétně pošle v těle požadavku a v URL je nutné specifikovat téma (topic) zprávy:

$ curl -d 'zprava1' http://127.0.0.1:4151/pub?topic=test
 
OK

Pokud se podíváte na terminál služby/uzlu nsqd, měly by se na něm objevit tyto dvě zprávy oznamující, že bylo vytvořeno nové téma a že zprávy budou meziuloženy v souboru nsqd.dat (způsob uložení zpráv je pochopitelně taktéž konfigurovatelný):

[nsqd] 2019/10/26 12:05:26.826143 INFO: TOPIC(test): created
[nsqd] 2019/10/26 12:05:26.826954 INFO: NSQ: persisting topic/channel metadata to nsqd.dat

Obrázek 4: Stav systému po poslání první zprávy.

Pochopitelně je možné poslat zprávu s binárním obsahem, načtenou z nějakého souboru:

$ curl -d 'zprava1' http://127.0.0.1:4151/pub?topic=test
 
OK

9. Komunikace s uzlem nsqlookupd přes REST API

Uzel typu nsqlookupd nabízí všem klientům jednoduše použitelné REST API. V této kapitole se podíváme na některé základní způsoby jeho použití. Vzhledem k tomu, že většina odpovědí je ve formě JSONu, který ovšem není naformátován (je vrácen na jediném řádku), budeme v dále uvedených příkladech používat následující alias, který JSON naformátuje a zobrazí na standardním výstupu:

$ alias pp='python -mjson.tool'

Dotaz, zda je služba funkční, jsme již viděli v šesté kapitole, takže jen krátce:

$ curl -s localhost:4161/ping 
 
OK

Základní informace o službě (v současnosti se zobrazí pouze verze):

$ curl -s localhost:4161/info | pp
 
{
    "version": "1.2.1-alpha"
}

Získání seznamu všech založených témat (téma „test“ jsme vytvořili v rámci předchozí kapitoly):

$ curl -s localhost:4161/topics | pp
 
{
    "topics": [
        "test"
    ]
}

Získání seznamu kanálů a producentů, kteří přispěli alespoň jednou zprávou do zvoleného tématu:

$ curl -s localhost:4161/lookup?topic=test | pp
 
{
    "channels": [
        "test"
        "nsq_to_file"
    ],
    "producers": [
        {
            "broadcast_address": "tester-ThinkPad-T410",
            "hostname": "tester-ThinkPad-T410",
            "http_port": 4151,
            "remote_address": "127.0.0.1:59503",
            "tcp_port": 4150,
            "version": "1.2.1-alpha"
        }
    ]
}

Poznámka: ve skutečnosti se vám zobrazí prázdný seznam kanálů, a to až do chvíle, dokud není spuštěn nějaký konzument.

Informace o kanálech, přes které se připojují klienti ke zvolenému tématu:

$ curl -s localhost:4161/channels?topic=test | pp
 
{
    "channels": [
        "test"
        "nsq_to_file"
    ]
}

Informace o všech producentech zpráv, včetně témat:

$ curl -s localhost:4161/nodes | pp
 
{
    "producers": [
        {
            "broadcast_address": "tester-ThinkPad-T410",
            "hostname": "tester-ThinkPad-T410",
            "http_port": 4151,
            "remote_address": "127.0.0.1:59503",
            "tcp_port": 4150,
            "tombstones": [
                false
            ],
            "topics": [
                "test"
            ],
            "version": "1.2.1-alpha"
        }
    ]
}

Vytvoření nového tématu (musí se použít metoda POST):

$ curl -X POST -s localhost:4161/topic/create?topic=dalsi_tema
 
$ curl -s localhost:4161/topics | pp
 
{
    "topics": [
        "dalsi_tema",
        "test"
    ]
}

Smazání tématu:

$ curl -X POST -s localhost:4161/topic/delete?topic=dalsi_tema
 
$ curl -s localhost:4161/topics | pp
 
{
    "topics": [
        "test"
    ]
}

10. Jednoduchý producent zpráv naprogramovaný v Pythonu

V této kapitole si ukážeme jednoduchého producenta zpráv naprogramovaného v Pythonu. Tento producent je (v poněkud jednodušší podobě) popsán na stránce https://pynsq.readthedocs.io/en/latest/writer.html. Je založen na třídě Writer a taktéž na periodickém volání callback funkce pub_message, která po poslání zprávy zavolá další callback funkci finish_pub:

import nsq
import tornado.ioloop
import time
 
cnt = 1
 
def pub_message():
    global cnt
    writer.pub('test', 'zprava {}'.format(cnt).encode(), finish_pub)
    cnt += 1
 
def finish_pub(conn, data):
    print("FINISHING")
    print(data.decode())
    print("---------")
 
writer = nsq.Writer(['127.0.0.1:4150'])
 
tornado.ioloop.PeriodicCallback(pub_message, 1000).start()
nsq.run()

Poznámka: původní kód byl určen pro pomalu ale jistě dosluhující Python 2.x. Úprava pro Python 3 je snadná, pouze musíme zajistit převod řetězců na pole bajtů a naopak (metody encode() a decode()).

Obrázek 5: Situace po zveřejnění několika zpráv producentem.

11. Konzument zpráv naprogramovaný v Pythonu

Konzumenta zpráv je možné v Pythonu naprogramovat doslova na několika programových řádcích. Postačuje vytvořit instanci objektu typu Reader a předat jí odkaz na callback funkci zavolanou pro každou přijatou zprávu, adresu běžící služby nsqlookupd, název tématu, které se bude zpracovávat a taktéž název kanálu. Pokud kanál před připojením konzumenta neexistoval, bude automaticky vytvořen a objeví se i v administrátorské konzoli popř. na REST API:

import nsq
 
def handler(message):
    print(message)
    return True
 
r = nsq.Reader(message_handler=handler,
        lookupd_http_addresses=['http://127.0.0.1:4161'],
        topic='test', channel='test', lookupd_poll_interval=15)
 
nsq.run()

Poznámka: povšimněte si, že adresa služby lookupd je umístěna v seznamu. Nejedná se o chybu, protože je skutečně možné specifikovat více adres a tím pádem se vyrovnat se situací, kdy jedna ze služeb lookupd není z nějakého důvodu dostupná.

Obrázek 6: Informace o počtu zpracovaných zpráv.

12. Použití systému nsq klienty naprogramovanými v Go

Jak jsme si již řekli v předchozích kapitolách, je možné klienty (typicky běžné producenty a konzumenty zpráv) vytvářet v různých programovacích jazycích. Ukažme si tedy, jak se vytvoří jednoduchý producent a konzument v programovacím jazyku Go. Budeme přitom potřebovat balíček nazvaný github.com/nsqio/go-nsq nainstalovaný v rámci páté kapitoly. Pro jednoduchého producenta zpráv postačuje zavolat konstruktor:

func NewProducer(addr string, config *Config) (*Producer, error)

Konstruktoru je nutné předat především adresu běžící (a dostupné) služby nsqd, na kterou se zpráva pošle. Druhým parametrem je struktura s konfigurací, která se vytvoří a inicializuje konstruktorem:

func NewConfig() *Config

Prozatím nám budou stačit výchozí parametry, takže obsah této struktury nebudeme modifikovat.

Producent může posílat zprávy více způsoby – synchronně, asynchronně, může poslat více zpráv atd.:

func (w *Producer) Publish(topic string, body []byte) error
func (w *Producer) PublishAsync(topic string, body []byte, doneChan chan *ProducerTransaction, args ...interface{}) error
func (w *Producer) MultiPublish(topic string, body [][]byte) error
func (w *Producer) MultiPublishAsync(topic string, body [][]byte, doneChan chan *ProducerTransaction, args ...interface{}) error

Pro vytvoření konzumenta zpráv se použije odlišný konstruktor:

func NewConsumer(topic string, channel string, config *Config) (*Consumer, error)

Povšimněte si, že musíme specifikovat téma (topic), jméno kanálu (channel) a taktéž strukturu s konfigurací. Ve chvíli, kdy je konzument inicializován, je nutné ho připojit buď přímo ke zvolenému nsqd, a to metodou:

func (r *Consumer) ConnectToNSQD(addr string) error

Alternativně je možné se nejdříve připojit ke službě nsqlookupd, která udržuje informace o aktuálně dostupných nsqd. Existují dvě možnosti – dotaz na jedinou službu nsqlookupd nebo dotaz poslaný na více služeb:

func (r *Consumer) ConnectToNSQLookupd(addr string) error
func (r *Consumer) ConnectToNSQLookupds(addresses []string) error

Pochopitelně je nutné zajistit i odpojení, ideálně v bloku defer:

func (r *Consumer) DisconnectFromNSQD(addr string) error
func (r *Consumer) DisconnectFromNSQLookupd(addr string) error

Samotné posílané zprávy jsou reprezentovány strukturou s obsahem zprávy, jejím ID, časovým razítkem a dalšími podrobnějšími informacemi:

type Message struct {
    ID        MessageID
    Body      []byte
    Timestamp int64
    Attempts  uint16
 
    NSQDAddress string
 
    Delegate MessageDelegate
}

13. Producent zpráv naprogramovaný v jazyce Go

Jednoduchý producent zpráv je naprogramován takovým způsobem, aby se připojil přímo ke zvolenému nsqd (na port 4150), poslal jedinou zprávu s tématem „test“ a následně se ukončil. Povšimněte si, že tělo zprávy je tvořeno sekvencí bajtů a na konci musíme producenta explicitně zastavit:

package main
 
import (
        "github.com/nsqio/go-nsq"
        "log"
)
 
func main() {
        config := nsq.NewConfig()
 
        producer, err := nsq.NewProducer("127.0.0.1:4150", config)
        if err != nil {
                log.Panic("Producer can't be constructed")
        }
 
        err = producer.Publish("test", []byte("zprava z Go"))
        if err != nil {
                log.Panic("Could not connect")
        }
 
        producer.Stop()
}

14. Využití bloku defer pro ukončení producenta zpráv

V praxi je lepší (a pro programovací jazyk Go i idiomatičtější) zajistit odpojení a ukončení producenta zpráv s využitím bloku defer, což je ukázáno v další verzi producenta (viz zvýrazněný řádek kódu):

package main
 
import (
        "github.com/nsqio/go-nsq"
        "log"
)
 
func main() {
        config := nsq.NewConfig()
 
        producer, err := nsq.NewProducer("127.0.0.1:4150", config)
        if err != nil {
                log.Panic("Producer can't be constructed")
        }
        defer producer.Stop()
 
        err = producer.Publish("test", []byte("zprava z Go"))
        if err != nil {
                log.Panic("Could not connect")
        }
}

15. Jednoduchý konzument zpráv používající synchronizaci pro ukončení gorutiny s příjemcem

Nyní si můžeme ukázat konzumenta zpráv. Ten je založen na využití callback funkce (handleru), jenž je zavolán ve chvíli, kdy byla přijata nová zpráva. Tento koncept jsme ostatně viděli už v konzumentovi naprogramovaném v Pythonu. Samotný handler lze vytvořit ve formě uzávěru, protože musí mít přístup k synchronizační entitě wait group (i s ní už jsme se seznámili, a to v paralelně běžícím seriálu o programovacím jazyce Go):

consumer.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error {
        log.Printf("Received a message: %s", string(message.Body))
        waitgroup.Done()
        return nil
}))

Synchronizační entita slouží k tomu, aby se po zaregistrování handleru aplikace neukončila. Ukončení zajistí až zavolání waithroup.Done():

package main
 
import (
        "github.com/nsqio/go-nsq"
        "log"
        "sync"
)
 
func main() {
        config := nsq.NewConfig()
 
        consumer, err := nsq.NewConsumer("test", "test", config)
        if err != nil {
                log.Panic("Consumer can't be constructed")
        }
 
        waitgroup := &sync.WaitGroup{}
        waitgroup.Add(1)
 
        consumer.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error {
                log.Printf("Received a message: %s", string(message.Body))
                waitgroup.Done()
                return nil
        }))
 
        err = consumer.ConnectToNSQD("127.0.0.1:4150")
        if err != nil {
                log.Panic("Could not connect")
        }
 
        log.Println("Waiting for message")
        waitgroup.Wait()
}

Poznámka: do programu můžete přidat i explicitní příkaz odpojení od nsqd. Není to sice striktně vyžadováno, protože k odpojení dojde i po ukončení aplikace, ale programy psané v Go bývají v tomto ohledu velmi explicitní:

defer consumer.DisconnectFromNSQD("127.0.0.1:4150")

16. Použití kanálu ve funkci synchronizační struktury

Pro čekání na příjem zprávy můžeme využít i kanály. Pro tento účel postačuje vytvořit kanál, do něhož se zapíše (libovolná) informace po přijetí zprávy. Na tuto informaci budeme čekat v hlavní gorutině. Výsledkem je (alespoň podle mého názoru) přehlednější kód (změněné části jsou opět zvýrazněny):

package main
 
import (
        "github.com/nsqio/go-nsq"
        "log"
)
 
func main() {
        config := nsq.NewConfig()
 
        consumer, err := nsq.NewConsumer("test", "test", config)
        if err != nil {
                log.Panic("Consumer can't be constructed")
        }
 
        done := make(chan bool)
 
        consumer.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error {
                log.Printf("Received a message: %s", string(message.Body))
                done <- true
                return nil
        }))
 
        err = consumer.ConnectToNSQD("127.0.0.1:4150")
        if err != nil {
                log.Panic("Could not connect")
        }
 
        log.Println("Waiting for message")
        <-done
}

Opět můžete přidat volání funkce pro explicitní odpojení klienta od nsqd:

defer consumer.DisconnectFromNSQD("127.0.0.1:4150")

17. Připojení klienta ke službě nsqlookupd

Klient, přesněji řečeno konzument zpráv, by se správně měl připojovat spíše ke službě nsqlookupd a nikoli přímo k nsqd. Proto program nepatrně upravíme a použijeme metodu ConnectToNSQLookupd namísto ConnectToNSQD. Samozřejmě budeme muset použít odlišnou adresu:

package main
 
import (
        "github.com/nsqio/go-nsq"
        "log"
)
 
func main() {
        config := nsq.NewConfig()
 
        consumer, err := nsq.NewConsumer("test", "test", config)
        if err != nil {
                log.Panic("Consumer can't be constructed")
        }
 
        done := make(chan bool)
 
        consumer.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error {
                log.Printf("Received a message: %s", string(message.Body))
                return nil
        }))
 
        err = consumer.ConnectToNSQLookupd("127.0.0.1:4161")
        if err != nil {
                log.Panic("Could not connect")
        }
 
        log.Println("Waiting for message")
        <-done
}

18. Obsah následující části seriálu

V další části seriálu o message brokere popis systému nsq dokončíme. Seznámíme se zejména se složitějšími architekturami, v nichž běží více služeb nsqlookupd i nsqd a k nimž se mohou připojovat klienti.

19. Repositář s demonstračními příklady

Zdrojové kódy všech dnes popsaných demonstračních příkladů vyvinutých v programovacím jazyku Python a Go byly uloženy do Git repositáře, který je dostupný na adrese https://github.com/tisnik/message-queues-examples (stále na GitHubu :-). V případě, že nebudete chtít klonovat celý repositář (ten je ovšem – alespoň prozatím – velmi malý, dnes má stále ještě doslova několik kilobajtů), můžete namísto toho použít odkazy na jednotlivé příklady, které naleznete v následující tabulce:

Příklad	Skript	Stručný popis	Cesta
1	producer.py	jednoduchý producent zpráv naprogramovaný v Pythonu	https://github.com/tisnik/message-queues-examples/blob/master/nsq/producer.py
2	consumer.py	konzument zpráv naprogramovaný v Pythonu	https://github.com/tisnik/message-queues-examples/blob/master/nsq/consumer.py

3	producer.go	producent jedné zprávy naprogramované v jazyce Go	https://github.com/tisnik/message-queues-examples/blob/master/nsq/producer.go
4	producer2.go	producent jedné zprávy naprogramované v jazyce Go s blokem defer	https://github.com/tisnik/message-queues-examples/blob/master/nsq/producer.go
5	consumer1.go	konzument naprogramovaný v Go, který využívá wait groups při čekání na doručení zprávy	https://github.com/tisnik/message-queues-examples/blob/master/nsq/consumer1.go
6	consumer2.go	konzument naprogramovaný v Go, který využívá kanál při čekání na doručení zprávy	https://github.com/tisnik/message-queues-examples/blob/master/nsq/consumer2.go
7	consumer3.go	ukázka použití metody ConnectToNSQLookupd	https://github.com/tisnik/message-queues-examples/blob/master/nsq/consumer3.go

20. Odkazy na Internetu

Stránka projektu NSQ
https://nsq.io/
Dokumentace k projektu NSQ
https://nsq.io/overview/design.html
Dokumentace ke klientovi pro Go
https://godoc.org/github.com/nsqio/go-nsq
Dokumentace ke klientovi pro Python
https://pynsq.readthedocs.io/en/latest/
Binární tarbally s NSQ
https://nsq.io/deployment/installing.html
GitHub repositář projektu NSQ
https://github.com/nsqio/nsq
Klienti pro NSQ
https://nsq.io/clients/client_libraries.html
Klient pro Go
https://github.com/nsqio/go-nsq
Klient pro Python
https://github.com/nsqio/pynsq
An Example of Using NSQ From Go
http://tleyden.github.io/blog/2014/11/12/an-example-of-using-nsq-from-go/
Go Go Gadget
https://word.bitly.com/post/29550171827/go-go-gadget
Simplehttp
https://github.com/bitly/simplehttp
Dramatiq: simple task processing
https://dramatiq.io/
Cookbook (for Dramatiq)
https://dramatiq.io/cookbook.html
Balíček dramatiq na PyPi
https://pypi.org/project/dramatiq/
Dramatiq dashboard
https://github.com/Bogdanp/dramatiq_dashboard
Dramatiq na Redditu
https://www.reddit.com/r/dramatiq/
A Dramatiq broker that can be used with Amazon SQS
https://github.com/Bogdanp/dramatiq_sqs
nanomsg na GitHubu
https://github.com/nanomsg/nanomsg
Referenční příručka knihovny nanomsg
https://nanomsg.org/v1.1.5/nanomsg.html
nng (nanomsg-next-generation)
https://github.com/nanomsg/nng
Differences between nanomsg and ZeroMQ
https://nanomsg.org/documentation-zeromq.html
NATS
https://nats.io/about/
NATS Streaming Concepts
https://nats.io/documentation/streaming/nats-streaming-intro/
NATS Streaming Server
https://nats.io/download/nats-io/nats-streaming-server/
NATS Introduction
https://nats.io/documentation/
NATS Client Protocol
https://nats.io/documentation/internals/nats-protocol/
NATS Messaging (Wikipedia)
https://en.wikipedia.org/wiki/NATS_Messaging
Stránka Apache Software Foundation
http://www.apache.org/
Informace o portu 5672
http://www.tcp-udp-ports.com/port-5672.htm
Třída MessagingHandler knihovny Qpid Proton
https://qpid.apache.org/releases/qpid-proton-0.27.0/proton/python/api/proton._handlers.MessagingHandler-class.html
Třída Event knihovny Qpid Proton
https://qpid.apache.org/releases/qpid-proton-0.27.0/proton/python/api/proton._events.Event-class.html
package stomp (Go)
https://godoc.org/github.com/go-stomp/stomp
Go language library for STOMP protocol
https://github.com/go-stomp/stomp
python-qpid-proton 0.26.0 na PyPi
https://pypi.org/project/python-qpid-proton/
Qpid Proton
http://qpid.apache.org/proton/
Using the AMQ Python Client
https://access.redhat.com/documentation/en-us/red_hat_amq/7.1/html-single/using_the_amq_python_client/
Apache ActiveMQ
http://activemq.apache.org/
Apache ActiveMQ Artemis
https://activemq.apache.org/artemis/
Apache ActiveMQ Artemis User Manual
https://activemq.apache.org/artemis/docs/latest/index.html
KahaDB
http://activemq.apache.org/kahadb.html
Understanding the KahaDB Message Store
https://access.redhat.com/documentation/en-US/Fuse_MQ_Enterprise/7.1/html/Configuring_Broker_Persistence/files/KahaDBOverview.html
Command Line Tools (Apache ActiveMQ)
https://activemq.apache.org/activemq-command-line-tools-reference.html
stomp.py 4.1.21 na PyPi
https://pypi.org/project/stomp.py/
Stomp Tutorial
https://access.redhat.com/documentation/en-US/Fuse_Message_Broker/5.5/html/Connectivity_Guide/files/FMBConnectivityStompTelnet.html
Heartbeat (computing)
https://en.wikipedia.org/wiki/Heartbeat_(computing)
Apache Camel
https://camel.apache.org/
Red Hat Fuse
https://developers.redhat.com/products/fuse/overview/
Confusion between ActiveMQ and ActiveMQ-Artemis?
https://serverfault.com/questions/873533/confusion-between-activemq-and-activemq-artemis
Staré stránky projektu HornetQ
http://hornetq.jboss.org/
Snapshot JeroMQ verze 0.4.4
https://oss.sonatype.org/content/repositories/snapshots/org/zeromq/jeromq/0.4.4-SNAPSHOT/
Difference between ActiveMQ vs Apache ActiveMQ Artemis
http://activemq.2283324.n4.nabble.com/Difference-between-ActiveMQ-vs-Apache-ActiveMQ-Artemis-td4703828.html
Microservices communications. Why you should switch to message queues
https://dev.to/matteojoliveau/microservices-communications-why-you-should-switch-to-message-queues–48ia
Stomp.py 4.1.19 documentation
https://stomppy.readthedocs.io/en/stable/
Repositář knihovny JeroMQ
https://github.com/zeromq/jeromq/
ØMQ – Distributed Messaging
http://zeromq.org/
ØMQ Community
http://zeromq.org/community
Get The Software
http://zeromq.org/intro:get-the-software
PyZMQ Documentation
https://pyzmq.readthedocs.io/en/latest/
Module: zmq.decorators
https://pyzmq.readthedocs.io/en/latest/api/zmq.decorators.html
ZeroMQ is the answer, by Ian Barber
https://vimeo.com/20605470
ZeroMQ RFC
https://rfc.zeromq.org/
ZeroMQ and Clojure, a brief introduction
https://antoniogarrote.wordpress.com/2010/09/08/zeromq-and-clojure-a-brief-introduction/
zeromq/czmq
https://github.com/zeromq/czmq
golang wrapper for CZMQ
https://github.com/zeromq/goczmq
ZeroMQ version reporting in Python
http://zguide.zeromq.org/py:version
A Go interface to ZeroMQ version 4
https://github.com/pebbe/zmq4
Broker vs. Brokerless
http://zeromq.org/whitepapers:brokerless
Learning ØMQ with pyzmq
https://learning-0mq-with-pyzmq.readthedocs.io/en/latest/
Céčková funkce zmq_ctx_new
http://api.zeromq.org/4–2:zmq-ctx-new
Céčková funkce zmq_ctx_destroy
http://api.zeromq.org/4–2:zmq-ctx-destroy
Céčková funkce zmq_bind
http://api.zeromq.org/4–2:zmq-bind
Céčková funkce zmq_unbind
http://api.zeromq.org/4–2:zmq-unbind
Céčková C funkce zmq_connect
http://api.zeromq.org/4–2:zmq-connect
Céčková C funkce zmq_disconnect
http://api.zeromq.org/4–2:zmq-disconnect
Céčková C funkce zmq_send
http://api.zeromq.org/4–2:zmq-send
Céčková C funkce zmq_recv
http://api.zeromq.org/4–2:zmq-recv
Třída Context (Python)
https://pyzmq.readthedocs.io/en/latest/api/zmq.html#context
Třída Socket (Python)
https://pyzmq.readthedocs.io/en/latest/api/zmq.html#socket
Python binding
http://zeromq.org/bindings:python
Why should I have written ZeroMQ in C, not C++ (part I)
http://250bpm.com/blog:4
Why should I have written ZeroMQ in C, not C++ (part II)
http://250bpm.com/blog:8
About Nanomsg
https://nanomsg.org/
Advanced Message Queuing Protocol
https://www.amqp.org/
Advanced Message Queuing Protocol na Wikipedii
https://en.wikipedia.org/wiki/Advanced_Message_Queuing_Protocol
Dokumentace k příkazu rabbitmqctl
https://www.rabbitmq.com/rabbitmqctl.8.html
RabbitMQ
https://www.rabbitmq.com/
RabbitMQ Tutorials
https://www.rabbitmq.com/getstarted.html
RabbitMQ: Clients and Developer Tools
https://www.rabbitmq.com/devtools.html
RabbitMQ na Wikipedii
https://en.wikipedia.org/wiki/RabbitMQ
Streaming Text Oriented Messaging Protocol
https://en.wikipedia.org/wiki/Streaming_Text_Oriented_Messaging_Protocol
Message Queuing Telemetry Transport
https://en.wikipedia.org/wiki/MQTT
Erlang
http://www.erlang.org/
pika 0.12.0 na PyPi
https://pypi.org/project/pika/
Introduction to Pika
https://pika.readthedocs.io/en/stable/
Langohr: An idiomatic Clojure client for RabbitMQ that embraces the AMQP 0.9.1 model
http://clojurerabbitmq.info/
AMQP 0–9–1 Model Explained
http://www.rabbitmq.com/tutorials/amqp-concepts.html
Part 1: RabbitMQ for beginners – What is RabbitMQ?
https://www.cloudamqp.com/blog/2015–05–18-part1-rabbitmq-for-beginners-what-is-rabbitmq.html
Downloading and Installing RabbitMQ
https://www.rabbitmq.com/download.html
celery na PyPi
https://pypi.org/project/celery/
Databáze Redis (nejenom) pro vývojáře používající Python
https://www.root.cz/clanky/databaze-redis-nejenom-pro-vyvojare-pouzivajici-python/
Databáze Redis (nejenom) pro vývojáře používající Python (dokončení)
https://www.root.cz/clanky/databaze-redis-nejenom-pro-vyvojare-pouzivajici-python-dokonceni/
Redis Queue (RQ)
https://www.fullstackpython.com/redis-queue-rq.html
Python Celery & RabbitMQ Tutorial
https://tests4geeks.com/python-celery-rabbitmq-tutorial/
Flower: Real-time Celery web-monitor
http://docs.celeryproject.org/en/latest/userguide/monitoring.html#flower-real-time-celery-web-monitor
Asynchronous Tasks With Django and Celery
https://realpython.com/asynchronous-tasks-with-django-and-celery/
First Steps with Celery
http://docs.celeryproject.org/en/latest/getting-started/first-steps-with-celery.html
node-celery
https://github.com/mher/node-celery
Full Stack Python: web development
https://www.fullstackpython.com/web-development.html
Introducing RQ
https://nvie.com/posts/introducing-rq/
Asynchronous Tasks with Flask and Redis Queue
https://testdriven.io/asynchronous-tasks-with-flask-and-redis-queue
rq-dashboard
https://github.com/eoranged/rq-dashboard
Stránky projektu Redis
https://redis.io/
Introduction to Redis
https://redis.io/topics/introduction
Try Redis
http://try.redis.io/
Redis tutorial, April 2010 (starší, ale pěkně udělaný)
https://static.simonwillison.net/static/2010/redis-tutorial/
Python Redis
https://redislabs.com/lp/python-redis/
Redis: key-value databáze v paměti i na disku
https://www.zdrojak.cz/clanky/redis-key-value-databaze-v-pameti-i-na-disku/
Praktický úvod do Redis (1): vaše distribuovaná NoSQL cache
http://www.cloudsvet.cz/?p=253
Praktický úvod do Redis (2): transakce
http://www.cloudsvet.cz/?p=256
Praktický úvod do Redis (3): cluster
http://www.cloudsvet.cz/?p=258
Connection pool
https://en.wikipedia.org/wiki/Connection_pool
Instant Redis Sentinel Setup
https://github.com/ServiceStack/redis-config
How to install REDIS in LInux
https://linuxtechlab.com/how-install-redis-server-linux/
Redis RDB Dump File Format
https://github.com/sripathikrishnan/redis-rdb-tools/wiki/Redis-RDB-Dump-File-Format
Lempel–Ziv–Welch
https://en.wikipedia.org/wiki/Lempel%E2%80%93Ziv%E2%80%93Welch
Redis Persistence
https://redis.io/topics/persistence
Redis persistence demystified
http://oldblog.antirez.com/post/redis-persistence-demystified.html
Redis reliable queues with Lua scripting
http://oldblog.antirez.com/post/250
Ost (knihovna)
https://github.com/soveran/ost
NoSQL
https://en.wikipedia.org/wiki/NoSQL
Shard (database architecture)
https://en.wikipedia.org/wiki/Shard_%28database_architecture%29
What is sharding and why is it important?
https://stackoverflow.com/questions/992988/what-is-sharding-and-why-is-it-important
What Is Sharding?
https://btcmanager.com/what-sharding/
Redis clients
https://redis.io/clients
Category:Lua-scriptable software
https://en.wikipedia.org/wiki/Category:Lua-scriptable_software
Seriál Programovací jazyk Lua
https://www.root.cz/serialy/programovaci-jazyk-lua/
Redis memory usage
http://nosql.mypopescu.com/post/1010844204/redis-memory-usage
Ukázka konfigurace Redisu pro lokální testování
https://github.com/tisnik/presentations/blob/master/redis/redis.conf
Resque
https://github.com/resque/resque
Nested transaction
https://en.wikipedia.org/wiki/Nested_transaction
Publish–subscribe pattern
https://en.wikipedia.org/wiki/Publish%E2%80%93subscribe_pattern
Messaging pattern
https://en.wikipedia.org/wiki/Messaging_pattern
Using pipelining to speedup Redis queries
https://redis.io/topics/pipelining
Pub/Sub
https://redis.io/topics/pubsub
ZeroMQ distributed messaging
http://zeromq.org/
ZeroMQ: Modern & Fast Networking Stack
https://www.igvita.com/2010/09/03/zeromq-modern-fast-networking-stack/
Publish/Subscribe paradigm: Why must message classes not know about their subscribers?
https://stackoverflow.com/questions/2908872/publish-subscribe-paradigm-why-must-message-classes-not-know-about-their-subscr
Python & Redis PUB/SUB
https://medium.com/@johngrant/python-redis-pub-sub-6e26b483b3f7
Message broker
https://en.wikipedia.org/wiki/Message_broker
RESP Arrays
https://redis.io/topics/protocol#array-reply
Redis Protocol specification
https://redis.io/topics/protocol
Redis Pub/Sub: Intro Guide
https://www.redisgreen.net/blog/pubsub-intro/
Redis Pub/Sub: Howto Guide
https://www.redisgreen.net/blog/pubsub-howto/
Comparing Publish-Subscribe Messaging and Message Queuing
https://dzone.com/articles/comparing-publish-subscribe-messaging-and-message
Apache Kafka
https://kafka.apache.org/
Iron
http://www.iron.io/mq
kue (založeno na Redisu, určeno pro node.js)
https://github.com/Automattic/kue
Cloud Pub/Sub
https://cloud.google.com/pubsub/
Introduction to Redis Streams
https://redis.io/topics/streams-intro
glob (programming)
https://en.wikipedia.org/wiki/Glob_(programming)
Why and how Pricing Assistant migrated from Celery to RQ – Paris.py
https://www.slideshare.net/sylvinus/why-and-how-pricing-assistant-migrated-from-celery-to-rq-parispy-2
Enqueueing internals
http://python-rq.org/contrib/
queue — A synchronized queue class
https://docs.python.org/3/library/queue.html
Queue – A thread-safe FIFO implementation
https://pymotw.com/2/Queue/
Queues
http://queues.io/
Windows Subsystem for Linux Documentation
https://docs.microsoft.com/en-us/windows/wsl/about
RestMQ
http://restmq.com/
ActiveMQ
http://activemq.apache.org/
Amazon MQ
https://aws.amazon.com/amazon-mq/
Amazon Simple Queue Service
https://aws.amazon.com/sqs/
Celery: Distributed Task Queue
http://www.celeryproject.org/
Disque, an in-memory, distributed job queue
https://github.com/antirez/disque
rq-dashboard
https://github.com/eoranged/rq-dashboard
Projekt RQ na PyPi
https://pypi.org/project/rq/
rq-dashboard 0.3.12
https://pypi.org/project/rq-dashboard/
Job queue
https://en.wikipedia.org/wiki/Job_queue
Why we moved from Celery to RQ
https://frappe.io/blog/technology/why-we-moved-from-celery-to-rq
Running multiple workers using Celery
https://serverfault.com/questions/655387/running-multiple-workers-using-celery
celery — Distributed processing
http://docs.celeryproject.org/en/latest/reference/celery.html
Chains
https://celery.readthedocs.io/en/latest/userguide/canvas.html#chains
Routing
http://docs.celeryproject.org/en/latest/userguide/routing.html#automatic-routing
Celery Distributed Task Queue in Go
https://github.com/gocelery/gocelery/
Python Decorators
https://wiki.python.org/moin/PythonDecorators
Periodic Tasks
http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html
celery.schedules
http://docs.celeryproject.org/en/latest/reference/celery.schedules.html#celery.schedules.crontab
Pros and cons to use Celery vs. RQ
https://stackoverflow.com/questions/13440875/pros-and-cons-to-use-celery-vs-rq
Priority queue
https://en.wikipedia.org/wiki/Priority_queue
Jupyter
https://jupyter.org/
How IPython and Jupyter Notebook work
https://jupyter.readthedocs.io/en/latest/architecture/how_jupyter_ipython_work.html
Context Managers
http://book.pythontips.com/en/latest/context_managers.html

Seriál: Message brokery

Durable Queue aneb implementace front zpráv bez použití klasického message brokera
Nsq – systém pro doručování zpráv bez centrálního message brokera
NSQ – systém pro doručování zpráv bez centrálního message brokera (dokončení)

Přečtěte si všechny díly seriálu Message brokery nebo sledujte jeho RSS

Vstoupit do diskuse (8 názorů)

Pavel Tišnovský

Vystudoval VUT FIT a v současné době pracuje na projektech vytvářených v jazycích Python a Go.

Jj ta druhá část takto funguje, tam by problém být neměl. Akorát to hledání lookupu, jejich seznam, aktualizace toho seznamu v runtime, zkoušení, kterej je v provozu, přepojení na jinej při pádu - to může být problém (neříkám že všude a vždycky, ale už to není úplně triviální setup).

Pavel Tišnovský

Zlatý podporovatel

Sdílet

Obsah

1. Systém nsq se představuje

2. Architektura systému nsq

3. Doručení zprávy konzumentům

4. Překlad a instalace systému nsq

5. Instalace knihoven používaných klienty

6. Spuštění jednotlivých uzlů systému nsq

7. Používané porty

8. Poslání zprávy přes REST API

9. Komunikace s uzlem nsqlookupd přes REST API

10. Jednoduchý producent zpráv naprogramovaný v Pythonu

11. Konzument zpráv naprogramovaný v Pythonu

12. Použití systému nsq klienty naprogramovanými v Go

13. Producent zpráv naprogramovaný v jazyce Go

14. Využití bloku defer pro ukončení producenta zpráv

15. Jednoduchý konzument zpráv používající synchronizaci pro ukončení gorutiny s příjemcem

16. Použití kanálu ve funkci synchronizační struktury

17. Připojení klienta ke službě nsqlookupd

18. Obsah následující části seriálu

19. Repositář s demonstračními příklady

20. Odkazy na Internetu

Durable Queue aneb implementace front zpráv bez použití klasického message brokera