Komunikace s využitím knihovny nanomsg a programovacího jazyka Python

9. 5. 2019

Doba čtení: 37 minut

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

V závěrečném článku o knihovně nanomsg se seznámíme s použitím této knihovny z dalších jazyků, zejména z Pythonu. Taktéž si ukážeme, že některé operace je možné provádět přímo z příkazové řádky.

Obsah

1. Využití různých komunikačních strategií z příkazového řádku

2. Alternativní jména příkazů

3. Konfigurace adresy pro operace bind a connect

4. Využití nástroje nanocat v praxi

5. Použití knihovny nanomsg v aplikacích naprogramovaných v Pythonu

6. Jednosměrná komunikace využívající strategii PUSH-PULL (PIPELINE)

7. Zpracování většího množství zpráv při použití strategie PUSH-PULL

8. Jednosměrná komunikace mezi dvojicí uzlů s využitím strategie PAIR

9. Obousměrná komunikace mezi dvojicí uzlů s využitím strategie PAIR

10. Použití socketů ve funkci správce kontextu

11. Vytvoření klienta a serveru založených na strategii REQ-REP

12. Chování serveru i klientů při vyšší zátěži

13. Implementace klasické komunikační strategie PUB-SUB

14. Explicitní nastavení tématu (topic)

15. Příjemci s filtrací zpráv podle nastaveného tématu

16. Klient a server používající strategii SURVEY

17. Větší množství klientů – respondentů

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

19. Odkazy na předchozí části seriálu o message brokerech

20. Odkazy na Internetu

1. Využití různých komunikačních strategií z příkazového řádku

Nejprve si řekněme, jakým způsobem je možné knihovnu nanomsg využít přímo na příkazovém řádku, tj. buď v interaktivně zadávaných příkazech nebo ve skriptech naprogramovaných například v BASHi. Součástí instalace knihovny nanomsg je totiž i nástroj nazvaný nanocat, který se spouští přímo na příkazovém řádku a který lze nakonfigurovat takovým způsobem, že bude ve vytvářené aplikaci zastupovat prakticky libovolný uzel (server, klient atd.). Alternativně je možné nanomsg použít při testování jednotlivých částí aplikace, podobně jako používáme curl pro testování webových služeb. Základní způsoby jeho použití jsou vypsány v příslušné manuálové stránce, takže si jen v krátkosti uveďme, s jakými parametry je tuto utilitu možné spustit:

nanocat --req {--connect ADDR|--bind ADDR} {--data DATA|--file PATH} [-i SEC] [-AQ]
nanocat --rep {--connect ADDR|--bind ADDR} {--data DATA|--file PATH} [-AQ]
nanocat --push {--connect ADDR|--bind ADDR} {--data DATA|--file PATH} [-i SEC]
nanocat --pull {--connect ADDR|--bind ADDR} [-AQ]
nanocat --pub {--connect ADDR|--bind ADDR} {--data DATA|--file PATH} [-i SEC]
nanocat --sub {--connect ADDR|--bind ADDR} [--subscribe PREFIX ...] [-AQ]
nanocat --surveyor {--connect ADDR|--bind ADDR} {--data DATA|--file PATH} [-i SEC] [-AQ]
nanocat --respondent {--connect ADDR|--bind ADDR} {--data DATA|--file PATH} [-AQ]
nanocat --bus {--connect ADDR|--bind ADDR} {--data DATA|--file PATH} [-i SEC] [-AQ]
nanocat --pair {--connect ADDR|--bind ADDR} {--data DATA|--file PATH} [-i SEC] [-AQ]

Povšimněte si, že první přepínač vždy uvádí režim, v jakém se uzel (ať již ho budeme považovat za klienta či naopak za server) spustí. Podporováno je všech šest komunikačních strategií, s nimiž jsme se již seznámili v předchozích článcích, tj.:

#	Strategie	Stručný popis	První typ uzlu	Druhý typ uzlu
1	PAIR	jedna z nejjednodušších komunikačních strategií s dvojicí uzlů a vazbou 1:1	nanocat –pair	nanocat –pair
2	PIPELINE	jednosměrná komunikace buď s vazbami 1:1 (jeden vysílač a jeden přijímač), popř. mezi více vysílači a několika přijímači	nanocat –push	nanocat –pull
3	PUBSUB	klasická komunikační strategie PUB-SUB neboli PUBLISH-SUBSCRIBE	nanocat –pub	nanocat –sub
4	REQREP	klasická komunikační strategie REQ-REP neboli REQUEST-RESPONSE	nanocat –req	nanocat –rep
5	SURVEY	speciální strategie umožňující získat stav více uzlů (procesů) jediným dotazem a mnoha odpovědmi	nanocat –surveyor	nanocat –respondent
6	BUS	složitější strategie, v níž se používá obecnější vazba M:N	nanocat –bus	nanocat –bus

2. Alternativní jména příkazů

V případě korektní instalace se kromě vlastního nástroje nanocat vytvoří i řada symbolických odkazů (symlinků), které všechny ukazují na totožný binární soubor „nanocat“, ovšem již nevyžadují použití prvního přepínače, jelikož funkce uzlu je plně patrná ze jména symlinku:

