🥇Dağıtılmış uygulamaların yapı taşları. İlk yaklaşım

Geçmişte Makale Reaktif mimarinin teorik temellerini inceledik. Veri akışları, reaktif Erlang/Elixir sistemlerini uygulama yolları ve bunlardaki mesajlaşma kalıpları hakkında konuşmanın zamanı geldi:

istek-yanıt
İstek Parçalı Yanıt
Taleple Yanıt
Yayınla-abone ol
Tersine çevrilmiş Yayınla-abone ol
Görev dağıtımı

SOA, MSA ve Mesajlaşma

SOA, MSA, sistemlerin oluşturulmasına ilişkin kuralları tanımlayan sistem mimarileridir; mesajlaşma ise bunların uygulanması için temelleri sağlar.

Şu veya bu sistem mimarisini tanıtmak istemiyorum. Belirli bir proje ve iş için en etkili ve faydalı uygulamaların kullanılmasından yanayım. Hangi paradigmayı seçersek seçelim, Unix yolunu göz önünde bulundurarak sistem blokları oluşturmak daha iyidir: minimum bağlantıya sahip, bireysel varlıklardan sorumlu bileşenler. API yöntemleri varlıklarla mümkün olan en basit eylemleri gerçekleştirir.

Messaging, adından da anlaşılacağı gibi bir mesaj aracısıdır. Temel amacı mesaj almak ve göndermektir. Bilgi gönderme arayüzlerinden, sistem içinde bilgi aktarımı için mantıksal kanalların oluşturulmasından, yönlendirme ve dengelemenin yanı sıra sistem düzeyinde arıza yönetiminden sorumludur.
Geliştirdiğimiz mesajlaşma, Rabbitmq ile rekabet etmeye veya onun yerini almaya çalışmıyor. Ana özellikleri:

Dağıtım.
Değişim noktaları tüm küme düğümlerinde, onları kullanan koda mümkün olduğunca yakın oluşturulabilir.
Sadelik.
Standart kodu en aza indirmeye ve kullanım kolaylığına odaklanın.
Daha iyi performans.
Rabbitmq'in işlevselliğini tekrarlamaya çalışmıyoruz, yalnızca OTP'ye mümkün olduğunca basit bir şekilde sığdırdığımız mimari ve taşıma katmanını vurgulayarak maliyetleri en aza indiriyoruz.
Esneklik.
Her hizmet birçok değişim şablonunu birleştirebilir.
Tasarım gereği dayanıklılık.
Ölçeklenebilirlik.
Uygulamayla birlikte mesajlaşma da büyüyor. Yük arttıkça değişim noktalarını ayrı ayrı makinelere taşıyabilirsiniz.

Not. Kod organizasyonu açısından meta projeler karmaşık Erlang/Elixir sistemleri için çok uygundur. Tüm proje kodları tek bir havuzda, yani bir şemsiye projede bulunur. Aynı zamanda, mikro hizmetler maksimum düzeyde yalıtılmıştır ve ayrı bir varlıktan sorumlu olan basit işlemleri gerçekleştirir. Bu yaklaşımla tüm sistemin API'sini korumak kolaydır, değişiklik yapmak kolaydır, birim ve entegrasyon testleri yazmak uygundur.

Sistem bileşenleri doğrudan veya bir komisyoncu aracılığıyla etkileşime girer. Mesajlaşma açısından bakıldığında, her hizmetin birkaç yaşam aşaması vardır:

Hizmet başlatma.
Bu aşamada hizmeti yürüten süreç ve bağımlılıkları yapılandırılır ve başlatılır.
Bir değişim noktası oluşturmak.
Hizmet, düğüm yapılandırmasında belirtilen statik bir değişim noktasını kullanabilir veya dinamik olarak değişim noktaları oluşturabilir.
Hizmet kaydı.
Hizmetin isteklere hizmet verebilmesi için değişim noktasında kayıtlı olması gerekir.
Normal işleyiş.
Hizmet yararlı işler üretir.
İş bitimi.
2 tür kapatma mümkündür: normal ve acil durum. Normal çalışma sırasında servisin değişim noktasından bağlantısı kesilir ve durur. Acil durumlarda mesajlaşma, yük devretme komut dosyalarından birini çalıştırır.

Oldukça karmaşık görünüyor, ancak kod o kadar da korkutucu değil. Yorumlu kod örnekleri biraz sonra şablonların analizinde verilecektir.

