🥇Δομικά στοιχεία κατανεμημένων εφαρμογών. Πρώτη προσέγγιση

Στο παρελθόν άρθρο Εξετάσαμε τα θεωρητικά θεμέλια της αντιδραστικής αρχιτεκτονικής. Ήρθε η ώρα να μιλήσουμε για ροές δεδομένων, τρόπους εφαρμογής αντιδραστικών συστημάτων Erlang/Elixir και μοτίβα μηνυμάτων σε αυτά:

Απαιτώ απάντηση
Αίτημα-Συγκεντρωμένη απάντηση
Απάντηση με αίτημα
Δημοσίευση-εγγραφή
Αντεστραμμένη Δημοσίευση-εγγραφή
Κατανομή εργασιών

SOA, MSA και Μηνύματα

Τα SOA, MSA είναι αρχιτεκτονικές συστημάτων που καθορίζουν τους κανόνες για τα κτιριακά συστήματα, ενώ η ανταλλαγή μηνυμάτων παρέχει πρωταρχικά στοιχεία για την υλοποίησή τους.

Δεν θέλω να προωθήσω αυτήν ή εκείνη την αρχιτεκτονική συστήματος. Είμαι υπέρ της χρήσης των πιο αποτελεσματικών και χρήσιμων πρακτικών για ένα συγκεκριμένο έργο και επιχείρηση. Όποιο παράδειγμα και αν επιλέξουμε, είναι καλύτερο να δημιουργήσουμε μπλοκ συστήματος με το βλέμμα στο Unix-way: στοιχεία με ελάχιστη συνδεσιμότητα, υπεύθυνα για μεμονωμένες οντότητες. Οι μέθοδοι API εκτελούν τις απλούστερες δυνατές ενέργειες με οντότητες.

Το Messaging είναι, όπως υποδηλώνει το όνομα, ένας μεσίτης μηνυμάτων. Ο κύριος σκοπός του είναι να λαμβάνει και να στέλνει μηνύματα. Είναι υπεύθυνο για τις διεπαφές για την αποστολή πληροφοριών, το σχηματισμό λογικών καναλιών για τη μετάδοση πληροφοριών εντός του συστήματος, τη δρομολόγηση και την εξισορρόπηση, καθώς και τη διαχείριση σφαλμάτων σε επίπεδο συστήματος.
Τα μηνύματα που αναπτύσσουμε δεν προσπαθούν να ανταγωνιστούν ή να αντικαταστήσουν το rabbitmq. Τα κύρια χαρακτηριστικά του:

Διανομή.
Τα σημεία ανταλλαγής μπορούν να δημιουργηθούν σε όλους τους κόμβους συμπλέγματος, όσο το δυνατόν πιο κοντά στον κώδικα που τα χρησιμοποιεί.
Απλότητα.
Επικεντρωθείτε στην ελαχιστοποίηση του κώδικα λέβητα και στην ευκολία χρήσης.
Καλύτερη απόδοση.
Δεν προσπαθούμε να επαναλάβουμε τη λειτουργικότητα του rabbitmq, αλλά τονίζουμε μόνο το αρχιτεκτονικό και το στρώμα μεταφοράς, το οποίο εντάσσουμε στο OTP όσο το δυνατόν πιο απλά, ελαχιστοποιώντας το κόστος.
Ευκαμψία.
Κάθε υπηρεσία μπορεί να συνδυάσει πολλά πρότυπα ανταλλαγής.
Ανθεκτικότητα από το σχεδιασμό.
Επεκτασιμότητα
Τα μηνύματα μεγαλώνουν με την εφαρμογή. Καθώς το φορτίο αυξάνεται, μπορείτε να μετακινήσετε τα σημεία ανταλλαγής σε μεμονωμένα μηχανήματα.

Σχόλιο. Όσον αφορά την οργάνωση κώδικα, τα μετα-έργα είναι κατάλληλα για πολύπλοκα συστήματα Erlang/Elixir. Όλος ο κώδικας του έργου βρίσκεται σε ένα αποθετήριο - ένα έργο ομπρέλα. Ταυτόχρονα, οι μικροϋπηρεσίες απομονώνονται στο μέγιστο και εκτελούν απλές λειτουργίες που είναι υπεύθυνες για μια ξεχωριστή οντότητα. Με αυτήν την προσέγγιση, είναι εύκολο να διατηρηθεί το API ολόκληρου του συστήματος, είναι εύκολο να κάνετε αλλαγές, είναι βολικό να γράφετε δοκιμές μονάδας και ολοκλήρωσης.

