מבוא לריאקטור – Intro to Reactor

שלום לכולם,
בפוסט הקודם בסדר כתבתי קצת מה זה תכנות ריאקטיבי,
מה ההיתרונות, ומה החסרונות של הפרדיגמה הזאת.
הפעם אנחנו נתחיל ללמוד על אחד המימושים הפופולריים לתכנות ריאקטיבי בג'אווה.
אני כמובן מדבר על ספריית Reactor.

ריאקטור היא מימוש מלא של תכנות ריאקטיבי עבור JVM, עם ניהול דרישות יעיל.
היא משתלבת ישירות עם ה-APIs של Java 8.
היא נותנת לנו APIs ברי הרכבה לניהול קוד אסינכרוני בצורת Flux, Mono (עוד עליהם בהמשך).

יאללה מתחילים

קודם כל, את רוב המידע שאתם צריכים אפשר למצוא בעמוד של ספריית ריאקטור.
אחרי שהלכתם קצת להסתכל קצת מה הולך שם, הדבר הראשון שאני ממליץ לעשות הוא לייבא את הספרייה לפרויקט לימוד שלכם.
איך עושים את זה?
ובכן יש כמה דרכים לעשות את זה, תלוי באיזה framework אתם משתמשים.

Maven

האופציה הראשונה היא דרך Maven.

אם אתם משתמשים ב Gradle, דלגו למטה לאזור שמסביר איך להוסיף ב Gradle.
כדי לייבא את הספרייה דרך maven, אתם צריכים להכניס את הקוד הבא אל הקובץ pom.xml שלכם:

<dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>io.projectreactor</groupId>
      <artifactId>reactor-bom</artifactId>
      <version>2022.0.0</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>

שימו לב ל- dependencyManagement, שהוא תוספת ל-dependency section הרגיל שיש בקבצי pom.xml
אם הוא כבר קיים לכם, אז הוסיפו רק את התוכן שלו

לאחר מכן הוסיפו את את ה-dependency של ה reactor project בצורה הבאה

<dependencies>
  <dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-core</artifactId>
  </dependency>
  <dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-test</artifactId>
    <scope>test</scope>
  </dependency>
</dependencies>

הראשון זה התלות ל -core library של reactor.
שימו לב שאנחנו לא מציינים לאיזו גרסה אנחנו מתכוונים.
התלות השניה היא עבור טסטים.

Gradle

על מנת לייבא את הפרויקט דרך Gradle, אתם צריכים שתהיה לכם תמיכה ב- gradle-dependency-management פלאגין.
כלומר Gradle מגרסה 5 ומעלה.

על מנת להוסיף תמיכה ב-reactor כל מה שעלינו לעשות הוא לייבא את ה-BOM ואת הספרייה עצמה כ-dependencies.

dependencies {
  implementation platform('io.projectreactor:reactor-bom:2020.0.24')
  implementation 'io.projectreactor:reactor-core'
}

Reactive Streams

על מנת להבין את ספריית reactor מומלץ בחום להבין מה הם reactive streams,
כיוון ש-reactor מתבססת על ה-API שהם הגדירו כדי להגדיר את אבני הבניין שלה.

אתם יכולים לקרוא לעומק על reactive stream כאן
אבל הנה סיכום קצר

המטרה של reactive streams היא לנהל את העברת המידע בין גבולות בקוד אסינכרוני
למשל העברה של מידע בין threads או threads-pool
ולוודא שהצד המקבל לא נאלץ לצבור כמויות גדולות של מידע (buffering).
הרעיון הוא ליצור תשתית אשר תאפשר ליצור תורים בשני הצדדים, בצד השולח ובצד המקבל
אשר הם מוגדרים בגודם, ולא אינסופיים (bounded).

ה-reactive streams מגדירים כמה סוגים של אותות (signals) אשר יכולים לעבור בין ערוצי המידע השונים:

• onNext – הודעה על אלמנט חדש שעובר בערוץ
• onError – קרתה שגיאה
• onComplete – הסתיים המידע שצריך להעביר והכל עבר בהצלחה
• onSubscribe – מישהו נרשם לרצף מידע

ה-API ש-reactive streams מגדירים מורכב מארבעה חלקים:

• Publisher
• Subscriber
• Subscription
• Processor

ה-Publisher הוא ספק של מספר לא מוגבל של איברים (0…N) ומפרסם אותם לפי הדרישה של ה-subscribers שלו.
הסיגנלים שהוא יודע להעביר הם:

onSubscribe onNext* (onError | onComplete)?

כלומר שהוא חייב לשלוח את ה-onSubscribe, ואז יש אפשרות לכמות לא מוגבלת של onNext לפי הבקשה של ה-subscriber.
ובסוף יגיע סיגנל של onError אם היתה שגיאה או onComplete כאשר אין יותר אלמנטים לפרסם.
זה נכון כל עוד ה-subscription לא בוטל.

הבסיס של reactor

הספריה הבסיסית של reactor היא reactor-core, והיא זאת שהבאנו בסעיף הקודם.
Reactor מציגה טיפוסים ריאקטיבים הניתנים להרכבה אשר מממשים את Publisher, אבל גם נותנים לנו הרבה אופרטורים לעבוד איתם:
ה-Flux וה-Mono.
אלו הן אבני הבניין של reactor, שעל בסיסן נבנה כל הקוד האסינכרוני.