Değişimleri

Değişim noktası, mesajlaşma şablonu içindeki bileşenlerle etkileşim mantığını uygulayan bir mesajlaşma işlemidir. Aşağıda sunulan tüm örneklerde, bileşenler değişim noktaları aracılığıyla etkileşime girer ve bunların birleşimi mesajlaşmayı oluşturur.

Mesaj alışveriş kalıpları (MEP'ler)

Küresel olarak değişim kalıpları iki yönlü ve tek yönlü olarak ikiye ayrılabilir. İlki gelen bir mesaja yanıt verilmesini ima eder, ikincisi ise etmez. İstemci-sunucu mimarisindeki iki yönlü modelin klasik bir örneği, İstek-yanıt modelidir. Şimdi şablona ve değişikliklerine bakalım.

İstek-yanıt veya RPC

RPC, başka bir süreçten yanıt almamız gerektiğinde kullanılır. Bu süreç aynı düğümde veya farklı bir kıtada çalışıyor olabilir. Aşağıda istemci ve sunucu arasındaki mesajlaşma yoluyla etkileşimin bir diyagramı bulunmaktadır.

Mesajlaşma tamamen eşzamansız olduğundan, müşteri için değişim 2 aşamaya bölünmüştür:

istek gönderiliyor
```
messaging:request(Exchange, ResponseMatchingTag, RequestDefinition, HandlerProcess).
```
Başka ürün ile değiştirme – değişim noktasının benzersiz adı
ResponseMatchingTag – yanıtı işlemek için yerel etiket. Örneğin, farklı kullanıcılara ait birden fazla aynı isteğin gönderilmesi durumunda.
İstek Tanımı - istek organı
İşleyici Süreci ‒ İşleyicinin PID'si. Bu işlem sunucudan bir yanıt alacaktır.

Yanıt işleniyor

handle_info(#'$msg'{exchange = EXCHANGE, tag = ResponseMatchingTag,message = ResponsePayload}, State)

YanıtYükü - Sunucu cevabı.

Sunucu için süreç ayrıca 2 aşamadan oluşur:

Değişim noktasının başlatılması
Alınan isteklerin işlenmesi

Bu şablonu kodla örnekleyelim. Diyelim ki tek bir kesin zaman yöntemi sağlayan basit bir hizmeti uygulamamız gerekiyor.

sunucu kodu

Hizmet API'sini api.hrl'de tanımlayalım:

%% =====================================================
%%  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 dosyasında servis denetleyicisini tanımlayalım

%% В примере показан только значимый код. Вставив его в шаблон 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.

Müşteri kodu

Hizmete bir istek göndermek için istemcinin herhangi bir yerinde mesajlaşma isteği API'sini arayabilirsiniz:

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

Dağıtılmış bir sistemde bileşenlerin konfigürasyonu çok farklı olabilir ve istek anında mesajlaşma henüz başlamayabilir veya hizmet denetleyicisi isteğe hizmet vermeye hazır olmayabilir. Bu nedenle mesajlaşma yanıtını kontrol etmemiz ve arıza durumunu ele almamız gerekiyor.
Başarılı gönderimden sonra istemci, hizmetten bir yanıt veya hata alacaktır.
Her iki durumu da tanıtıcı_info'da ele alalım:

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};

İstek Parçalı Yanıt

Çok büyük mesajlar göndermekten kaçınmak en iyisidir. Tüm sistemin yanıt verebilirliği ve kararlı çalışması buna bağlıdır. Bir sorguya verilen yanıt çok fazla bellek kaplıyorsa, onu parçalara bölmek zorunludur.

Bu tür durumlara birkaç örnek vereyim:

Bileşenler, dosyalar gibi ikili verileri değiştirir. Yanıtı küçük parçalara bölmek, her boyuttaki dosyayla verimli bir şekilde çalışmanıza ve bellek taşmalarını önlemenize yardımcı olur.
Listeler. Örneğin veritabanındaki devasa bir tablodaki tüm kayıtları seçip başka bir bileşene aktarmamız gerekiyor.

Ben bu tepkilere lokomotif diyorum. Her durumda, 1024 MB'lık 1 mesaj, 1 GB'lık tek bir mesajdan daha iyidir.

Erlang kümesinde, ek bir fayda elde ederiz - yanıtlar, değişim noktasını atlayarak anında alıcıya gönderildiğinden, değişim noktası ve ağ üzerindeki yükü azaltır.

Taleple Yanıt

Bu, diyalog sistemleri oluşturmak için RPC modelinin oldukça nadir görülen bir modifikasyonudur.

Olay odaklı sistemler, veriler hazır olur olmaz bunları tüketicilere ulaştırır. Bu nedenle sistemler, çekme veya yoklama modelinden ziyade itme modeline daha yatkındır. Bu özellik, sürekli olarak veri isteyip bekleyerek kaynak israfını önlemenizi sağlar.
Şekil, belirli bir konuya abone olan tüketicilere bir mesajın dağıtılma sürecini göstermektedir.

Bu modeli kullanmanın klasik örnekleri, durum dağıtımıdır: bilgisayar oyunlarındaki oyun dünyası, borsalardaki pazar verileri, veri akışlarındaki faydalı bilgiler.

Abone koduna bakalım:

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.

Kaynak, bir mesajı uygun herhangi bir yerde yayınlamak için işlevi çağırabilir:

messaging:publish_message(Exchange, Key, Message).

Başka ürün ile değiştirme - değişim noktasının adı,
anahtar - yönlendirme anahtarı
Mesaj - yük

Pub-sub'ı genişleterek, günlüğe kaydetmeye uygun bir model elde edebilirsiniz. Kaynaklar ve tüketiciler kümesi tamamen farklı olabilir. Şekilde bir tüketicinin ve birden fazla kaynağın olduğu bir durum gösterilmektedir.

Görev dağıtım modeli

Hemen hemen her proje, rapor oluşturma, bildirimleri iletme ve üçüncü taraf sistemlerden veri alma gibi ertelenmiş işleme görevlerini içerir. Bu görevleri gerçekleştiren sistemin verimi, işleyiciler eklenerek kolayca ölçeklendirilebilir. Bizim için geriye kalan tek şey, bir işlemci kümesi oluşturmak ve görevleri aralarında eşit olarak dağıtmaktır.

3 işleyici örneğini kullanarak ortaya çıkan durumlara bakalım. Görev dağıtımı aşamasında bile dağıtımın adaleti ve sorumluların taşması sorunu ortaya çıkıyor. Adillikten sorumlu olacak ve işleyicilerin taşması durumundan kaçınmak için bir kısıtlama getireceğiz. prefetch_limit. Geçici koşullarda prefetch_limit bir işleyicinin tüm görevleri almasını engelleyecektir.

Mesajlaşma kuyrukları ve işlem önceliğini yönetir. İşlemciler görevleri geldikçe alırlar. Görev başarıyla tamamlanabilir veya başarısız olabilir:

messaging:ack(Tack) - mesaj başarıyla işlenirse çağrılır
messaging:nack(Tack) - tüm acil durumlarda aranır. Görev geri verildiğinde, mesajlaşma onu başka bir işleyiciye aktaracaktır.

Üç görevin işlenmesi sırasında karmaşık bir hatanın meydana geldiğini varsayalım: işlemci 1, görevi aldıktan sonra, değişim noktasına herhangi bir şey bildirmeye zaman bulamadan çöktü. Bu durumda değişim noktası, onaylama zaman aşımı süresi dolduktan sonra görevi başka bir işleyiciye aktaracaktır. Bazı nedenlerden dolayı işleyici 3 görevi bıraktı ve nack gönderdi; bunun sonucunda görev, başarıyla tamamlayan başka bir işleyiciye de devredildi.

Ön özet

Dağıtılmış sistemlerin temel yapı taşlarını ele aldık ve bunların Erlang/Elixir'deki kullanımına ilişkin temel bir anlayış kazandık.

Temel kalıpları birleştirerek ortaya çıkan sorunları çözmek için karmaşık paradigmalar oluşturabilirsiniz.

Serinin son bölümünde hizmetlerin organizasyonu, yönlendirme ve dengeleme gibi genel konulara bakacağız, ayrıca sistemlerin ölçeklenebilirliği ve hata toleransının pratik yönünden de bahsedeceğiz.

İkinci bölümün sonu.

Fotoğraf Marius Christensen
Websequencediagrams.com kullanılarak hazırlanan çizimler

Kaynak: habr.com

Dağıtılmış uygulamaların yapı taşları. İlk yaklaşım