Τα στοιχεία του συστήματος αλληλεπιδρούν άμεσα ή μέσω ενός μεσίτη. Από την άποψη της ανταλλαγής μηνυμάτων, κάθε υπηρεσία έχει διάφορες φάσεις ζωής:

Αρχικοποίηση υπηρεσίας.
Σε αυτό το στάδιο, η διαδικασία εκτέλεσης της υπηρεσίας και οι εξαρτήσεις της διαμορφώνονται και εκκινούνται.
Δημιουργία σημείου ανταλλαγής.
Η υπηρεσία μπορεί να χρησιμοποιήσει ένα στατικό σημείο ανταλλαγής που καθορίζεται στη διαμόρφωση του κόμβου ή να δημιουργήσει σημεία ανταλλαγής δυναμικά.
Εγγραφή υπηρεσίας.
Για να εξυπηρετεί η υπηρεσία αιτήματα θα πρέπει να έχει καταχωρηθεί στο σημείο ανταλλαγής.
Κανονική λειτουργία.
Η υπηρεσία παράγει χρήσιμο έργο.
ΤΕΡΜΑΤΙΣΜΟΣ ΛΕΙΤΟΥΡΓΙΑΣ.
Υπάρχουν 2 πιθανοί τύποι διακοπής λειτουργίας: κανονικός και έκτακτος. Κατά την κανονική λειτουργία, η υπηρεσία αποσυνδέεται από το σημείο ανταλλαγής και σταματά. Σε καταστάσεις έκτακτης ανάγκης, η ανταλλαγή μηνυμάτων εκτελεί ένα από τα σενάρια ανακατεύθυνσης.

Φαίνεται αρκετά περίπλοκο, αλλά ο κώδικας δεν είναι τόσο τρομακτικός. Παραδείγματα κώδικα με σχόλια θα δοθούν στην ανάλυση των προτύπων λίγο αργότερα.

Χρηματιστήρια

Το σημείο ανταλλαγής είναι μια διαδικασία ανταλλαγής μηνυμάτων που εφαρμόζει τη λογική της αλληλεπίδρασης με στοιχεία εντός του προτύπου ανταλλαγής μηνυμάτων. Σε όλα τα παραδείγματα που παρουσιάζονται παρακάτω, τα στοιχεία αλληλεπιδρούν μέσω σημείων ανταλλαγής, ο συνδυασμός των οποίων σχηματίζει μηνύματα.

Μοτίβα ανταλλαγής μηνυμάτων (ΜΕΚ)

Σε παγκόσμιο επίπεδο, τα πρότυπα ανταλλαγής μπορούν να χωριστούν σε αμφίδρομα και μονόδρομα. Τα πρώτα υποδηλώνουν απάντηση σε ένα εισερχόμενο μήνυμα, τα δεύτερα όχι. Ένα κλασικό παράδειγμα αμφίδρομου μοτίβου στην αρχιτεκτονική πελάτη-διακομιστή είναι το μοτίβο Αίτησης-απόκρισης. Ας δούμε το πρότυπο και τις τροποποιήσεις του.

Αίτημα-απάντηση ή RPC

Το RPC χρησιμοποιείται όταν χρειάζεται να λάβουμε απάντηση από άλλη διεργασία. Αυτή η διαδικασία μπορεί να εκτελείται στον ίδιο κόμβο ή να βρίσκεται σε διαφορετική ήπειρο. Ακολουθεί ένα διάγραμμα της αλληλεπίδρασης μεταξύ πελάτη και διακομιστή μέσω μηνυμάτων.

Δεδομένου ότι η ανταλλαγή μηνυμάτων είναι εντελώς ασύγχρονη, για τον πελάτη η ανταλλαγή χωρίζεται σε 2 φάσεις:

