Spring Boot + Kafka Producer API

წინა პოსტში განვიხილეთ, როგორ გავაგზავნოთ და მივიღოთ შეტყობინებები Apache Kafka-ს მეშვეობით command line-ის მეშვეობით. ახლა ცოტა წინ წავიწიოთ და ვნახოთ როგორ შეიძლება ეს ყველაფერი გავაკეთოთ Java-ში, კერძოდ კი Spring Boot აპლიკაციით. ჯერ საწყის ეტაპზე, შეგვიძლია შევქმნათ მხოლოდ Producer აპლიკაცია. ამისთვის შევიდეთ Spring-ის initializer-ზე (https://start.spring.io)  და დავამატოთ შესაბამისი dependency-ები.

Generate Project ღილაკზე დაჭერით გადმოვწეროთ და ამოვაარქივოთ გადმოწერილი zip ფაილი. შემდეგ გავხსნათ ჩვენთვის სასურველ IDE-ში (ამ შემთხვევაში IntelliJ).

საწყის ეტაპზე application.properties ფაილში ჩავწეროთ მისამართი იმ სერვერისა, რომელზეც დაინსტალირებულია Kafka.

ახლა შევქმნათ KafkaProducerConfig კლასი, რომ დავაკონფიგურიროთ Producer. საწყის ეტაპზე გავაგზავნოთ მხოლოდ String ტიპის შეტყობინებები, ხოლო შემდეგში ვნახავთ თუ როგორ გავაგზავნოთ ობიექტები.

ყურადღება მივაქციოთ იმას, რომ KEY_SERIALIZER_CLASS_CONFIG-სა და VALUE_SERIALIZER_CLASS_CONFIG ის მნიშვნელობები ორივე არის StringSerializer.class, ხოლო KafkaTemplate-ც შესაბამისად არის KafkaTemplate<String,String>. bootstrapAddress არის მნიშვნელობა, რომელსაც @Value ანოტაციით ვკითხულობთ application.properties ფაილიდან.

ახლა შევქმნათ REST endpoint. RestController, სადაც მარტივად მივიღებთ HttpRequest-ს და გავაგზავნით შეტყობინებას, რომელსაც Consumer მიიღებს (ამ ეტაპზე Consumer იქნება command line-ით).

თუ ჩვენს Spring Boot აპლიკაციას გავუშვებთ და გამოვიძახებთ შემდეგ ბრძანებას (URL-ს) http://localhost:8080/publish/message?msg=Hello  და პარალელურად Consumer დასტარტული გვექნება სერვერზე (sudo kafka-console-consumer.sh --zookeeper localhost:2181 --topic kafka_topic  --from-beginning) , რომელიც მოუსმენს kafka_topic TOPIC-ს, მაშინ Command line-ში მივიღებთ შემდეგ შეტყობინებას

ახლა ვნახოთ როგორ შეიძლება Spring Boot აპლიკაციით გავაგზავნოთ Java ობიექტი. ამისთვის საჭიროა მცირედი ცვლილებების შეტანა კონფიგურაციაში. VALUE_SERIALIZER_CLASS_CONFIG-ის მნიშვნელობა უნდა გავხადოთ JsonSerializer.class და ProducerFactory უნდა იყოს ProducerFactory<String,User> , თუ გვინდა User კლასის ტიპის ობიექტების გაგზავნა. შესაბამისად KafkaTemplate-ც უნდა იყოს KafkaTemplate<String,User> ტიპის.

რა იცვლება Producer კლასში (RestController-ში) ? ამ ეტაპისთვის, შევქმნათ User კლასის ობიექტი მეთოდშივე, რომლითაც გვინდა გაგზავნა. User კლასს აქვს მხოლოდ ორივე ველი id და username და შესაბამისად getter და setter მეთოდები.

თუ დავარესტარტებთ ჩვენს აპლიკაციას და გამოვიძახებთ შემდეგ URL-ს

http://localhost:8080/publish/object

და Consumer API ძველებურად დასტარტულია და უსმენს kafka_topic TOPIC-ს, მაშინ command line-ში უნდა მივიღოთ user ობიექტის შესაბამისი JSON String.

Java 8 HashMaps, Keys და Comparable Interface

Java 8-ში წინა ვერსიებთან შედარებით ბევრი რამ გაუმჯობესებულია. ბევრი კლასი და-update-და. მათ შორის გამონაკლისი არ ყოფილა HashMap, რომელიც ერთ-ერთი ყველაზე ფართოდ გამოყენებადია. ამ პოსტში გავამახვილებ ყურადღებას იმაზე, თუ რა ხდება როცა Hash კოლიზიას აქვს ადგილი.

უპირველეს ყოვლისა, რომელი არის ყველაზე მარტივი გზა იმისა, რომ მოვახდინოთ კოლიზია HashMap-ში ? შევქმნათ კლასი, რომელსაც ექნება თავისი hash ფუნქცია და დავწეროთ მისი ყველაზე ცუდი იმპლემენტაცია (hashCode მეთოდის). ამის ერთ-ერთი მაგალითია ის, რომ hashCode() მეთოდმა ყოველთვის დააბრუნოს რაიმე კონსტანტა. როგორ ფიქრობთ რა მოხდება ამ დროს ? ტექნიკურ ინტერვიებზე ამ კითხვაზე ხშირად პასუხობენ, რომ ყოველი ახალი ჩამატებული ელემენტი HashMap-ში შეცვლის ძველს, რაც სიმართლეს არ შეესაბამება. Hash Collision არ გულისხმობს იმას, რომ HashMap-ში მონაცემები ერთმანეთს გადაეწერება, ეს ხდება მაშინ და მხოლოდ მაშინ როცა ისეთი ორი ელემენტის ჩასმას ვცდილობთ HashMap-ში, რომელთა Key-ები ტოლია equals() მეთოდზე დაყრდნობით, ანუ equals() მეთოდი აბრუნებს true-ს. ელემენტები არა ტოლი გასაღებებით და ერთნაირი hashCode-ებით მოთავსდება ერთსა და იმავე bucket-ში რომელიღაც ტიპის მონაცემთა სტრუქტურაში (შესამჩნევია რომ ამ შემთხვევაში map-ის საერთო performance გაუარესდება, რადგან ელემენტის ჩასმაც და წაკითხვაც უფრო დიდ დროს წაიღებს).

დასაწყისისთვის მოვიყვანოთ პატარა მაგალითი, რომელიც რა თქმა უნდა, გაზვიადებულია და პრაქტიკაში ნაკლებათ მოსახდენია (თუ საერთოდ მოხდა), მაგრამ ეს მაგალითი კარგად დაგვანახებს განსხვავებას. ამ მაგალითში, კოლიზიები იმაზე მეტი იქნება, ვიდრე ეს რეალურად პრაქტიკაში მოხდება.

მაგალითში, Person ტიპის ობიექტი გვქონდეს HashMap-ის გასაღებად, ხოლო მნიშვნელობები იყოს String.

ახლა კი მოვახდინოთ კოლიზია.

ამ კოდმა საკმაოდ მძლავრ კომპიუტერზე მუშაობას მოანდომა 2.5 საათი და საბოლოოდ ელემენტის წაკითხვამ კი დაახლოებით 40 მილიწამი. ახლა, ყოველგვარი წინასწარი ახნის გარეშე მოდით მოვახდინოთ Peron კლასის მცირე ცვლილება, მოდით მან იმპლემენტაცია გაუკეთოს Comparable<Person> ინტერფეისს და Person კლასში დავამატოთ კოდის შემდეგი ფრაგმენტი 

ზემოთ აღნიშნულმა კოდმა, იმავე კომპიუტერზე, მუშაობას მოანდომა ერთ წუთზე ნაკლები და საბოლოოდ ელემენტის წაკითხვას 1 მილიწამი - ანუ ის 150-ჯერ სწრაფად მუშაობს ამ შემთხვევაში. 

მოდით ახლა ავხსნათ რა მოხდა. როგორც დასაწყისში აღვნიშნეთ, Java 8-ში წინა ვერსიებთან შედარებით ბევრი კლასი გაუმჯობესდა. Java 7-ში კოლიზირებული (ერთი და იმავე hashCode-ის მქონე ელემენტები) ინახებოდა bucket-ში რომლის სტრუქტურაც იყო LinkedList. Java 8-დან დაწყებული, თუ კოლიზირებული ელემენტების რიცხვი არის გარკვეულ რიცხვზე მეტი (ეს რიცხვი არის 8), და Map-ის capacity არის 64-ზე მეტი, მაშინ HashMap-ის იმპლემენტაცია LinkedList-ის ნაცვლად იქნება BinaryTree.

აქ გაგვიჩნდება აზრი, რომ non-comparable გასაღებების შემთხვევაში, ეს ხე იქნება არაბალანსირებული, მაშინ როცა სხვა შემთხვევაში ის იქნება მეტ-ნაკლებად ბალანსირებული. მაგრამ ეს არ შეესაბამება სიმართლეს. Tree-ის იმპლემენტაცია HashMap-ში არის Red-Black Tree, რაც იმას ნიშნავს, რომ ის ყოველთვის არის ბალანსირებული. თუ ჩვენ დავწერთ utility კლასს, რომელიც ზემთ აღნიშნული მაგალითისთვის შეამოწმებს ხის სიღრმეს, ის იქნება 19.

ახლა მთავარი კითხვა, როგორ მივიღეთ ასეთი დიდი განსხვავება Comparable ინტერფესისის იმპლემენტაციით ? როცა HashMap ცდილობს იპოვოს ხეში ახალი ელემენტისთვის ადგილი, ის ამოწმებს ახალი და ძველი ელემენტები (უკვე არსებული) არის თუ არა ადვილად სადარი (Comparable interface). ჩვენი მაგალითის პირველ შემთხვევაში ის იძახებს tieBreakOrder(Object a, Object b) მეთოდს. ეს მეთოდი პირველად ცდილობს ობიექტების შედარებას class name-ზე დაყრდნობით, და შემდეგ იყენებს System.identityHashCode-ს. მაგრამ, როცა გასაღები აიმპლემენტირებს Comparable-ს ეს პროესი გაცილებით მარტივია. გასაღები (key), თვითონვე განსაზღვრავს როგორ უნდა შეედაროს სხვა ელემენტს და ელემენტის ჩამატება და წაკითხვა უფრო სწრაფად ხდება.  tieBreakOrder(Object a, Object b)  მეთოდის გამოძახება ხდება მხოლოდ მაშინ თუ ორი ელემენტის compareTo მეთოდი დააბრუნებს 0-ს (ანუ თუ ტოლია).

რომ შევაჯამოთ, Java 8-ში, თუ HashMap-ის bucket-ში ელემენტების რიცხვა გადააჭარბა 8-ს და HashMap-ის ზომამ 64-ს, მაშინ bucket-ის იმპლემენტაცია LinkedList-ის ნაცვლად იქნება Red-Black Tree, რომელიც ყოველთვის დაბალანსებულია და თუ Map-ის გასაღები აიმპლემენტირებს Comparable-ს, მაშინ ელემენტის ჩამატება და წაკითხვა იქნება გაცილებით სწრაფი. მაგრამ უნდა აღინიშნოს ისიც, რომ თუ კოლიზიების რიცხვი არც ისე დიდია, Comparable/ non-Comparable იდეას უკვე დიდი აზრი აღარ აქვს.

Java StreamAPI Examples

 სანამ უშუალოდ მაგალითების განხილვაზე გადავიდოდეთ, მოდით ორი სიტყვით შევეხოთ იმას თუ რა არის Stream-ები ჯავაში და რატომ არის მისი გამოყენება უკეთესი. Stream-ის ცნება ჯავაში მე-8 ვერსიიდან გამოჩნდა და თავიდანვე საკმაოდ პოპულარული გახდა დეველოპერებში. Stream არის ობიექტების მიმდევრობა, რომელზეც შეგვიძლია განვახორციელოთ აგრეგაციული ოპერაციები და ასევე ფილტრები, დაჯგუფებები, გარდაქმნები, სტატისტიკური შეჯამებები.  მოდით ორი სიტყვით შევეხოთ Stream-ის თვისებებს:

• Sequence of Elements  - Stream გვაძლევს კონკრეტული ტიპის ელემენტების მიმდევრობას და მას შეუძლია ელემენტის ამოღება და გამოთვლა მოთხოვნების შესაბამისად. ის არასდროს ინახავს ელემენტს.
• Source - Stream იღებს კოლექციებს, მასივებს, I/O რესურსებს, როგორც მისი input.
• Aggregate Operations - Stream-ს შეუძლია განახორციელოს ისეთი აგრეგაციული ოპერაციები, როგორებიცაა: filter, map, limit, reduce, find, match და ა.შ. 
• Pipelining - უმრავლესობა Stream-ის ოპერაციებისა, აბრუნებს ისევ Stream-ს, ასე რომ ისინი შეიძლება გამოყენებული იყოს როგორც pipeline. ასეთ ოპერაციებს შუალედური ოპერაციები ეწოდება და მათი ფუნქციაა მიიღოს input, დაამუშავოს და დააბრუნოს output თავის ადრესატთან, target-თან. collect() მეთოდი არის ტერმინალური ოპერაცია რომელიც როგორც წესი pipelining-ის ბოლოში გამოიყენება და აღნიშნავს stream-ის დასრულებას. 
• Automatic iterations - Stream ოპერაციები იტერაციას აკეთებს ავტომატურად source ელემენტებზე, რაც კოლექციების შემთხვევაში შეუძლებელი იყო.

Stream-ის მიღება ჯავაში ორი მეთოდით შეიძლება, ესენია : Stream კლასის სტატიკური მეთოდებით, ან Collection Framework-ში განსაზღვრული მეთოდებით.

ახლა განვიხილოთ Stream-ებზე, რამდენიმე მაგალითი და აგრეგაციული ფუნქცია:

forEach

Stream გვთავაზობს forEach მეთოდს, რომელიც მისი თითოეული ელემენტის გადავლას ახორციელებს.  forEach მეთოდს არგუმენტად გადაეცემა ბრძანება, რომელიც გვეუბნება რა უნდა მოხდეს თითოეული ელემენტისთვის. 

map

map მეთოდის გამოიყენება თითოეული ელემენტის შესაბამის რეზულტატთან დასაკავშირებლად (დასა-map-ად). კოდის ფრაგმენტი, რომელიც ქვემოთ არის მოყვანილი ბეჭდავს უნიკალურ კვადრატებს რიცხვებისა. map პარამეტრად იღებს Producer-ის ტიპს

filter

filter მეთოდი აკეთებს ელემენტების გამოხშირვას stream-დან გაკრვეული კრიტერიუმის მიხედვით.  ქვემოთ მოცემული ფრაგმენტი დათვლის ცარიელი სტრიქონების რაოდენობას. filter პარამეტრად იღებს პრედიკატს. 

parallel processing

ქვემოთ მოყვანილი კოდის ფრაგმენტი დაითვლის ცარიელი სტრიქონების რაოდენობას ოღონდ parallelStream-ის მეშვეობით, რომელიც არის stream-ის ალტერნატივა პარალელურ პროცესინგში.

Collectors

კოლექტორები გამოიყენება რეზულტატების კომბინირებისთვის/შეერთებისთვის. კოლექტორები გამოიყენება List-ის ან String-ის დასაბრუნებლად.