nn_req {--connect ADDR|--bind ADDR} {--data DATA|--file PATH} [-i SEC] [-AQ]
nn_rep {--connect ADDR|--bind ADDR} {--data DATA|--file PATH} [-AQ]
nn_push {--connect ADDR|--bind ADDR} {--data DATA|--file PATH} [-i SEC]
nn_pull {--connect ADDR|--bind ADDR} [-AQ]
nn_pub {--connect ADDR|--bind ADDR} {--data DATA|--file PATH} [-i SEC]
nn_sub {--connect ADDR|--bind ADDR} [--subscribe PREFIX ...] [-AQ]
nn_surveyor {--connect ADDR|--bind ADDR} {--data DATA|--file PATH} [-i SEC] [-AQ]
nn_respondent {--connect ADDR|--bind ADDR} {--data DATA|--file PATH} [-AQ]
nn_bus {--connect ADDR|--bind ADDR} {--data DATA|--file PATH} [-i SEC] [-AQ]
nn_pair {--connect ADDR|--bind ADDR} {--data DATA|--file PATH} [-i SEC] [-AQ]

Význam jednotlivých symbolických odkazů je pravděpodobně zřejmý z jejich názvů:

#	Přepínač pro nanocat	Název symlinku	Typ socketu
1	--req	nn_req	NN_REQ
2	--rep	nn_rep	NN_REP
3	--push	nn_push	NN_PUSH
4	--pull	nn_pull	NN_PULL
5	--pub	nn_pub	NN_PUB
6	--sub	nn_sub	NN_SUB
7	--surveyor	nn_surveyor	NN_SURVEYOR
8	--respondent	nn_respondent	NN_RESPONDENT
9	--bus	nn_bus	NN_BUS
10	--pair	nn_pair	NN_PAIR

3. Konfigurace adresy pro operace bind a connect

I význam dalšího přepínače zadaného na příkazovém řádku je zřejmý:

--connect ADDR
--bind ADDR

Tímto přepínačem se určuje režim otevření socketu. Buď se použije funkce nn_bind(), která otevře port, na němž bude nástroj naslouchat a čekat na připojení dalších uzlů. Nebo se naopak použije funkce nn_connect() pro připojení k otevřenému portu. To, která konkrétní metoda se použije, vyplývá z toho, které uzly považujeme za „servery“ a které za „klienty“. Typicky bývá serverem ten uzel, který je z hlediska architektury celé aplikace stabilnější, především z hlediska jeho dohledatelnosti a dostupnosti. Jak již víme z předchozích dvou článků, není nanomsg plnohodnotným message brokerem (i když je možné message brokera na nanomsg postavit), takže funkci centrálního uzlu většinou přenášíme na uzly jiné. To může způsobovat problémy v těch případech, kdy jsou funkce komunikujících uzlů symetrické.

Existují i další varianty těchto přepínačů, které se již vztahují k jednotlivým typům podporovaných komunikačních kanálů. Připomeňme si, že knihovna nanomsg podporuje následující způsoby komunikace jednotlivých uzlů:

#	Přenosový mechanismus	Stručný popis
1	INPROC	komunikace v rámci jednoho procesu, například typizovaná komunikace mezi několika vlákny
2	IPC	komunikace mezi několika procesy běžícími na jednom počítači
3	TCP	komunikace mezi procesy běžícími na různých počítačích s využitím protokolu TCP
4	WS	komunikace mezi procesy běžícími na různých počítačích s využitím web socketů

A právě pro dva velmi často používané komunikační kanály je možné použít speciální formu přepínače, který současně volí jak typ komunikačního kanálu, tak i příslušnou adresu (jejíž formát je pochopitelně konkrétně spjatý s určitým kanálem):

Typ kanálu	Přepínač pro nn_bind()	Přepínač pro nn_connect()
IPC	--bind-ipc,–X	--connect-ipc,–x
TCP na localhost	--bind-local,–L	--connect-local,–l

4. Využití nástroje nanocat v praxi

Podívejme se nyní na způsob použití jednotlivých komunikačních strategií. Příklady, které si uvedeme v dalším textu, budou zhruba odpovídat svým céčkovým protějškům, které byly popsány minule a předminule.

Spuštění serveru používajícího strategii PUSH-PULL:

$ nanocat --pull -Q -v --bind ipc:///tmp/example

Poslání (PUSH) zprávy na server:

$ nanocat --push --connect ipc:///tmp/example --data "test"

Poslání obsahu celého souboru na server:

$ nanocat --push --connect ipc:///tmp/example --file .bashrc

Spuštění příjemce zpráv založeného na strategii PUB-SUB:

$ nanocat --sub -Q -v --bind ipc:///tmp/example

Periodické posílání zprávy každou sekundu:

$ nanocat --pub --connect ipc:///tmp/example -i 1 --data "test"

V případě potřeby můžeme změnit i formát výpisu zpráv, namísto řetězců (uzavřených do uvozovek) se může zapnout hexadecimální výstup:

$ nanocat --sub --hex -v --bind ipc:///tmp/example 
 
"\x74\x65\x73\x74"
"\x74\x65\x73\x74"

Výpis přijatých zpráv ve formě ASCII textů:

$ nanocat --sub --ascii -v --bind ipc:///tmp/example 
 
test
test

Podobný příklad, ovšem používající jednodušší komunikační strategii PAIR:

$ nanocat --pair -Q -v --bind ipc:///tmp/example

druhý uzel (klient):

$ nanocat --pair --connect ipc:///tmp/example -i 1 --data "test"