Αποστολή αιτήματος
```
messaging:request(Exchange, ResponseMatchingTag, RequestDefinition, HandlerProcess).
```
ανταλλαγή ‒ μοναδικό όνομα του σημείου ανταλλαγής
ResponseMatchingTag ‒ τοπική ετικέτα για την επεξεργασία της απάντησης. Για παράδειγμα, στην περίπτωση αποστολής πολλών πανομοιότυπων αιτημάτων που ανήκουν σε διαφορετικούς χρήστες.
Προσδιορισμός αιτήματος - όργανο αιτήματος
HandlerProcess ‒ PID του χειριστή. Αυτή η διαδικασία θα λάβει απάντηση από τον διακομιστή.
Επεξεργασία της απάντησης
```
handle_info(#'$msg'{exchange = EXCHANGE, tag = ResponseMatchingTag,message = ResponsePayload}, State)
```
ResponsePayload - απόκριση διακομιστή.

Για τον διακομιστή, η διαδικασία αποτελείται επίσης από 2 φάσεις:

Αρχικοποίηση του σημείου ανταλλαγής
Επεξεργασία ληφθέντων αιτημάτων

Ας απεικονίσουμε αυτό το πρότυπο με κώδικα. Ας υποθέσουμε ότι πρέπει να εφαρμόσουμε μια απλή υπηρεσία που παρέχει μια ενιαία μέθοδο ακριβούς χρόνου.

Κωδικός διακομιστή

Ας ορίσουμε το API υπηρεσίας στο api.hrl:

%% =====================================================
%%  entities
%% =====================================================
-record(time, {
  unixtime :: non_neg_integer(),
  datetime :: binary()
}).

-record(time_error, {
  code :: non_neg_integer(),
  error :: term()
}).

%% =====================================================
%%  methods
%% =====================================================
-record(time_req, {
  opts :: term()
}).
-record(time_resp, {
  result :: #time{} | #time_error{}
}).

Ας ορίσουμε τον ελεγκτή υπηρεσίας στο time_controller.erl

%% В примере показан только значимый код. Вставив его в шаблон gen_server можно получить рабочий сервис.

