\nWysy\u0142anie \u017c\u0105dania<\/p>\n
<\/p>\n
messaging:request(Exchange, ResponseMatchingTag, RequestDefinition, HandlerProcess).<\/code><\/pre>\n<\/p>\n
Exchange<\/em> \u2012 unikalna nazwa punktu wymiany
\nResponseMatchingTag<\/em> \u2012 lokalna etykieta do przetwarzania odpowiedzi. Na przyk\u0142ad w przypadku wysy\u0142ania wielu identycznych \u017c\u0105da\u0144 od r\u00f3\u017cnych u\u017cytkownik\u00f3w.
\nRequestDefinition<\/em> \u2012 tre\u015b\u0107 \u017c\u0105dania
\nHandlerProcess<\/em> \u2012 PID przetwarzacza. Ten proces otrzyma odpowied\u017a od serwera.<\/p>\n\n<\/li>\n
\nPrzetwarzanie odpowiedzi<\/p>\n
<\/p>\n
handle_info(#'$msg'{exchange = EXCHANGE, tag = ResponseMatchingTag,message = ResponsePayload}, State)<\/code><\/pre>\n<\/p>\n
ResponsePayload<\/em> \u2012 odpowied\u017a serwera.<\/p>\n\n<\/li>\n<\/ol>\n
<\/p>\n
Dla serwera proces sk\u0142ada si\u0119 z 2 faz:<\/p>\n
<\/p>\n
\n- Inicjalizacja punktu wymiany<\/li>\n
- Przetwarzanie przychodz\u0105cych \u017c\u0105da\u0144<\/li>\n<\/ol>\n
<\/p>\n
Ilustrujemy ten szablon kodem. Za\u0142\u00f3\u017cmy, \u017ce musimy zrealizowa\u0107 prost\u0105 us\u0142ug\u0119, kt\u00f3ra udost\u0119pnia jedn\u0105 metod\u0119 zwracaj\u0105c\u0105 dok\u0142adny czas.<\/p>\n
<\/p>\n
Kod serwera<\/h4>\n
<\/p>\n
Wyodr\u0119bnijmy definicj\u0119 API us\u0142ugi do api.hrl:<\/p>\n
<\/p>\n
%% =====================================================\n%% encje\n%% =====================================================\n-record(time, {\n unixtime :: non_neg_integer(),\n datetime :: binary()\n}).\n\n-record(time_error, {\n code :: non_neg_integer(),\n error :: term()\n}).\n\n%% =====================================================\n%% metody\n%% =====================================================\n-record(time_req, {\n opts :: term()\n}).\n-record(time_resp, {\n result :: #time{} | #time_error{}\n}).<\/code><\/pre>\n<\/p>\n
Zdefiniujmy kontroler us\u0142ugi w time_controller.erl<\/p>\n
<\/p>\n
%% W przyk\u0142adzie pokazany jest tylko warto\u015bciowy kod. Wstawiaj\u0105c go do szablonu gen_server, mo\u017cna uzyska\u0107 dzia\u0142aj\u0105c\u0105 us\u0142ug\u0119.\n\n%% inicjalizacja gen_server\ninit(Args) ->\n %% po\u0142\u0105czenie z punktem wymiany\n messaging:monitor_exchange(req_resp, ?EXCHANGE, default, self())\n {ok, #{}}.\n\n%% obs\u0142uga zdarzenia utraty po\u0142\u0105czenia z punktem wymiany. To samo zdarzenie pojawia si\u0119, je\u015bli punkt wymiany jeszcze nie zosta\u0142 uruchomiony.\nhandle_info(#exchange_die{exchange = ?EXCHANGE}, State) ->\n erlang:send(self(), monitor_exchange),\n {noreply, State};\n\n%% obs\u0142uga API\nhandle_info(#time_req{opts = _Opts}, State) ->\n messaging:response_once(Client, #time_resp{\nresult = #time{ unixtime = time_utils:unixtime(now()), datetime = time_utils:iso8601_fmt(now())}\n });\n {noreply, State};\n\n%% zako\u0144czenie pracy gen_server\nterminate(_Reason, _State) ->\n messaging:demonitor_exchange(req_resp, ?EXCHANGE, default, self()),\n ok.<\/code><\/pre>\n<\/p>\n
Kod klienta<\/h4>\n
<\/p>\n
Aby wys\u0142a\u0107 \u017c\u0105danie do us\u0142ugi, w dowolnym miejscu klienta mo\u017cna wywo\u0142a\u0107 interfejs API messaging request:<\/p>\n
<\/p>\n
case messaging:request(?EXCHANGE, tag, #time_req{opts = #{}}, self()) of\n ok -> ok;\n _ -> %% logika powt\u00f3rzenia lub b\u0142\u0119du\nend<\/code><\/pre>\n<\/p>\n
W rozproszonym systemie konfiguracja komponent\u00f3w mo\u017ce by\u0107 bardzo r\u00f3\u017cna, a w momencie \u017c\u0105dania messaging mo\u017ce jeszcze nie by\u0107 uruchomiony, albo kontroler us\u0142ugi nie jest gotowy, aby obs\u0142u\u017cy\u0107 \u017c\u0105danie. Dlatego musimy sprawdzi\u0107 odpowied\u017a messaging i obs\u0142u\u017cy\u0107 przypadek odmowy.
\nPo pomy\u015blnym wys\u0142aniu klient otrzyma odpowied\u017a lub b\u0142\u0105d od us\u0142ugi.
\nObs\u0142u\u017cmy oba przypadki w handle_info:<\/p>\n
<\/p>\n
handle_info(#'$msg'{exchange = ?EXCHANGE, tag = tag, message = #time_resp{result = #time{unixtime = Utime}}}, State) ->\n ?debugVal(Utime),\n {noreply, State};\n\nhandle_info(#'$msg'{exchange = ?EXCHANGE, tag = tag, message = #time_resp{result = #time_error{code = ErrorCode}}}, State) ->\n ?debugVal({error, ErrorCode}),\n {noreply, State};<\/code><\/pre>\n<\/p>\n
\u017b\u0105danie-odpowied\u017a dzielona<\/h3>\n
<\/p>\n
Lepiej unika\u0107 przesy\u0142ania du\u017cych wiadomo\u015bci. Zale\u017cy od tego responsywno\u015b\u0107 i stabilno\u015b\u0107 ca\u0142ego systemu. Je\u015bli odpowied\u017a na zapytanie zajmuje du\u017co pami\u0119ci, podzia\u0142 na cz\u0119\u015bci jest obowi\u0105zkowy.<\/p>\n
<\/p>\n
![\"Bloki](\"\/wp-content\/uploads\/2019\/04\/bd60742a933a8e4e3eeaf7f4adf00f2a.jpg\")
<\/p>\n
<\/p>\n
Podam kilka przyk\u0142ad\u00f3w takich przypadk\u00f3w:<\/p>\n
<\/p>\n
\n- Komponenty wymieniaj\u0105 si\u0119 danymi binarnymi, na przyk\u0142ad plikami. Podzia\u0142 odpowiedzi na ma\u0142e cz\u0119\u015bci umo\u017cliwia efektywne przetwarzanie plik\u00f3w o dowolnym rozmiarze i unikanie przepe\u0142nienia pami\u0119ci.<\/li>\n
- Listy. Na przyk\u0142ad, musimy wybra\u0107 wszystkie zapisy z ogromnej tabeli w bazie danych i przekaza\u0107 je innemu komponentowi.<\/li>\n<\/ul>\n
<\/p>\n
Nazywam takie odpowiedzi lokomotyw\u0105. W ka\u017cdym razie, 1024 wiadomo\u015bci po 1 MB jest lepsze ni\u017c jedna wiadomo\u015b\u0107 o wielko\u015bci 1 GB. <\/p>\n
<\/p>\n
W klastrze Erlang uzyskujemy dodatkowy zysk \u2012 zmniejszenie obci\u0105\u017cenia punktu wymiany i sieci, poniewa\u017c odpowiedzi s\u0105 od razu kierowane do odbiorcy, omijaj\u0105c punkt wymiany.<\/p>\n
<\/p>\n
Odpowied\u017a z \u017c\u0105daniem<\/h3>\n
<\/p>\n
Jest to do\u015b\u0107 rzadkie modyfikowanie wzorca RPC do budowania system\u00f3w dialogowych.<\/p>\n
<\/p>\n
![\"Bloki](\"\/wp-content\/uploads\/2019\/04\/9b350a0b48ca8acdca973da2f7eadb76.jpg\")
<\/p>\n
<\/p>\n
Publish-subscribe (drzewo dystrybucji danych)<\/h3>\n
<\/p>\n
Systemy zorientowane na zdarzenia dostarczaj\u0105 dane konsumentom w miar\u0119 ich gotowo\u015bci. W ten spos\u00f3b systemy s\u0105 bardziej sk\u0142onne do modelu push ni\u017c pull czy poll. Ta cecha pozwala nie marnowa\u0107 zasob\u00f3w na ci\u0105g\u0142e pytanie i oczekiwanie na dane.
\nNa ilustracji przedstawiono proces rozprzestrzeniania wiadomo\u015bci do konsument\u00f3w, kt\u00f3rzy s\u0105 subskrybowani na dany temat.<\/p>\n
<\/p>\n
![\"Bloki](\"\/wp-content\/uploads\/2019\/04\/e211d5dcfeb5ce8513699eac8e3b6d84.jpg\")
<\/p>\n
<\/p>\n
Klasycznymi przyk\u0142adami u\u017cycia tego wzorca jest dystrybucja stanu: \u015bwiata gry w grach komputerowych, danych rynkowych na gie\u0142dach, u\u017cytecznych informacji w zasilaczach danych.<\/p>\n
<\/p>\n
Rozwa\u017cmy kod subskrybenta:<\/p>\n
<\/p>\n
init(_Args) ->\n %% subskrybujemy si\u0119 na punkt wymiany, klucz = key\n messaging:subscribe(?SUBSCRIPTION, key, tag, self()),\n {ok, #{}}.\n\nhandle_info(#exchange_die{exchange = ?SUBSCRIPTION}, State) ->\n %% je\u015bli punkt wymiany jest niedost\u0119pny, pr\u00f3bujemy ponownie si\u0119 po\u0142\u0105czy\u0107\n messaging:subscribe(?SUBSCRIPTION, key, tag, self()),\n {noreply, State};\n\n%% przetwarzamy nadchodz\u0105ce wiadomo\u015bci\nhandle_info(#'$msg'{exchange = ?SUBSCRIPTION, message = Msg}, State) ->\n ?debugVal(Msg),\n {noreply, State};\n\n%% przy zatrzymaniu konsumenta - roz\u0142\u0105czamy si\u0119 od punktu wymiany\nterminate(_Reason, _State) ->\n messaging:unsubscribe(?SUBSCRIPTION, key, tag, self()),\n ok.<\/code><\/pre>\n<\/p>\n
\u0179r\u00f3d\u0142o mo\u017ce wywo\u0142a\u0107 funkcj\u0119 publikacji wiadomo\u015bci w dowolnym miejscu:<\/p>\n
<\/p>\n
messaging:publish_message(Exchange, Key, Message).<\/code><\/pre>\n<\/p>\n
Exchange<\/em> \u2012 nazwa punktu wymiany,
\nKlucz<\/em> \u2012 klucz routingu
\nWiadomo\u015b\u0107<\/em> \u2012 \u0142adunek u\u017cyteczny<\/p>\n<\/p>\n
Odwr\u00f3cone Pub-sub<\/h2>\n
<\/p>\n
![\"Bloki](\"\/wp-content\/uploads\/2019\/04\/2ef7cd7de459455937bdcfa8a11fd028.jpg\")
<\/p>\n
<\/p>\n
Rozwijaj\u0105c pub-sub, mo\u017cna uzyska\u0107 wz\u00f3r odpowiedni do logowania. Zestaw \u017ar\u00f3de\u0142 i konsument\u00f3w mo\u017ce by\u0107 zupe\u0142nie r\u00f3\u017cny. Na rysunku przedstawiono przypadek z jednym konsumentem i wieloma \u017ar\u00f3d\u0142ami.<\/p>\n
<\/p>\n
Wz\u00f3r dystrybucji zada\u0144<\/h2>\n
<\/p>\n
W prawie ka\u017cdym projekcie wyst\u0119puj\u0105 zadania o odroczonym przetwarzaniu, takie jak generowanie raport\u00f3w, dostarczanie powiadomie\u0144, uzyskiwanie danych z zewn\u0119trznych system\u00f3w. Przepustowo\u015b\u0107 systemu wykonuj\u0105cego te zadania \u0142atwo skalowa\u0107 poprzez dodanie przetw\u00f3rc\u00f3w. Wszystko, co nam pozostaje, to utworzy\u0107 klaster przetw\u00f3rc\u00f3w i r\u00f3wnomiernie rozdziela\u0107 zadania mi\u0119dzy nimi.<\/p>\n
<\/p>\n
Rozwa\u017cmy sytuacje, kt\u00f3re mog\u0105 wyst\u0105pi\u0107 na przyk\u0142adzie trzech przetw\u00f3rc\u00f3w. Ju\u017c na etapie rozdzielania zada\u0144 pojawia si\u0119 kwestia sprawiedliwo\u015bci dystrybucji oraz przepe\u0142nienia przetw\u00f3rc\u00f3w. Za sprawiedliwo\u015b\u0107 odpowiada dystrybucja round-robin, a aby unikn\u0105\u0107 sytuacji przepe\u0142nienia przetw\u00f3rc\u00f3w, wprowadzimy ograniczenie prefetch_limit<\/em>. W trybach przej\u015bciowych prefetch_limit<\/em> nie pozwoli jednemu przetw\u00f3rcy na otrzymanie wszystkich zada\u0144.<\/p>\n<\/p>\n
Messaging zarz\u0105dza kolejkami i priorytetami przetwarzania. Przetw\u00f3rcy otrzymuj\u0105 zadania w miar\u0119 ich przybywania. Wykonanie zadania mo\u017ce zako\u0144czy\u0107 si\u0119 sukcesem lub niepowodzeniem:<\/p>\n
<\/p>\n
\nmessaging:ack(Tack)<\/code> \u2012 wywo\u0142ywane w przypadku pomy\u015blnego przetworzenia wiadomo\u015bci<\/li>\nmessaging:nack(Tack)<\/code> \u2012 wywo\u0142ywane w nieprzewidzianych sytuacjach. Po zwr\u00f3ceniu zadania messaging przeka\u017ce je innemu przetw\u00f3rcy.<\/li>\n<\/ul>\n<\/p>\n
![\"Bloki](\"\/wp-content\/uploads\/2019\/04\/c2c632b6411247ee8b0dcfb3250a4503.jpg\")
<\/p>\n
<\/p>\n
Przypu\u015b\u0107my, \u017ce podczas przetwarzania trzech zada\u0144 wyst\u0105pi\u0142a powa\u017cna awaria: przetw\u00f3rca 1 po otrzymaniu zadania upad\u0142, nie zd\u0105\u017cywszy przekaza\u0107 nic punktowi wymiany. W takim przypadku punkt wymiany, po up\u0142ywie czasu na ack, przeka\u017ce zadanie innemu przetw\u00f3rcy. Przetw\u00f3rca 3 z jakiego\u015b powodu odm\u00f3wi\u0142 przyj\u0119cia zadania i wys\u0142a\u0142 nack, w efekcie zadanie r\u00f3wnie\u017c trafi\u0142o do innego przetw\u00f3rcy, kt\u00f3ry je pomy\u015blnie wykona\u0142.<\/p>\n
<\/p>\n
Wst\u0119pny podsumowanie<\/h2>\n
<\/p>\n
Om\u00f3wili\u015bmy podstawowe elementy system\u00f3w rozproszonych i uzyskali\u015bmy podstawowe zrozumienie ich zastosowania w Erlang'u\/Elixirze.<\/p>\n
<\/p>\n
\u0141\u0105cz\u0105c podstawowe wzory, mo\u017cna budowa\u0107 z\u0142o\u017cone paradygmaty do rozwi\u0105zywania pojawiaj\u0105cych si\u0119 zada\u0144.<\/p>\n
<\/p>\n
W ostatniej cz\u0119\u015bci cyklu om\u00f3wimy og\u00f3lne zagadnienia dotycz\u0105ce organizacji us\u0142ug, routingu i balansowania, a tak\u017ce poruszymy praktyczne aspekty skalowalno\u015bci i odporno\u015bci system\u00f3w na awarie.<\/p>\n
<\/p>\n
Koniec drugiej cz\u0119\u015bci.<\/p>\n
<\/p>\n
Zdj\u0119cie
![\"Bloki](\"\/wp-content\/uploads\/2019\/04\/7ef485e452075775ce317b557671274c.jpg\")
<\/p>\n