5. Použití knihovny nanomsg v aplikacích naprogramovaných v Pythonu

Ve druhé části dnešního článku si popíšeme, jakým způsobem je možné použít knihovnu nanomsg z programovacího jazyka Python. Python pochopitelně není jediným programovacím jazykem, pro nějž vznikla knihovna s rozhraním pro nanomsg, protože kromě nativního céčka (nanomsg je psána totiž psána, jak již ostatně víme, v čistém C) existují rozhraní i pro tyto programovací jazyky, popř. platformy:

Jazyk (platforma)	Knihovna/projekt s rozhraním
C	nanomsg
	NNG (nová reimplementace, viz poznámka v úvodní kapitole)
C++	nanomsgxx
	cppnanomsg
	nngpp (pro NNG)
Clojure	jnanomsg (voláno přes standardní Java interop)
D	nanomsg-wrapper
Dylan	nanomsg-dylan
Erlang	enm
Fortran	nanofort
Go	mangos (reimplementace v Go)
	mangos v2 (druhá verze)
	go-nanomsg
Haskell	nanomsg-haskell
	nanomsg
Haxe	hx-nanomsg
Swift	swiftc nanomsg
Java	jnano
	jnanomsg
	nngjvm (pro NNG)
JavaScript (Node.js)	node-nanomsg
Lua	lua-nanomsg
	luajit-nanomsg (pro systém LuaJIT)
	luananomsg
.NET	NNanomsg
Ocaml	onanomsg
Perl	NanoMsg::Raw
PHP	php-nano
PicoLisp	picolisp-nanomsg FFI bindings
Python	nanomsg-python
	pynanomsg
	nnpy
	pynng (pro NNG, prozatím ve vývoji)
R	rnanomsg
Ruby	nn-core
	nanomsg
Rust	rust-nanomsg
	nng-rs (opět pro NNG)
Scheme (CHICKEN)	chicken-nanomsg
Smalltalk	NanoStrand

V případě Pythonu existuje několik rozhraní mezi tímto programovacím jazykem (přesněji řečeno mezi aplikacemi psanými v Pythonu) a knihovnou nanomsg, ovšem nejvíce doporučovaná je knihovna nazvaná jednoduše nanomsg-python, kterou naleznete na adrese https://github.com/tonysimpson/nanomsg-python. Tato knihovna je pochopitelně dostupná i na PyPi, takže její instalace může proběhnout naprosto stejným způsobem, jaký jsme si již popsali u jiných pythoních knihoven – pomocí nástroje pip, popř. pip3:

$ pip3 install nanomsg

Poznámka: před spuštěním demonstračních příkladů popsaných v navazujících kapitolách je nutné zajistit, aby příklady našly binární verzi knihovny nanomsg:

export LD_LIBRARY_PATH=/usr/local/lib64/

Poznámka²: ve všech příkladech použijeme komunikační kanál funkční pro procesy běžící na stejném počítači. Použije se sdílený Unixový socket (speciální typ souboru) a adresa bude začínat prefixem „ipc:///“, za nímž se uvede jméno dočasného speciálního souboru představujícího socket.

6. Jednosměrná komunikace využívající strategii PUSH-PULL (PIPELINE)

Strategie nazvaná PIPELINE zajišťuje jednosměrný přenos zpráv od vysílající aplikace (vlákna, procesu) k aplikaci přijímající. V tom nejjednodušším případě existuje pouze jediný vysílač (neboli zdroj zpráv, producent) a jediný přijímač (konzument). Nejprve se podívejme na implementaci producenta, která je nepatrně jednodušší, zejména s ohledem na to, že se nemusí žádným způsobem pracovat s bufferem pro příjem zprávy.

Implementace zdroje zpráv je v případě Pythonu velmi krátká (pokud ovšem prozatím vynecháme kontroly chybových stavů). Zdrojový kód producenta zpráv naleznete na adrese https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/01_sender_receiver/sender.py:

from nanomsg import Socket, PUSH
 
URL = "ipc:///tmp/example1"
 
socket = Socket(PUSH)
print("Socket created")
 
socket.connect(URL)
print("Connected to URL {}".format(URL))
 
socket.send("Hello world!")
print("Message has been sent")
 
socket.close()
print("Socket closed")

Podobným způsobem je pochopitelně možné realizovat i klienta, který bude zprávy přijímat. Vysílací strana používá socket typu PUSH, strana přijímací tedy použije socket PULL (což je kontrolováno při navazování připojení) a výsledný zdrojový kód příjemce zpráv bude vypadat následovně:

from nanomsg import Socket, PULL
 
URL = "ipc:///tmp/example1"
 
socket = Socket(PULL)
print("Socket created")
 
socket.bind(URL)
print("Bound to URL {}".format(URL))
 
message = socket.recv()
print("Received: {}".format(message))
 
socket.close()
print("Socket closed")

Pro zajímavost se podívejme na to, jak jsme předminule implementovali zdroj zpráv, tentokrát v céčku. Oproti pythoní verzi je příklad rozsáhlejší:

#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <nanomsg/nn.h>
#include <nanomsg/pipeline.h>
 
const char *URL = "ipc:///tmp/example1";
 