%% инициализация gen_server
init(Args) ->
  %% подключение к точке обмена
  messaging:monitor_exchange(req_resp, ?EXCHANGE, default, self())
  {ok, #{}}.

%% обработка события потери связи с точкой обмена. Это же событие приходит, если точка обмена еще не запустилась.
handle_info(#exchange_die{exchange = ?EXCHANGE}, State) ->
  erlang:send(self(), monitor_exchange),
  {noreply, State};

%% обработка API
handle_info(#time_req{opts = _Opts}, State) ->
  messaging:response_once(Client, #time_resp{
result = #time{ unixtime = time_utils:unixtime(now()), datetime = time_utils:iso8601_fmt(now())}
  });
  {noreply, State};

%% завершение работы gen_server
terminate(_Reason, _State) ->
  messaging:demonitor_exchange(req_resp, ?EXCHANGE, default, self()),
  ok.

Κωδικός πελάτη

Για να στείλετε ένα αίτημα στην υπηρεσία, μπορείτε να καλέσετε το API αιτήματος ανταλλαγής μηνυμάτων οπουδήποτε στον πελάτη:

case messaging:request(?EXCHANGE, tag, #time_req{opts = #{}}, self()) of
    ok -> ok;
    _ -> %% repeat or fail logic
end

Σε ένα κατανεμημένο σύστημα, η διαμόρφωση των στοιχείων μπορεί να είναι πολύ διαφορετική και τη στιγμή της αίτησης, τα μηνύματα ενδέχεται να μην ξεκινούν ακόμη ή ο ελεγκτής υπηρεσίας δεν θα είναι έτοιμος να εξυπηρετήσει το αίτημα. Επομένως, πρέπει να ελέγξουμε την απόκριση των μηνυμάτων και να χειριστούμε την περίπτωση αποτυχίας.
Μετά την επιτυχή αποστολή, ο πελάτης θα λάβει απάντηση ή σφάλμα από την υπηρεσία.
Ας χειριστούμε και τις δύο περιπτώσεις στο handle_info:

handle_info(#'$msg'{exchange = ?EXCHANGE, tag = tag, message = #time_resp{result = #time{unixtime = Utime}}}, State) ->
  ?debugVal(Utime),
  {noreply, State};

handle_info(#'$msg'{exchange = ?EXCHANGE, tag = tag, message = #time_resp{result = #time_error{code = ErrorCode}}}, State) ->
  ?debugVal({error, ErrorCode}),
  {noreply, State};

Αίτημα-Συγκεντρωμένη απάντηση

Είναι καλύτερο να αποφεύγετε να στέλνετε τεράστια μηνύματα. Η απόκριση και η σταθερή λειτουργία ολόκληρου του συστήματος εξαρτάται από αυτό. Εάν η απάντηση σε ένα ερώτημα καταλαμβάνει πολλή μνήμη, τότε ο χωρισμός του σε μέρη είναι υποχρεωτικός.

Επιτρέψτε μου να σας δώσω μερικά παραδείγματα τέτοιων περιπτώσεων:

Τα στοιχεία ανταλλάσσουν δυαδικά δεδομένα, όπως αρχεία. Το σπάσιμο της απόκρισης σε μικρά μέρη σάς βοηθά να εργάζεστε αποτελεσματικά με αρχεία οποιουδήποτε μεγέθους και να αποφεύγετε τις υπερχειλίσεις μνήμης.
Καταχωρίσεις. Για παράδειγμα, πρέπει να επιλέξουμε όλες τις εγγραφές από έναν τεράστιο πίνακα στη βάση δεδομένων και να τις μεταφέρουμε σε άλλο στοιχείο.

Ονομάζω αυτές τις απαντήσεις ατμομηχανή. Σε κάθε περίπτωση, 1024 μηνύματα του 1 MB είναι καλύτερα από ένα μήνυμα του 1 GB.

Στο σύμπλεγμα Erlang, έχουμε ένα πρόσθετο όφελος - μείωση του φόρτου στο σημείο ανταλλαγής και στο δίκτυο, καθώς οι απαντήσεις αποστέλλονται αμέσως στον παραλήπτη, παρακάμπτοντας το σημείο ανταλλαγής.

Απάντηση με αίτημα

Αυτή είναι μια μάλλον σπάνια τροποποίηση του προτύπου RPC για την κατασκευή συστημάτων διαλόγου.

Τα συστήματα που βασίζονται σε συμβάντα τα παραδίδουν στους καταναλωτές μόλις τα δεδομένα είναι έτοιμα. Έτσι, τα συστήματα είναι πιο επιρρεπή σε ένα μοντέλο ώθησης παρά σε ένα μοντέλο έλξης ή δημοσκόπησης. Αυτή η δυνατότητα σάς επιτρέπει να αποφύγετε τη σπατάλη πόρων ζητώντας συνεχώς και αναμένοντας δεδομένα.
Το σχήμα δείχνει τη διαδικασία διανομής ενός μηνύματος σε καταναλωτές που έχουν εγγραφεί σε ένα συγκεκριμένο θέμα.

Κλασικά παραδείγματα χρήσης αυτού του προτύπου είναι η κατανομή της κατάστασης: ο κόσμος των παιχνιδιών στα παιχνίδια υπολογιστών, τα δεδομένα της αγοράς για τις ανταλλαγές, οι χρήσιμες πληροφορίες στις ροές δεδομένων.

Ας δούμε τον κωδικό συνδρομητή:

init(_Args) ->
  %% подписываемся на обменник, ключ = key
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {ok, #{}}.

handle_info(#exchange_die{exchange = ?SUBSCRIPTION}, State) ->
  %% если точка обмена недоступна, то пытаемся переподключиться
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {noreply, State};

%% обрабатываем пришедшие сообщения
handle_info(#'$msg'{exchange = ?SUBSCRIPTION, message = Msg}, State) ->
  ?debugVal(Msg),
  {noreply, State};

%% при остановке потребителя - отключаемся от точки обмена
terminate(_Reason, _State) ->
  messaging:unsubscribe(?SUBSCRIPTION, key, tag, self()),
  ok.

Η πηγή μπορεί να καλέσει τη συνάρτηση για να δημοσιεύσει ένα μήνυμα σε οποιοδήποτε βολικό μέρος:

messaging:publish_message(Exchange, Key, Message).

ανταλλαγή - όνομα του σημείου ανταλλαγής,
Κλειδί - κλειδί δρομολόγησης
Μήνυμα - ωφέλιμο φορτίο

Επεκτείνοντας το pub-sub, μπορείτε να αποκτήσετε ένα μοτίβο βολικό για την καταγραφή. Το σύνολο των πηγών και των καταναλωτών μπορεί να είναι εντελώς διαφορετικό. Το σχήμα δείχνει μια περίπτωση με έναν καταναλωτή και πολλές πηγές.

Μοτίβο κατανομής εργασιών

Σχεδόν κάθε έργο περιλαμβάνει εργασίες επεξεργασίας με αναβολή, όπως η δημιουργία αναφορών, η παράδοση ειδοποιήσεων και η ανάκτηση δεδομένων από συστήματα τρίτων. Η απόδοση του συστήματος που εκτελεί αυτές τις εργασίες μπορεί εύκολα να κλιμακωθεί με την προσθήκη χειριστών. Το μόνο που μας μένει είναι να σχηματίσουμε ένα σύμπλεγμα επεξεργαστών και να κατανείμουμε ομοιόμορφα τις εργασίες μεταξύ τους.

Ας δούμε τις καταστάσεις που προκύπτουν χρησιμοποιώντας το παράδειγμα 3 χειριστών. Ακόμη και στο στάδιο της κατανομής εργασιών, τίθεται το ζήτημα της δίκαιης κατανομής και της υπερχείλισης των χειριστών. Η διανομή κυκλικής ροής θα είναι υπεύθυνη για τη δικαιοσύνη και για να αποφευχθεί μια κατάσταση υπερχείλισης των χειριστών, θα εισαγάγουμε έναν περιορισμό prefetch_limit. Σε παροδικές συνθήκες prefetch_limit θα εμποδίσει έναν χειριστή να λάβει όλες τις εργασίες.

Η ανταλλαγή μηνυμάτων διαχειρίζεται τις ουρές και την προτεραιότητα επεξεργασίας. Οι επεξεργαστές λαμβάνουν εργασίες καθώς φτάνουν. Η εργασία μπορεί να ολοκληρωθεί με επιτυχία ή να αποτύχει:

messaging:ack(Tack) - καλείται εάν το μήνυμα επεξεργαστεί με επιτυχία
messaging:nack(Tack) - καλείται σε όλες τις καταστάσεις έκτακτης ανάγκης. Μόλις επιστραφεί η εργασία, τα μηνύματα θα τη διαβιβάσουν σε άλλον χειριστή.

Ας υποθέσουμε ότι συνέβη μια περίπλοκη αποτυχία κατά την επεξεργασία τριών εργασιών: ο επεξεργαστής 1, μετά την παραλαβή της εργασίας, κατέρρευσε χωρίς να έχει χρόνο να αναφέρει οτιδήποτε στο σημείο ανταλλαγής. Σε αυτήν την περίπτωση, το σημείο ανταλλαγής θα μεταφέρει την εργασία σε άλλο χειριστή αφού λήξει το χρονικό όριο αναγνώρισης. Για κάποιο λόγο, ο χειριστής 3 εγκατέλειψε την εργασία και έστειλε nack· ως αποτέλεσμα, η εργασία μεταφέρθηκε επίσης σε άλλον χειριστή που την ολοκλήρωσε με επιτυχία.

Προκαταρκτική περίληψη

Καλύψαμε τα βασικά δομικά στοιχεία των κατανεμημένων συστημάτων και αποκτήσαμε μια βασική κατανόηση της χρήσης τους στο Erlang/Elixir.

Συνδυάζοντας βασικά μοτίβα, μπορείτε να δημιουργήσετε σύνθετα παραδείγματα για την επίλυση αναδυόμενων προβλημάτων.

Στο τελευταίο μέρος της σειράς, θα εξετάσουμε γενικά θέματα οργάνωσης υπηρεσιών, δρομολόγησης και εξισορρόπησης, και επίσης θα μιλήσουμε για την πρακτική πλευρά της επεκτασιμότητας και της ανοχής σφαλμάτων των συστημάτων.

Τέλος δεύτερου μέρους.

Φωτογραφία Μάριους Κρίστενσεν
Εικονογραφήσεις που προετοιμάστηκαν χρησιμοποιώντας το websequencediagrams.com

Πηγή: www.habr.com

Δομικά στοιχεία κατανεμημένων εφαρμογών. Πρώτη προσέγγιση