Bir İşlem Hattı Çalıştırma
Bir grafiği somutlaştırıp çalıştırın.
Bir İşlem Hattı Çalıştırma, CoddyKit'te ücretsiz bir Scala for Backend Engineering & Functional Programming dersidir. Bu, 4 dersinin 4. dersidir. Aşağıdan dersin tamamını ücretsiz okuyabilir, sonra tarayıcıda yerleşik kod editörü ve 7/24 yapay zeka koçu ile uygulamalı olarak pratik yapabilirsin. Bu, Scala for Backend Engineering & Functional Programming öğrenme yolunun bir parçasıdır ve ilerlemeniz web ve CoddyKit uygulaması arasında senkronize olur. Scala for Backend Engineering & Functional Programming kursu toplamda 4 dersten oluşur.
Taslak Aşamasından Çalıştırmaya
Buraya kadar işlem hattı yalnızca saf bir taslaktı. Somutlaştırma, bu taslağı verileri gerçekten taşıyan çalışan actor'lere dönüştüren süreçtir.
Grafiği açıkça çalıştırana kadar hiçbir şey gerçekleşmez; Akka Streams'i birleştirilebilir ve yeniden kullanılabilir yapan da budur.
ActorSystem
Somutlaştırma, akışın aşamalarını destekleyen iş parçacıklarını ve dağıtıcıyı sağlayan bir ActorSystem gerektirir. Modern Akka'da sistem, örtük somutlaştırıcı olarak da görev yapar.
Tek bir ActorSystem genellikle bir uygulamanın tamamına ve eşzamanlı çalışan birçok akışa hizmet eder.
import akka.actor.ActorSystem
implicit val system: ActorSystem =
ActorSystem("data-pipeline")
import system.dispatcher // ExecutionContextrunWith
Bir Source'u çalıştırmanın en doğrudan yolu, bir Sink bağlayan ve tek adımda somutlaştırma yapan runWith'tir; bu işlem, söz konusu Sink'in somutlaştırılmış değerini döndürür.
Buradaki sonuç, akış tamamlandığında toplamla tamamlanan bir Future[Int]'tir.
import akka.stream.scaladsl.{Source, Sink}
import scala.concurrent.Future
val total: Future[Int] =
Source(1 to 100).runWith(Sink.fold(0)(_ + _))Bir RunnableGraph üzerinde run
to veya toMat ile kapalı bir RunnableGraph oluşturduysanız, onu somutlaştırmak için run() çağrısını yapın. Dönüş değeri, grafiğin koruduğu somutlaştırılmış değer neyse odur.
Bu, işlem hattının oluşturulmasını çalıştırılmasından temiz bir şekilde ayırır.
import akka.stream.scaladsl.Keep
import scala.concurrent.Future
val graph =
Source(1 to 100)
.toMat(Sink.fold(0)(_ + _))(Keep.right)
val result: Future[Int] = graph.run()Kolaylık Sağlayan run İşleçleri
Source'lar kısayollar sunar: runForeach, runFold ve runReduce, ilgili Sink'i bağlayıp hemen çalıştırır.
Bunlar, bir Source üzerindeki yaygın terminal işlemleri için kısa ve kullanışlıdır.
import scala.concurrent.Future
val printed: Future[akka.Done] =
Source(1 to 10).runForeach(println)
val sum: Future[Int] =
Source(1 to 10).runFold(0)(_ + _)Sonuç Future'ı ile Çalışma
Terminal Sink'ler, akış sona erdiğinde veya başarısız olduğunda tamamlanan bir Future döndürür. Başarıya veya hataya tepki vermek için onComplete ile geri çağırmalar kaydedin.
Bu geri çağırmalar için örtük ExecutionContext olarak ActorSystem'in dağıtıcısını kullanın.
import scala.util.{Success, Failure}
total.onComplete {
case Success(value) => println(s"Sum = $value")
case Failure(ex) => println(s"Failed: ${ex.getMessage}")
}Gerçekçi Bir İşlem Hattı
Tipik bir veri işleme hattı bir Kaynak'tan okur, Akışlarla dönüştürür, mapAsync ile eşzamansız G/Ç işlemleri gerçekleştirir, grouped ile gruplar ve bir Hedef'e yazar.
Her aşama küçüktür ve bütün yapı tek bir run ile gerçekleştirilir.
val done =
lineSource
.map(parse)
.mapAsync(4)(validate)
.grouped(500)
.runWith(bulkWriteSink)Başarısız Akışları Yeniden Başlatma
Dayanıklılık sağlamak için bir Kaynağı veya Akışı RestartSource.withBackoff ile sararak geçici hataların (örneğin bağlantının kopmasının) üstel bekleme süreleriyle otomatik olarak yeniden başlatmayı tetiklemesini sağlayabilirsiniz.
Bu, uzun süre çalışan veri alma hatlarını manuel denetim olmadan ayakta tutar.
import akka.stream.scaladsl.RestartSource
import akka.stream.RestartSettings
import scala.concurrent.duration._
val resilient = RestartSource.withBackoff(
RestartSettings(1.second, 30.seconds, 0.2))(() => flakySource)KillSwitch ile Düzenli Kapatma
Bir KillSwitch, dış kodun çalışan bir akışı düzgün biçimde durdurmasını sağlar. KillSwitches.single öğesini viaMat ile ekleyin ve daha sonra shutdown() çağrısını yapmak için oluşturulan değeri saklayın.
Bu, uygulama kapatılırken durdurulması gereken uzun ömürlü akışlar için zorunludur.
import akka.stream.{KillSwitches, KillSwitch}
import akka.stream.scaladsl.Keep
val (switch, done) =
source
.viaMat(KillSwitches.single)(Keep.right)
.toMat(Sink.ignore)(Keep.both)
.run()
// later: switch.shutdown()Kaynakları Serbest Bırakma
Uygulama çıktığında, iş parçacıklarını serbest bırakmak için ActorSystem'i sonlandırın. Kapatmanın düzenli gerçekleşmesi için terminate çağrısını akışın tamamlanmasını temsil eden gelecek değerin sonrasına zincirleyin.
Kapatılmayan bir ActorSystem, JVM'in çalışmaya devam etmesine ve kaynakların açık tutulmasına neden olur.
done.onComplete { _ =>
system.terminate()
}Gerçekleştiriciyi Yeniden Kullanma
Aynı taslağı birden çok kez gerçekleştirmek, ActorSystem'in kaynaklarını paylaşan bağımsız çalışan akışlar oluşturur. Taslağın kendisi değişmez ve yan etkisiz kalır.
Bu sayede bir veri işleme hattını bir kez tanımlayıp gelen her iş için isteğe bağlı olarak çalıştırmak güvenli olur.
val blueprint =
Source(1 to 5).toMat(Sink.seq)(Keep.right)
val run1 = blueprint.run()
val run2 = blueprint.run() // independent executionHızlı Kontrol
Bir akışın öğeleri gerçekten işleyebilmesi için nelerin gerekli olduğunu düşünün.
Özet
Bir veri işleme hattını çalıştırmak, bir taslağı bir ActorSystem ile run, runWith veya kolaylık işleçlerini kullanarak gerçekleştirmek demektir; bu işlemlerin her biri gelecekte elde edilecek bir sonuç döndürür.
Gerçekçi çok aşamalı veri işleme hatlarını, bekleme süreleriyle otomatik yeniden başlatmayı, KillSwitch aracılığıyla düzenli kapatmayı, system.terminate() ile kaynakların temizlenmesini ve değişmez bir taslağın birbirinden bağımsız çalıştırmalar arasında güvenli biçimde yeniden kullanılmasını gördünüz.
Yapay zeka eğitmeniyle Scala öğren — ücretsiz
Tarayıcında gerçek kod yaz ve çalıştır, 7/24 yapay zeka eğitmeninden anında yardım al; web'de ya da uygulamada kaldığın yerden devam et.
- Kurslar
- 39
- Dersler
- 143
Sıkça Sorulan Sorular
“Bir İşlem Hattı Çalıştırma” dersi ücretsiz mi?
Evet — “Bir İşlem Hattı Çalıştırma” dersin tüm metni burada web'de ücretsiz olarak okunabilir. Etkileşimli olarak pratik yapmak (yerleşik kod editörü ve 7/24 yapay zeka koçu) ve Scala for Backend Engineering & Functional Programming kursunun geri kalanını açmak için CoddyKit PRO'ya yükselt. Scala for Backend Engineering & Functional Programming kursu toplamda 4 dersten oluşur.
“Bir İşlem Hattı Çalıştırma” dersinde ne öğreneceğim?
Bir grafiği somutlaştırıp çalıştırın. Scala for Backend Engineering & Functional Programming ile uygulamalı kodu tarayıcıda doğrudan çalıştırarak pratik yaparsın ve 7/24 yapay zeka koçu dersi çalışırken sorularını yanıtlar.
Scala for Backend Engineering & Functional Programming öğrenmeye başlamak için deneyim gerekli mi?
Önceden deneyim gerekmez. CoddyKit'te Scala for Backend Engineering & Functional Programming, başlangıçtan ileri seviyeye kadar yapılandırıldığı için buradan başlayabilir veya başından başlayıp kendi hızında ilerleme yapabilirsin. Bu, 4 dersinin 4. dersidir.
“Bir İşlem Hattı Çalıştırma” dersi ne kadar sürer?
Çoğu CoddyKit dersi yaklaşık 5–10 dakika sürer. Her biri kısa ve etkileşimli olduğu için sabit ilerleme yaparsın ve web ile uygulama arasında tam olarak bıraktığın yerden devam edebilirsin.
Bu Scala for Backend Engineering & Functional Programming dersinde kod yazıp çalıştırabilir miyim?
Evet. Her Scala for Backend Engineering & Functional Programming dersi yerleşik bir kod editörü içerir, bu sayede tarayıcıda gerçek kod yazıp çalıştırabilir ve anlık yapay zeka geri bildirimi alırsın — yerel kurulum gerekli değildir.
Bu kursun tüm dersleri
- Source, Flow ve Sink
- Akışları Dönüştürme
- Geri Basınç
- Bir İşlem Hattı Çalıştırma