void sender(const char *url, const char *message)
{
    int message_size = strlen(message) + 1;
    int socket;
    int endpoint;
    int bytes;
 
    socket = nn_socket(AF_SP, NN_PUSH);
    puts("Socket created");
 
    endpoint = nn_connect(socket, url);
    puts("Remote endpoint added to the socket");
 
    printf("Sending message '%s'\n", message);
    bytes = nn_send(socket, message, message_size, 0);
 
    printf("Message with length %d bytes sent, flushing", bytes);
    sleep(1);
    puts("Done");
 
    nn_shutdown(socket, endpoint);
}
 
int main(const int argc, const char **argv)
{
    sender(URL, "Hello");
    sender(URL, "world");
    sender(URL, "!");
 
    return 0;
}

Příjemce zpráv, opět vytvořený v céčku, by mohl vypadat následovně:

#include <stdio.h>
#include <nanomsg/nn.h>
#include <nanomsg/pipeline.h>
 
const char *URL = "ipc:///tmp/example1";
 
void receiver(const char *url)
{
    int socket;
 
    socket = nn_socket(AF_SP, NN_PULL);
    puts("Socket created");
 
    nn_bind(socket, url);
    puts("Endpoint bound to socket");
 
    puts("Waiting for messages...");
    while (1) {
        char *message = NULL;
        int bytes = nn_recv(socket, &message, NN_MSG, 0);
        printf("Received message '%s' with length %d bytes\n", message, bytes);
        nn_freemsg(message);
    }
}
 
int main(int argc, char **argv)
{
    receiver(URL);
    return 0;
}

7. Zpracování většího množství zpráv při použití strategie PUSH-PULL

Samozřejmě si můžeme naprogramovat i klienta, který serveru pošle větší množství zpráv s využitím strategie PUSH-PULL (PIPELINE). Mezi zprávy je vhodné vkládat nepatrné pauzy, aby se korektně vyprázdnily síťové buffery a zpráva byla skutečně přenesena na server (to je principiální problém nanomsg, s nímž se taktéž seznámili minule a předminule).

Zdrojový kód zdroje deseti zpráv poslaných příjemci (klientovi) naleznete na adrese https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/02_sender_receiver_more_messages/sender.py:

from nanomsg import Socket, PUSH
from time import sleep
 
URL = "ipc:///tmp/example2"
 
socket = Socket(PUSH)
print("Socket created")
 
socket.connect(URL)
print("Connected to URL {}".format(URL))
 
for i in range(10):
    socket.send("Hello world #{}".format(i+1))
    print("Message has been sent")
    sleep(0.1)
 
socket.close()
print("Socket closed")

Server, který zprávy bude přijímat, lze naprogramovat například následujícím způsobem. Povšimněte si, že zprávy jsou zpracovávány v nekonečné programové smyčce a v jediném vláknu (což je pochopitelně to nejjednodušší řešení, v němž ovšem neprovádíme korektní uzavření socketu).

Zdrojový kód příjemce zpráv je uložen na adrese https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/02_sender_receiver_more_messages/receiver.py:

from nanomsg import Socket, PULL
 
URL = "ipc:///tmp/example2"
 
socket = Socket(PULL)
print("Socket created")
 
socket.bind(URL)
print("Bound to URL {}".format(URL))
 
while True:
    message = socket.recv()
    print("Received: {}".format(message))
 
socket.close()
print("Socket closed")

8. Jednosměrná komunikace mezi dvojicí uzlů s využitím strategie PAIR

Druhá komunikační strategie se jmenuje PAIR. Tato strategie umožňuje, aby mezi sebou jednotlivé uzly komunikovaly oboustranně, což nebylo při použití strategie PUSH-PULL/PIPELINE možné (přesněji řečeno to možné bylo, ovšem otevřením dvou komunikačních kanálů, což je velmi křehké řešení). I u strategie PAIR vystupuje jeden z komunikujících uzlů ve funkci serveru (otevírá svůj port a očekává, že se na něj připojí klient) a druhý uzel ve funkci klienta. Ovšem až na tento rozdíl jsou si po navázání spojení oba uzly rovnocenné, tj. každý z nich může vysílat i přijímat zprávy, a to libovolným způsobem, který si zvolí sám programátor. Ten například může implementovat jednoduchý systém typu dotaz-odpověď (což je ovšem lepší realizovat strategií REQREP popsanou níže) či skutečně použít plnohodnotný oboustranný komunikační kanál.

Opět se podívejme na úplný zdrojový kód prvního uzlu, který po svém spuštění pošle zprávu uzlu druhému a ihned poté se ukončí. Jedná se tedy o čistě jednosměrnou komunikaci:

from nanomsg import Socket, PAIR
 
URL = "ipc:///tmp/example3"
 
socket = Socket(PAIR)
print("Socket created")
 
socket.bind(URL)
print("Bound to URL {}".format(URL))
 
socket.send("Hello world!")
print("Message has been sent")
 
socket.close()
print("Socket closed")

Druhý uzel se pouze pokusí přijmout zprávu od uzlu prvního. Na tuto zprávu nijak neodpovídá, pouze ji zobrazí na standardním výstupu. Zatímco první uzel vystupoval v roli serveru, je druhý uzel klientem. Následuje výpis zdrojového kódu druhého uzlu:

from nanomsg import Socket, PAIR
 
URL = "ipc:///tmp/example3"
 
socket = Socket(PAIR)
print("Socket created")
 
