لنفترض أن لدينا مثل هذه المهمة.
هناك مصدر للمعاملات في سوق الأوراق المالية. يرسل لنا هذا المصدر المعاملات عبر واجهة Rest.
نحتاج إلى الحصول على هذه المعاملات وحفظها في قاعدة البيانات وإنشاء مساحة تخزين ملائمة في الذاكرة.
يجب أن يؤدي هذا المستودع الوظائف التالية:
- إعادة قائمة الصفقات ؛
- إعادة الوظيفة الكاملة ، أي جدول "الصك" - "العدد الحالي للأوراق المالية" ؛
- عودة الموقف لأداة معينة.
كيف نتعامل مع هذه المهمة؟
وفقًا لمبادئ أسلوب الخدمات المصغرة ، نحتاج إلى تقسيم المهمة إلى مكونات الخدمات المصغرة:
- استلام معاملة بواسطة Rest ؛
- حفظ المعاملة في قاعدة البيانات ؛
- تخزين في الذاكرة لعرض بيانات الموقع.
دعنا نجعل الخدمتين الأولى والثالثة في إطار هذا البرنامج التعليمي ، ونترك الثانية للجزء الثاني (اكتب في التعليقات إذا كانت مثيرة للاهتمام).
لذلك ، لدينا خدمتان صغيرتان.
الأول يتلقى البيانات من الخارج.
الثاني يعالج هذه البيانات ويستجيب للطلبات الواردة.
بالطبع ، نريد الحصول على مقياس أفقي وتحديثات مستمرة وفوائد أخرى للخدمات المصغرة.
ما هي المهمة الصعبة جدا التي تنتظرنا؟
يوجد الكثير منهم بالفعل ، لكن دعنا الآن نتحدث عن كيفية تدفق البيانات بين هذه الخدمات الصغيرة. يمكنك أيضًا جعل الراحة بينهما ، يمكنك وضع نوع من قائمة الانتظار ، يمكنك الخروج بالكثير من الأشياء بمزاياها وعيوبها.
دعونا نلقي نظرة على نهج واحد ممكن - الاتصال غير المتزامن من خلال إطار عمل Axon .
ما هي مزايا هذا الحل؟
أولاً ، يزيد الاتصال غير المتزامن من المرونة (نعم ، هناك ناقص هنا ، لكننا نتحدث فقط عن المحترفين حتى الآن).
ثانيًا ، نحصل على Event Sourcing و CQRS مباشرة من الصندوق .
ثالثًا ، توفر Axon بنية تحتية جاهزة ، ونحتاج فقط إلى التركيز على تطوير منطق الأعمال.
هيا بنا نبدأ.
سيكون لدينا المشروع على gradle. سيكون له ثلاث وحدات:
- مشترك. وحدة مع هياكل البيانات المشتركة (نحن لا نحب النسخ واللصق) ؛
- خالق التجارة. وحدة مع خدمة مصغرة لقبول المعاملات على الراحة ؛
- استفسارات. وحدة مع خدمة مصغرة لعرض الموقف.
لنأخذ Spring Boot كأساس ونقوم بتوصيل بداية Axon.
تعمل Axon بشكل جيد بدون الربيع ، لكننا سنستخدمها معًا.
نحن هنا بحاجة إلى التوقف وإخبارك ببعض الكلمات عن Axon.
إنه نظام خادم العميل. يوجد خادم - هذا تطبيق منفصل ، سنقوم بتشغيله في docker.
وهناك عملاء يقومون بتضمين أنفسهم في الخدمات المصغرة.
هذه هي الصورة. أولاً ، يتم تشغيل خادم Axon (في عامل الإرساء) ، ثم خدماتنا المصغرة.
عند بدء التشغيل ، تبحث الخدمات المصغرة عن خادم وتبدأ في التفاعل معه. يمكن تقسيم التفاعل بشكل مشروط إلى نوعين: التقني والتجاري.
العامل التقني هو تبادل الرسائل "أنا على قيد الحياة" (يمكن رؤية مثل هذه الرسائل في وضع تسجيل التصحيح).
الأعمال تحجبها رسائل مثل "صفقة جديدة".
ميزة مهمة ، بعد بدء الخدمة المصغرة ، يمكن أن تطلب من خادم Axon "ماذا حدث" ويرسل الخادم الأحداث المتراكمة إلى الخدمة المصغرة. وبالتالي ، يمكن إعادة تشغيل الخدمة المصغرة بأمان نسبيًا دون فقدان البيانات.
باستخدام مخطط التبادل هذا ، يمكننا بسهولة تشغيل العديد من مثيلات الخدمات المصغرة
وعلى مضيفين مختلفين.
نعم ، هناك مثيل واحد من Axon Server غير موثوق به ، ولكن حتى الآن.
نحن نعمل في نماذج تحديد مصادر الأحداث ونماذج CQRS. هذا يعني أنه يجب أن يكون لدينا "فرق" و "أحداث" و "عينات".
سيكون لدينا أمر واحد: "إنشاء صفقة" ، وحدث واحد "تم إنشاء صفقة" وثلاثة اختيارات: "إظهار كل الصفقات" ، "إظهار المركز" ، "إظهار موضع للأداة".
مخطط العمل على النحو التالي:
- تقبل خدمة TradeCreator المصغرة معاملة Rest.
- تنشئ الخدمة المصغرة الخاصة بـ tradeCreator أمر "إنشاء تجارة" وترسله إلى خادم Axon.
- يتلقى خادم Axon الأمر ويعيد توجيه الأمر إلى المستلم المهتم ، وفي حالتنا هذه هي خدمة التجارة المصغرة tradeCreator.
- تتلقى خدمة tradeCreator الصغيرة أمرًا ، وتقوم بإنشاء حدث "تم إنشاؤه بصفقة" وإرساله إلى خادم Axon.
- يستقبل خادم Axon الحدث ويعيد توجيهه إلى المشتركين المهتمين.
- الآن لدينا مستلم واحد مهتم فقط - خدمة tradeQueries الصغيرة.
- تستقبل الخدمة الصغيرة الخاصة بـ tradeQueries الحدث وتقوم بتحديث البيانات الداخلية.
(من المهم أنه في اللحظة التي يتم فيها تكوين الحدث ، قد لا تتوفر خدمة tradeQueries Microservice ، ولكن بمجرد أن تبدأ ، ستتلقى الحدث على الفور).
نعم ، الخادم المحوري في مركز الاتصال ، تمر جميع الرسائل من خلاله.
دعنا ننتقل إلى الترميز.
من أجل عدم ازدحام المنشور بالرمز ، سأقدم أدناه أجزاء فقط ، وسيكون الرابط إلى المثال بأكمله أدناه.
لنبدأ بالوحدة المشتركة.
الأجزاء المشتركة فيه هي الحدث (فئة CreatedTradeEvent). انتبه إلى الاسم ، في الواقع ، هذا هو اسم الفريق الذي أنشأ هذا الحدث ، ولكن في زمن الماضي. في الماضي بسبب أولاً ، يظهر الأمر ، مما يؤدي إلى إنشاء الحدث.
تشمل الهياكل الشائعة الأخرى فئات لوصف المركز (مركز الفئة) ، والتجارة (تجارة الطبقة) وجانب التجارة (جانب التعداد) ، أي شراء أو بيع.
دعنا ننتقل إلى وحدة tradeCreator.
تحتوي هذه الوحدة على واجهة Rest (فئة TradeController) لقبول التداولات.
يتم تكوين الأمر "إنشاء صفقة" من الصفقة المستلمة وإرسالها إلى خادم المحور.
@PostMapping("/trade")
public ResponseEntity<String> create(@RequestBody Trade trade) {
var createTradeCommand = CreateTradeCommand.builder()
.tradeId(trade.getTradeId())
...
.build();
var result = commandGateway.sendAndWait(createTradeCommand, 3, TimeUnit.SECONDS);
return ResponseEntity.ok(result.get().toString());
}
لمعالجة الأمر ، يتم استخدام الفئة TradeAggregate.
للعثور عليه ، نضيف التعليق التوضيحيAggregate.
تبدو طريقة معالجة الأمر كما يلي (مع اختصار):
@CommandHandler
public TradeAggregate(CreateTradeCommand command) {
log.info("command: {}", command);
var event = CreatedTradeEvent.builder()
.tradeId(command.tradeId())
....
.build();
AggregateLifecycle.apply(event);
}
يتم إنشاء حدث من الأمر وإرساله إلى الخادم.
الأمر موجود في فئة CreateTradeCommand.
الآن دعنا نلقي نظرة على آخر وحدة tradeQueries.
التحديدات موصوفة في حزمة الاستعلامات.
تحتوي هذه الوحدة أيضًا على واجهة
TradeController Rest من الفئة العامة.
على سبيل المثال ، دعنا نرى معالجة الطلب: "إظهار جميع المعاملات".
@GetMapping("/trade/all")
public List<Trade> findAllTrades() {
return queryGateway.query(new FindAllTradesQuery(),
ResponseTypes.multipleInstancesOf(Trade.class)).join();
}
يتم إنشاء طلب الجلب وإرساله إلى الخادم.
يتم استخدام فئة TradesEventHandler لمعالجة طلب الجلب.
لها طريقة مشروحة
@QueryHandler
public List<Position> handleFindCurrentPositionQuery(FindCurrentPositionQuery query)
هو المسؤول عن جلب البيانات من التخزين في الذاكرة.
السؤال الذي يطرح نفسه حول كيفية تحديث المعلومات في هذا المتجر.
بادئ ذي بدء ، هذه مجرد مجموعة من خرائط ConcurrentHashMaps المصممة لاختيارات معينة.
لتحديثها ، يتم تطبيق الطريقة:
@EventHandler
public void on(CreatedTradeEvent event) {
log.info("event:{}", event);
var trade = Trade.builder()
...
.build();
trades.put(event.tradeId(), trade);
position.merge(event.shortName(), event.size(),
(oldValue, value) -> event.side() == Side.BUY ? oldValue + value : oldValue - value);
}يتلقى الحدث "تم إنشاء الصفقة" ويقوم بتحديث الخرائط.
هذه هي النقاط البارزة في تطوير الخدمات المصغرة.
ماذا عن عيوب Axon؟
أولاً ، هذا هو تعقيد البنية التحتية ، فقد ظهرت نقطة فشل - خادم Axon ، تمر جميع الاتصالات من خلاله.
ثانيًا ، يتجلى بوضوح عيب هذه الأنظمة الموزعة - عدم تناسق البيانات المؤقت. في حالتنا ، قد يمر وقت طويل بشكل غير مقبول بين تلقي صفقة جديدة وتحديث بيانات العينات.
ماذا بقي وراء الكواليس؟
لم يُقال أي شيء على الإطلاق عن "مصادر الأحداث" و "CQRS" ، وما هو وما هو الغرض منه.
بدون الكشف عن هذه المفاهيم ، قد لا تكون بعض النقاط واضحة.
ربما تتطلب بعض أجزاء التعليمات البرمجية توضيحًا أيضًا.
تحدثنا عن هذا في ندوة مفتوحة عبر الإنترنت .
مثال كامل .