ה-Flux הוא אובייקט המייצג רצף מידע אסינכרוני של 0…N איברים, כלומר איברים מרובים.
ה-Mono מייצג ערך בודד, שיכול להיות ריק (0…1)

בואו נכיר את שני האובייקטים הללו יותר לעומק

Flux

התמונה הבאה מציגה בצורה חזותית את הצורה ב-Flux מנהל את האובייקטים שעוברים דרכו:

בקצרה, השורה העליונה היא רצף הזמן בו ה-Flux שלנו מקבל את האובייקטים שלו.
אני מזכיר שאנחנו מדברים על תכנות ריאקטיבי ואסינכרוני, כלומר שהאיברים מגיעים לאורך ציר זמן מסויים ואנחנו מטפלים בהם כשהם מגיעים אלינו.
כל עיגול הוא איבר שה-Flux מטפל בו
הקו האנכי בסוף הקו העליון מציין שה-Flux סיים את הפעולה שלו בצורה תקינה.
אחרי שה-Flux מקבל פריט מידע, הוא מפעיל עליו אופרטור כלשהו (בהמשך נכיר את האופרטורים).
השם בתוך המלבן, הוא האופרטור אותו אנחנו מפעילים.
הקו התחתון הוא התוצאה של האופרטור.
אם יש עיגול, סימן שהאופרטור עבד כמו שצריך, ואם יש שם איקס אדום, אז קרתה לנו שגיאה.

כל Flux מצפה לקבל אובייקטים מסוג מסוים (בדומה לאיך שכותבים List), ולכן נהוג לציין אותו כ Flux<T>

והוא publisher סטנדרטי, אשר מייצג רצף מידע אסינכרוני של איברים מרובים (0…N)
והוא יכול להסתיים בסיגנל של completion או בשגיאה.
כמו שמוגדר ב-reactive streams יש שלושה סיגנלים שהוא יכול לשלוח: onNext, onComplete, onError.
כל אחד מהסיגנלים הללו הוא מתודה של המחלקה Flux.

ל-Flux אין מטרה ספציפית, אלא הוא פתוח לשימוש גנרי.
וחשוב לדעת שאין חובה להשתמש באף אחד מהסיגנלים, כולם אופציונאליים.
אם לא נשלח onNext אבל כן נשלח onComplete – זה יציין רצף ריק וסופי של מידע, כלומר אנחנו לא מקבלים איברים ויודעים לקבל הודעה על סיום הרצף.
אבל אם נסיר את ה-onComplete, אז קיבלנו רצף ריק ואינסופי של מידע, כיוון שאנחנו לא שולחים הודעה על סיום. (לא שימושי במיוחד, חוץ מטסטים).

Mono

בדומה ל-Flux גם כאן יש לנו תיאור של המידע המנוהל על ידי ה-Mono.
ההבדל העיקרי הוא ש-Mono מנהל רק איבר אחד, אם בכלל.

גם ה-Mono הוא סוג של Publisher גנרי אשר יודע לקבל סוג מסוים של מידע בצורה. והוא מוגדר ככה Mono<T>

הוא יכול לשלוח סיגנל יחיד של onNext כאשר מגיע האיבר שמצפים לו, ויכול לסיים ב-onComplete או ב-onError.
עם אותן משמעויות כמו של ה-Flux.

רוב המימושים של Mono מצפים לקבל קריאה של onComplete על ה-subscriber שלהם אחרי שקוראים ל onNext.
כיוון שאין צפי לקבל עוד איברים אחרי האיבר הזה.

ה-Mono מאפשר שימוש רק בחלק מהאופרטורים שיש ל-Flux
ויש אופרטורים שיודעים לקחת Mono ולהחזיר Flux (בשילוב עם publisher אחר).

שימו לב שאתם יכולים להשתמש ב-Mono על מנת תהליך אסינכרוני שלא עובר בו שום מידע
אבל כן יש משמעות ל-onComplete.
כדי ליצור אחד אנחנו צריכים להשתמש ב-Mono ריק Mono<Void>

אז איך יוצרים Flux או Mono ונרשמים אליהם?

הדרך הקלה ביותר ליצור Flux או Mono היא להשתמש באחד מה-factory methods שיש להם.
למשל כדי ליצור רצף של Strings אפשר להשתמש באיטראטור, או להכניס אותם ל-collection ולבנות מזה Flux

הדוגמה הראשונה מבצעת enumerate על האיברים
ואילו הדוגמה השניה היא עם collection.

דוגמאות נוספות של יצירה כוללות:

שימו לב שבשורה הראשונה ה-factory method מכבד את הסוג הגנרי למרות שאין שום מידע שעובר בתוכו.
בדוגמה האחרונה הפרמטר הראשון הוא האיבר הראשון שאנחנו מעבירים, והפרמטר השני הוא כמה איברים אנחנו שולחים.

סיכום

בפוסט הזה התחלנו לדבר על ספריית reactor
שהיא המימוש של תכנות ריאקטיבי ל-JVM
למדנו מהם ה-reactive streams ומהו הAPI שהם מגדירים שריאקטור מממשת.
הכרנו את שני אבני הבניין העיקריות של ריאקטור Flux ו-Mono.
ויצרנו כמה כאלה בכמה דוגמאות שונות.

בפעם הבאה נלמד איך נרשמים לכל אחד מהם, ואיך מקבלים את המידע שהם מעבירים

את הקוד המלא אפשר למצוא כאן
וכמו תמיד, אני אשמח לקבל כל הערה, הארה או שאלה שיש לכם