socket.connect(URL)
print("Connected to URL {}".format(URL))
 
message = socket.recv()
print("Received: {}".format(message))
 
socket.close()
print("Socket closed")

9. Obousměrná komunikace mezi dvojicí uzlů s využitím strategie PAIR

Vzhledem k tomu, že v komunikační strategii PAIR jsou si oba uzlu rovnocenné v tom, že každý může zprávy přijímat i odesílat, můžeme předchozí příklad upravit takovým způsobem, že první uzel nejdříve pošle zprávu o poté si přečte odpověď. Tento příklad naleznete na adrese https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/04_two_way_pair_communication/node1.py:

from nanomsg import Socket, PAIR
 
URL = "ipc:///tmp/example4"
 
socket = Socket(PAIR)
print("Socket created")
 
socket.bind(URL)
print("Bound to URL {}".format(URL))
 
socket.send("Hello world!")
print("Message has been sent")
 
message = socket.recv()
print("Received response: {}".format(message))
 
socket.close()
print("Socket closed")

Druhý uzel musí postupovat přesně naopak – nejdříve zprávu přijmout a poté na ni odpovědět. V kódu tedy bude volání metod recv a send prohozené, jak je to ostatně patrné již při pohledu na jeho zdrojový kód:

from nanomsg import Socket, PAIR
 
URL = "ipc:///tmp/example4"
 
socket = Socket(PAIR)
print("Socket created")
 
socket.connect(URL)
print("Connected to URL {}".format(URL))
 
message = socket.recv()
print("Received: {}".format(message))
 
socket.send("Thanks for your message '{}'".format(message))
print("Response has been sent")
 
socket.close()
print("Socket closed")

Poznámka: ve skutečnosti uzly zcela rovnocenné nebudou, a to z toho prostého důvodu, že jeden z nich musí být serverem a druhý klientem.

10. Použití socketů ve funkci správce kontextu

Volání dvojice párových příkazů Socket() a socket.close() je možné ve skutečnosti naprogramovat ještě jednodušším a mnohem čitelnějším způsobem. Samotnou instanci socketu lze použít ve funkci správce kontextu (context manager), což konkrétně v programovacím jazyku Python znamená, že se socket vytvoří v konstrukci with a na konci této konstrukce (přesněji řečeno na konci celého bloku with) se socket automaticky uzavře.

První komunikující uzel z předchozí kapitoly tedy můžeme přepsat takto:

from nanomsg import Socket, PAIR
 
URL = "ipc:///tmp/example5"
 
with Socket(PAIR) as socket:
    print("Socket created")
 
    socket.bind(URL)
    print("Bound to URL {}".format(URL))
 
    socket.send("Hello world!")
    print("Message has been sent")
 
    message = socket.recv()
    print("Received response: {}".format(message))
 
print("Socket closed")

Podobně můžeme přepsat i druhý uzel, opět s použitím with:

from nanomsg import Socket, PAIR
 
URL = "ipc:///tmp/example5"
 
with Socket(PAIR) as socket:
    socket = Socket(PAIR)
    print("Socket created")
 
    socket.connect(URL)
    print("Connected to URL {}".format(URL))
 
    message = socket.recv()
    print("Received: {}".format(message))
 
    socket.send("Thanks for your message '{}'".format(message))
    print("Response has been sent")
 
print("Socket closed")

11. Vytvoření klienta a serveru založených na strategii REQ-REP

Další velmi často používaná komunikační strategie se nazývá Request-Response nebo zkráceně REQ-REP popř. pouze REQREP. Při použití této strategie spolu komunikují dva typy uzlů – server a teoreticky neomezené množství klientů. Server přijímá požadavky (request) a odpovídá na ně (response), přičemž je možné, aby požadavky posílalo několik klientů (a jeden klient naopak může v případě potřeby posílat požadavky více serverům). Tato velmi asymetrická komunikace se strategií REQ-REP je v praxi poměrně častá, ostatně je na ní založen i známý protokol HTTP a jeho pozdější varianty (dokonce se někdy nad HTTP staví jiné komunikační strategie, což je ovšem neoptimální řešení).

Nejprve si ukažme, jak lze naprogramovat klienta, který musí použít socket typu REQ a celá komunikace vždy probíhá ve dvojici příkazů socket.send(), který musí být následovaný příkazem socket.recv():

from nanomsg import Socket, REQ
 
URL = "ipc:///tmp/example6"
 
 
def send_request(socket, request):
    socket.send(request)
 
 
def receive_response(socket):
    response = socket.recv()
    print("Received response: '{}'".format(response))
 
 
with Socket(REQ) as socket:
    print("Socket created")
 
    socket.connect(URL)
    print("Connected to URL {}".format(URL))
 
    send_request(socket, "Hello from 'first'!")
    print("Waiting for response...")
    receive_response(socket)
 
print("Socket closed")

Server je naproti tomu naprogramován symetricky – na každou přijatou zprávu (socket.recv() odpovídá pomocí socket.send(). Zdrojový kód serveru vypadá takto:

from nanomsg import Socket, REP
 
URL = "ipc:///tmp/example6"
 
 
def receive_request(socket):
    request = socket.recv()
    print("Received request: '{}'".format(request))
 
 
def send_response(socket, response):
    socket.send(response)
 
 
with Socket(REP) as socket:
    print("Socket created")
 
    socket.bind(URL)
    print("Bound to URL {}".format(URL))
 
    print("Waiting for requests...")
    while True:
        receive_request(socket)
        send_response(socket, "ACK!")
 
print("Socket closed")

12. Chování serveru i klientů při vyšší zátěži

Strategie REQ-REP je v knihovně nanomsg velmi robustní, takže by nemělo docházet ke ztrátě zpráv. Ostatně si to můžeme velmi snadno otestovat. Nejdříve upravíme server takovým způsobem, aby počítal všechny přijaté zprávy, což je triviální úprava:

from nanomsg import Socket, REP
 
URL = "ipc:///tmp/example7"
 
 
def receive_request(socket):
    request = socket.recv()
    print("Received request: '{}'".format(request))
 
 
def send_response(socket, response):
    socket.send(response)
 
 
# pocitadlo pozadavku
received = 0
 
with Socket(REP) as socket:
    print("Socket created")
 
    socket.bind(URL)
    print("Bound to URL {}".format(URL))
 
    print("Waiting for requests...")
    while True:
        receive_request(socket)
        received += 1
        print("Received {}th request".format(received))
        send_response(socket, "ACK!")
 
print("Socket closed")

Dále vytvoříme dvojici klientů. První z nich pošle 1000 zpráv:

from nanomsg import Socket, REQ
 
URL = "ipc:///tmp/example7"
 
 
def send_request(socket, request):
    socket.send(request)
 
 
def receive_response(socket):
    response = socket.recv()
    print("Received response: '{}'".format(response))
 
 
with Socket(REQ) as socket:
    print("Socket created")
 
    socket.connect(URL)
    print("Connected to URL {}".format(URL))
 
    for i in range(1000):
        send_request(socket, "Hello #{} from 'first'!".format(i))
        print("Waiting for response...")
        receive_response(socket)
 
print("Socket closed")

Druhý taktéž pošle 1000 zpráv, bez jakýchkoli pauz mezi nimi:

from nanomsg import Socket, REQ
 
URL = "ipc:///tmp/example7"
 
 
def send_request(socket, request):
    socket.send(request)
 
 
def receive_response(socket):
    response = socket.recv()
    print("Received response: '{}'".format(response))
 
 
with Socket(REQ) as socket:
    print("Socket created")
 
    socket.connect(URL)
    print("Connected to URL {}".format(URL))
 
    for i in range(1000):
        send_request(socket, "Hello #{} from 'second'!".format(i))
        print("Waiting for response...")
        receive_response(socket)
 
print("Socket closed")

Otestování může proběhnout způsobem:

$ python3 server.py > server.out&
$ python3 client1.py & python3 client2.py &

Server by měl zpracovat přesně 2000 zpráv.

13. Implementace klasické komunikační strategie PUB-SUB

V této kapitole si ukážeme způsob využití komunikační strategie PUBSUB, s níž jsme se již v tomto seriálu mnohokrát setkali. Tato strategie umožňuje rozesílat zprávy libovolnému množství příjemců. Zdroj zpráv bude v tomto případě implementován jako server a typ použitého socketu bude PUB. Podívejme se na zdrojový kód producenta/zdroje zpráv:

from time import sleep
from nanomsg import Socket, PUB
 
URL = "ipc:///tmp/example8"
 
 
with Socket(PUB) as socket:
    print("Socket created")
 
    socket.bind(URL)
    print("Bound to URL {}".format(URL))
 
    for i in range(1, 1000):
        message = "Message #{}".format(i)
        print("Publishing message {}".format(message))
        socket.send(message)
        sleep(0.5)
 
print("Socket closed")

Nejjednodušší, ovšem nefunkční :-) varianta příjemce zpráv se vytvoří nepatrnou úpravou předchozích příkladů. Problém ovšem spočívá v tom, že u komunikační strategie PUB-SUB je nutné specifikovat téma (topic), protože v opačném případě budou všechny zprávy zahozeny. Nicméně se podívejme na úplný zdrojový kód tohoto NEFUNKČNÍHO příkladu:

from time import sleep
from nanomsg import Socket, SUB, SUB_SUBSCRIBE
 
URL = "ipc:///tmp/example8"
 
 
def receive_message(socket):
    message = socket.recv()
    print("Received message: '{}'".format(message))
 
 
with Socket(SUB) as socket:
    print("Socket created")
 
    socket.connect(URL)
    print("Connected to URL {}".format(URL))
 
    while True:
        receive_message(socket)
 
print("Socket closed")

14. Explicitní nastavení tématu (topic)

Aby klient dokázal přijímat všechny zpráv s libovolným tématem, musíme nastavit následující volbu:

socket.set_string_option(SUB, SUB_SUBSCRIBE, "")

Tato volba určuje, že se mají přijmout a zpracovat naprosto všechny zprávy, bez ohledu na nastavené téma.

from time import sleep
from nanomsg import Socket, SUB, SUB_SUBSCRIBE
 
URL = "ipc:///tmp/example9"
 
 
def receive_message(socket):
    message = socket.recv()
    print("Received message: '{}'".format(message))
 
 
with Socket(SUB) as socket:
    print("Socket created")
 
    socket.connect(URL)
    socket.set_string_option(SUB, SUB_SUBSCRIBE, "")
    print("Connected to URL {}".format(URL))
 
    while True:
        receive_message(socket)
 
print("Socket closed")

15. Příjemci s filtrací zpráv podle nastaveného tématu

Pouze pro úplnost si ukažme, jak se budou zpracovávat zprávy s nastaveným tématem. Zdroj zpráv po svém spuštění vytvoří 2000 zpráv, přičemž tisíc z nich bude začínat prefixem „Message A“ a další tisíc prefixem „Message B“:

from time import sleep
from nanomsg import Socket, PUB
 
URL = "ipc:///tmp/example10"
 
 
with Socket(PUB) as socket:
    print("Socket created")
 
    socket.bind(URL)
    print("Bound to URL {}".format(URL))
 
    for i in range(1, 1000):
        message = "Message A #{}".format(i)
        print("Publishing message {}".format(message))
        socket.send(message)
        sleep(0.5)
        message = "Message B #{}".format(i)
        print("Publishing message {}".format(message))
        socket.send(message)
        sleep(0.5)
 
print("Socket closed")

První příjemce je nakonfigurován takovým způsobem, aby přijímal (filtroval) pouze zprávy s prefixem „Message A“:

from time import sleep
from nanomsg import Socket, SUB, SUB_SUBSCRIBE
 
URL = "ipc:///tmp/example10"
 
 
def receive_message(socket):
    message = socket.recv()
    print("Received message: '{}'".format(message))
 
 
with Socket(SUB) as socket:
    print("Socket created")
 
    socket.connect(URL)
    socket.set_string_option(SUB, SUB_SUBSCRIBE, "Message A")
    print("Connected to URL {}".format(URL))
 
    while True:
        receive_message(socket)
 
print("Socket closed")

Druhý příjemce je nakonfigurován takovým způsobem, aby přijímal (filtroval) pouze zprávy s prefixem „Message B“:

from time import sleep
from nanomsg import Socket, SUB, SUB_SUBSCRIBE
 
URL = "ipc:///tmp/example10"
 
 
def receive_message(socket):
    message = socket.recv()
    print("Received message: '{}'".format(message))
 
 
with Socket(SUB) as socket:
    print("Socket created")
 
    socket.connect(URL)
    socket.set_string_option(SUB, SUB_SUBSCRIBE, "Message B")
    print("Connected to URL {}".format(URL))
 
    while True:
        receive_message(socket)
 
print("Socket closed")

Po spuštění by mělo být vidět, že filtrace zpráv skutečně probíhá na základě nastavených filtrů.

16. Klient a server používající strategii SURVEY

V pořadí pátou komunikační strategií, se kterou jsme se seznámili minule, je strategie nazvaná Survey. Jedná se o dosti příhodný název, protože v této strategii vystupuje jeden uzel, který se dotáže ostatních uzlů například na jejich stav a uzly následně mohou odpovídat. Dotazující se uzel má nastavenu dobu čekání, takže může jednotlivé odpovědi zaznamenat a po uplynutí zadaného časového okamžiku zjistit, které uzly vůbec neodpověděly (nebo to nestihly, což je ovšem z pohledu dotazujícího to samé).

Server nejdříve deset sekund počká na připojení klientů, pošle jim otázku a zpracuje jejich případné odpovědi:

from nanomsg import Socket, SURVEYOR
from time import sleep
 
URL = "ipc:///tmp/example11"
 
 
def send_survey(socket, message):
    socket.send(message)
 
 
def receive_answer(socket):
    answer = socket.recv()
    print("Received answer: '{}'".format(answer))
 
 
def wait_for_clients(seconds):
    print("Waiting for clients to connect...")
    for i in range(seconds, 0, -1):
        print(i)
        sleep(1)
 
 
with Socket(SURVEYOR) as socket:
    print("Socket created")
 
    socket.bind(URL)
    print("Bound to URL {}".format(URL))
 
    wait_for_clients(10)
 
    send_survey(socket, "What do you get when you multiply six by nine?")
    print("Survey send, waiting for answers...")
 
    answers = 0
 
    while True:
        receive_answer(socket)
        answers += 1
        print("Processed {} answers so far".format(answers))
 
print("Socket closed")

Klient (respondent) pro každou otázku přijatou od serveru vygeneruje příslušnou odpověď:

from nanomsg import Socket, RESPONDENT
from random import seed, randint
 
URL = "ipc:///tmp/example11"
 
 
def receive_question(socket):
    question = socket.recv()
    print("Received question: '{}'".format(question))
 
 
def send_answer(socket, answer):
    print("Sending answer: '{}'".format(answer))
    socket.send(answer)
 
 
seed(None)
 
with Socket(RESPONDENT) as socket:
    print("Socket created")
 
    socket.connect(URL)
    print("Connected to URL {}".format(URL))
 
    while True:
        receive_question(socket)
        print("Question received")
 
        number = randint(0, 100)
        answer = "It must be {}".format(number)
 
        send_answer(socket, answer)
        print("Answer sent")
 
print("Socket closed")

17. Větší množství klientů – respondentů

Samozřejmě, že je možné naprogramovat větší množství respondentů, které serveru budou na dotazy odpovídat. Druhý respondent:

from nanomsg import Socket, RESPONDENT
from random import seed, randint
 
URL = "ipc:///tmp/example12"
 
 
def receive_question(socket):
    question = socket.recv()
    print("Received question: '{}'".format(question))
 
 
def send_answer(socket, answer):
    print("Sending answer: '{}'".format(answer))
    socket.send(answer)
 
 
seed(None)
 
with Socket(RESPONDENT) as socket:
    print("Socket created")
 
    socket.connect(URL)
    print("Connected to URL {}".format(URL))
 
    while True:
        receive_question(socket)
        print("Question received")
 
        number = randint(0, 100)
        answer = "Hello, I'm client #1. It must be {}".format(number)
 
        send_answer(socket, answer)
        print("Answer sent")
 
print("Socket closed")

Třetí respondent:

from nanomsg import Socket, RESPONDENT
from random import seed, randint
 
URL = "ipc:///tmp/example12"
 
 
def receive_question(socket):
    question = socket.recv()
    print("Received question: '{}'".format(question))
 
 
def send_answer(socket, answer):
    print("Sending answer: '{}'".format(answer))
    socket.send(answer)
 
 
seed(None)
 
with Socket(RESPONDENT) as socket:
    print("Socket created")
 
    socket.connect(URL)
    print("Connected to URL {}".format(URL))
 
    while True:
        receive_question(socket)
        print("Question received")
 
        number = randint(0, 100)
        answer = "Hello, I'm client #2. It must be {}".format(number)
 
        send_answer(socket, answer)
        print("Answer sent")
 
print("Socket closed")

Pořadí spuštění by mělo být následující:

Server (začne odpočítávat deset sekund do položení otázky)
Všichni klienty (například spuštění na pozadí ve stejném terminálu)

18. 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 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á doslova několik kilobajtů), můžete namísto toho použít odkazy na jednotlivé příklady, které naleznete v následující tabulce. Každý příklad se skládá ze dvou samostatně překládaných a spouštěných souborů – producenta zpráv a konzumenta zpráv:

Příklad	Skript/kód	Popis	Cesta
1	sender.py	zdroj jedné zprávy používající strategii PUSH-PULL (PIPELINE)	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/01_sender_receiver/sender.py
1	receiver.py	příjemce jedné zprávy používající strategii PUSH-PULL (PIPELINE)	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/01_sender_receiver/receiver.py

2	sender.py	zdroj deseti zpráv používající strategii PUSH-PULL (PIPELINE)	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/02_sender_receiver_more_messages/sender.py
2	receiver.py	příjemce libovolného množství zpráv používající strategii PUSH-PULL (PIPELINE)	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/02_sender_receiver_more_messages/receiver.py

3	node1.py	jednosměrná komunikace s využitím strategie PAIR	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/03_pair_communication/node1.py
3	node2.py	jednosměrná komunikace s využitím strategie PAIR	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/03_pair_communication/node2.py

4	node1.py	obousměrná komunikace s využitím strategie PAIR	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/04_two_way_pair_communication/node1.py
4	node2.py	obousměrná komunikace s využitím strategie PAIR	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/04_two_way_pair_communication/node2.py

5	node1.py	použití socketu ve funkci správce kontextu	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/05_socket_as_context_manager/node1.py
5	node2.py	použití socketu ve funkci správce kontextu	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/05_socket_as_context_manager/node2.py

6	client.py	klient používající strategii REQ-REP	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/06_reqest_response/client.py
6	server.py	server používající strategii REQ-REP	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/06_reqest_response/server.py

7	server.py	server používající strategii REQ-REP	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/07_reqest_response_more_load/server.py
7	client1.py	první klient používající strategii REQ-REP	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/07_reqest_response_more_load/client1.py
7	client2.py	druhý klient používající strategii REQ-REP	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/07_reqest_response_more_load/client2.py

8	publisher.py	zdroj zpráv pro strategii PUB-SUB	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/08_pub_sub/publisher.py
8	subscriber.py	příjemce zpráv pro strategii PUB-SUB bez filtrace zpráv	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/08_pub_sub/subscriber.py

9	publisher.py	zdroj zpráv pro strategii PUB-SUB	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/09_pub_sub_topic/publisher.py
9	subscriber.py	příjemce zpráv pro strategii PUB-SUB s nastavením filtrace zpráv	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/09_pub_sub_topic/subscriber.py

10	publisher.py	zdroj zpráv pro strategii PUB-SUB	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/10_pub_sub_topics/publisher.py
10	subscriberA.py	příjemce zpráv pro strategii PUB-SUB s filtrem zpráv	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/10_pub_sub_topics/subscriberA.py
10	subscriberB.py	příjemce zpráv pro strategii PUB-SUB s filtrem zpráv	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/10_pub_sub_topics/subscriberB.py

11	server.py	dotazující se server (SURVEYOR)	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/11_survey/server.py
11	client.py	klient (RESPONDENT) odpovídající na dotazy	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/11_survey/client.py

12	server.py	dotazující se server (SURVEYOR)	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/12_survey_more_clients/server.py
12	client1.py	klient (RESPONDENT) odpovídající na dotazy	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/12_survey_more_clients/client1.py
12	client2.py	klient (RESPONDENT) odpovídající na dotazy	https://github.com/tisnik/message-queues-examples/blob/master/nanomsg-python/12_survey_more_clients/client2.py

19. Odkazy na předchozí části seriálu o message brokerech

V této kapitole jsou uvedeny odkazy na všech sedmnáct předchozích částí seriálu, v němž se zabýváme různými způsoby implementace front zpráv a k nim přidružených technologií message brokerů: