क्विक स्टार्ट: गो + अपाचे काफ्का + रेडिस

हाल ही में, आवश्यकता के कारण, मैंने गो डेवलपर्स के लिए सभी रिक्तियों के माध्यम से देखा, और उनमें से आधे (कम से कम) में अपाचे काफ्का संदेश प्रसंस्करण मंच और रेडिस नोएसक्यूएल डेटाबेस का उल्लेख है। खैर, हर कोई, बेशक, उम्मीदवार को डॉकटर और उसके जैसे अन्य लोगों को जानना चाहता है। हमें इन सभी आवश्यकताओं को, जिन्होंने सिस्टम इंजीनियरों के विचारों को देखा है, किसी न किसी तरह से क्षुद्र या कुछ और लगते हैं। खैर, वास्तव में, एक पंक्ति दूसरे से कैसे भिन्न होती है? NoSQL डेटाबेस के साथ स्थिति, निश्चित रूप से, अधिक विविध है, लेकिन फिर भी वे किसी भी MS SQL सर्वर की तुलना में सरल लगते हैं। यह सब, निश्चित रूप से, मेरा व्यक्तिगत, धूर्त - क्रूगर प्रभाव है , जिसका कई बार हैबे पर उल्लेख किया गया है।
इसलिए, चूंकि सभी नियोक्ता मांग करते हैं, इसलिए इन तकनीकों का अध्ययन करना आवश्यक है। लेकिन शुरुआत से अंत तक सभी दस्तावेज पढ़ने के साथ शुरू करना बहुत दिलचस्प नहीं है। मेरी राय में, यह परिचय को पढ़ने, कार्यशील प्रोटोटाइप बनाने, त्रुटियों को ठीक करने, समस्याओं में चलाने, उन्हें हल करने के लिए अधिक उत्पादक है। और इस सब के बाद, समझ के साथ, दस्तावेज़ीकरण, या यहां तक ​​कि एक अलग पुस्तक पढ़ें।



जो लोग इन उत्पादों की बुनियादी क्षमताओं से परिचित होने के लिए थोड़े समय में रुचि रखते हैं, कृपया पर पढ़ें।

प्रशिक्षण कार्यक्रम संख्या में कारक होगा। इसमें एक बड़ी संख्या जनरेटर, एक नंबर प्रोसेसर, एक कतार, स्तंभ भंडारण और एक वेब सर्वर शामिल होगा।

विकास के दौरान, निम्नलिखित डिज़ाइन पैटर्न लागू किए जाएंगे:

• कन्वेयर ( पाइप लाइन )
• स्प्रेयर ( प्रशंसक से बाहर )
• कलेक्टर ( प्रशंसक )

सिस्टम आर्किटेक्चर इस तरह दिखेगा:



तस्वीर में, अंडाकार कन्वेयर डिजाइन पैटर्न को इंगित करता है। मैं इस पर अधिक विस्तार से ध्यान दूंगा।

टेम्पलेट "कन्वेयर" मानता है कि जानकारी एक धारा के रूप में आती है और चरणों में संसाधित होती है। आमतौर पर कुछ जनरेटर (सूचना का स्रोत) और एक या अधिक प्रोसेसर (सूचना प्रोसेसर) होता है। इस मामले में, जनरेटर गो पर एक प्रोग्राम होगा जो यादृच्छिक बड़ी संख्याओं को कतार में रखता है। और प्रोसेसर (केवल एक) एक प्रोग्राम होगा जो कतार से डेटा लेता है और फैक्टराइजेशन करता है। शुद्ध गो पर, यह पैटर्न चैनलों (चान) का उपयोग करके लागू करना बहुत आसान है। ऊपर एक उदाहरण के साथ मेरे जीथब का लिंक है। यहां, संदेश कतार चैनलों की भूमिका निभाएगी।

फैन-इन - फैन-आउट टेम्प्लेट आमतौर पर एक साथ उपयोग किए जाते हैं और, जैसा कि गो पर लागू होता है, का अर्थ है गोरोइन का उपयोग करके गणना का समानांतरण, इसके बाद परिणामों का सारांश और उन्हें स्थानांतरित करना, उदाहरण के लिए, पाइपलाइन नीचे। एक उदाहरण के लिए एक लिंक भी ऊपर दिया गया है। फिर से, चैनल को कतार से बदल दिया गया, गोरोनेट्स जगह पर बने रहे।

अब अपाचे काफ्का के बारे में कुछ शब्द। कफ़्का एक संदेश प्रबंधन प्रणाली है जिसमें उत्कृष्ट क्लस्टरिंग टूल हैं, संदेशों को संग्रहीत करने के लिए एक लेनदेन लॉग (बिल्कुल RDBMS की तरह) का उपयोग करता है, और कतार मॉडल और प्रकाशक / ग्राहक मॉडल दोनों का समर्थन करता है। संदेश प्राप्तकर्ताओं के समूहों के माध्यम से उत्तरार्द्ध हासिल किया जाता है। प्रत्येक संदेश समूह का केवल एक सदस्य (समानांतर प्रसंस्करण) प्राप्त करता है, लेकिन संदेश प्रत्येक समूह में एक बार वितरित किया जाएगा। ऐसे कई समूह हो सकते हैं, साथ ही साथ प्रत्येक समूह में प्राप्तकर्ता भी हो सकते हैं।

काफ्का के साथ काम करने के लिए मैं पैकेज "github.com/segmentio/kafka-go" का उपयोग करूंगा।
दूसरी ओर, Redis, मेमोरी में एक कुंजी-मूल्य स्तंभ डेटाबेस है जो डेटा को स्थायी रूप से संग्रहीत करने की क्षमता का समर्थन करता है। कुंजियों और मूल्यों के लिए मुख्य डेटा प्रकार तार है, लेकिन कुछ अन्य हैं। रेडिस को अपनी कक्षा में सबसे तेज (या सबसे) डेटाबेस में से एक माना जाता है। सभी प्रकार के आँकड़ों, मीट्रिक, संदेश प्रवाह आदि को संग्रहीत करना अच्छा है।
रेडिस के साथ काम करने के लिए मैं "github.com/go-redis/redis" पैकेज का उपयोग करूंगा।

चूँकि यह लेख एक त्वरित शुरुआत है, इसलिए हम DockerHub से तैयार चित्रों का उपयोग करके दोनों सिस्टमों को तैनात करेंगे। मैं इस तरह के docker-compose.yml फ़ाइल के साथ लिनक्स वीएम पर कंटेनर मोड में विंडोज 10 पर डॉकटर-कंपोज (स्वचालित रूप से डॉकर वीएम द्वारा बनाया गया) का उपयोग करता हूं:

version: '2' services: zookeeper: image: wurstmeister/zookeeper ports: - "2181:2181" kafka: image: wurstmeister/kafka:latest ports: - "9092:9092" environment: KAFKA_ADVERTISED_HOST_NAME: 127.0.0.1 KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 KAFKA_CREATE_TOPICS: "Generated:1:1,Solved:1:1,Unsolved:1:1" KAFKA_DELETE_TOPIC_ENABLE: "true" volumes: - /var/run/docker.sock:/var/run/docker.sock redis: image: redis ports: - "6379:6379" 

इस फ़ाइल को सहेजें, इसके साथ डायरेक्टरी पर जाएं और निष्पादित करें:

 docker-compose up -d 

तीन कंटेनरों को डाउनलोड और शुरू करना चाहिए: काफ्का (कतार), ज़ुकीपर (काफ्का के लिए कॉन्फ़िगरेशन सर्वर) और (रेडिस)।

आप यह सत्यापित कर सकते हैं कि कंटेनर कमांड का उपयोग करके काम करते हैं:

 docker-compose ps 

यह कुछ इस तरह होना चाहिए:

 Name State Ports -------------------------------------------------------------------------------------- docker-compose_kafka_1 Up 0.0.0.0:9092->9092/tcp docker-compose_redis_1 Up 0.0.0.0:6379->6379/tcp docker-compose_zookeeper_1 Up 0.0.0.0:2181->2181/tcp, 22/tcp, 2888/tcp, 3888/tcp 

Yml-file के अनुसार, तीन कतारें स्वचालित रूप से बनाई जानी चाहिए, आप उन्हें कमांड से देख सकते हैं:

 docker exec kafka-container_kafka_1 /opt/kafka_2.12-2.1.0/bin/kafka-topics.sh --list --zookeeper zookeeper:2181 

इसमें कतारें (विषय - काफ्का के संदर्भ में विषय) उत्पन्न, हल और अनसुलझी होनी चाहिए।

रैंडम डिले के साथ डेटा जेनरेटर असीम रूप से कतार में खड़ा हो जाता है। इसका कोड बेहद सरल है। आप आदेश का उपयोग करके उत्पन्न कतार में संदेशों की उपस्थिति को सत्यापित कर सकते हैं:

 docker exec kafka-container_kafka_1 /opt/kafka_2.12-2.1.0/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic Generated --from-beginning 

अगला प्रोसेसर है - यहां आपको निम्नलिखित कोड ब्लॉक में कतार से मूल्यों के प्रसंस्करण को समानांतर करने पर ध्यान देना चाहिए:

  var wg sync.WaitGroup c := 0 //counter for { //       15     ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() //      //    -     m, err := r.ReadMessage(ctx) if err != nil { fmt.Println("3") fmt.Println(err) break } wg.Add(1) //       10      goCtx, goCcancel := context.WithTimeout(context.Background(), 10*time.Millisecond) defer goCcancel() //     () go process(goCtx, c, &wg, m) c++ } //     wg.Wait() 

संदेश कतार से पढ़ने के बाद से कार्यक्रम अवरुद्ध हो जाता है, मैंने एक संदर्भ बनाया। 15 सेकंड के टाइमआउट के साथ कॉनटेक्स्ट ऑब्जेक्ट। यदि यह कतार लंबे समय तक खाली है, तो यह समय समाप्त हो जाएगा।

साथ ही, संख्या को फैक्टर करने वाले प्रत्येक गोरुटिन के लिए, अधिकतम ऑपरेटिंग समय भी निर्धारित किया जाता है। मैं उन संख्याओं को चाहता था जो एक डेटाबेस में लिखे जाने में सक्षम थे। और जो संख्याएं आवंटित समय में फैक्ट नहीं की जा सकती थीं, उन्हें दूसरे डेटाबेस में स्थानांतरित कर दिया गया।

अनुमानित समय निर्धारित करने के लिए, बेंचमार्क का उपयोग किया गया था:

 func BenchmarkFactorize(b *testing.B) { ch := make(chan []int) var factors []int for i := 1; i < bN; i++ { num := 2345678901234 go factorize(num, ch) factors = <-ch b.Logf("\n%d   %+v\n\n", num, factors) } } 

गो में बेंचमार्क परीक्षण की किस्में हैं और उन्हें परीक्षण के साथ फाइल में रखा गया है। इस माप के आधार पर, यादृच्छिक संख्या जनरेटर के लिए अधिकतम संख्या का चयन किया गया था। मेरे कंप्यूटर पर, संख्याओं के भाग को कारक बनाने का समय था, और भाग - नहीं।

जिन संख्याओं को विघटित किया जा सकता है, उन्हें DB नंबर 1 में लिखा गया है, जो असंबद्ध संख्याएँ DB नंबर 1 में हैं।
यहाँ मुझे कहना होगा कि रेडिस में शास्त्रीय अर्थों में तालिकाओं और तालिकाओं नहीं हैं। डिफ़ॉल्ट रूप से, DBMS में प्रोग्रामर के पास 16 डेटाबेस उपलब्ध हैं। ये आधार उनकी संख्या में भिन्न होते हैं - 0 से 15 तक।

प्रोसेसर में गोरआउट के लिए समय सीमा संदर्भ और चयन कथन का उपयोग करके प्रदान की गई थी:

  //   go factorize(n, outChan) var item data select { case factors = <-outChan: { fmt.Printf("\ngoroutine #%d, input: %d, factors: %+v\n", counter, n, factors) item.Number = n item.Factors = factors err = storeSolved(item) if err != nil { fmt.Println("6") log.Fatal(err) } } case <-ctx.Done(): { fmt.Printf("\ngoroutine #%d, input: %d, exited on context timeout\n", counter, n) err = storeUnsolved(n) if err != nil { fmt.Println("7") log.Fatal(err) } return nil } } 

यह गो पर ठेठ विकास की चालों में से एक है। इसका अर्थ यह है कि चयन विवरण चैनलों पर प्रसारित होता है और पहले सक्रिय चैनल के अनुरूप कोड निष्पादित करता है। इस स्थिति में, या तो गोरआउट अपने चैनल को परिणाम देगा, या टाइमआउट के साथ संदर्भ चैनल बंद हो जाएगा। संदर्भ के बजाय, आप एक मनमाने चैनल का उपयोग कर सकते हैं जो प्रबंधक के रूप में कार्य करेगा और गोरआउट्स की मजबूर समाप्ति प्रदान करेगा।

डेटाबेस में लिखने के लिए सबरूटीन वांछित डेटाबेस (0 या 1) का चयन करने के लिए कमांड निष्पादित करते हैं और अनकम्पोज्ड संख्याओं के लिए पार्स किए गए नंबरों या (नंबर - नंबर) के लिए फॉर्म (संख्या - कारक) के जोड़े लिखते हैं।

 func storeSolved(item data) (err error) { //    0 cmd := redis.NewStringCmd("select", 0) err = client.Process(cmd) b, err := json.Marshal(item.Factors) err = client.Set(strconv.Itoa(item.Number), string(b), 0).Err() return err } 

अंतिम भाग एक वेब सर्वर होगा, जो कि जसन के रूप में विघटित और अनिर्दिष्ट संख्याओं की सूची प्रदर्शित करेगा। उसके पास दो अंतिम बिंदु होंगे:

  http.HandleFunc("/solved", solvedHandler) http.HandleFunc("/unsolved", unsolvedHandler) 

Redis से डेटा प्राप्त करने और json के रूप में इसे वापस करने के साथ http अनुरोध हैंडलर इस तरह दिखता है:

 func solvedHandler(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") w.Header().Set("Access-Control-Allow-Origin", "*") w.Header().Set("Access-Control-Allow-Methods", "GET") w.Header().Set("Access-Control-Allow-Headers", "Accept, Content-Type, Content-Length, Accept-Encoding, X-CSRF-Token, Authorization") //   №0 -   cmd := redis.NewStringCmd("select", 0) err := client.Process(cmd) if err != nil { w.WriteHeader(http.StatusInternalServerError) return } //      keys := client.Keys("*") var solved []data var item data //          for _, key := range keys.Val() { item.Key = key val, err := client.Get(key).Result() if err != nil { w.WriteHeader(http.StatusInternalServerError) return } item.Val = val solved = append(solved, item) } //    JSON err = json.NewEncoder(w).Encode(solved) if err != nil { w.WriteHeader(http.StatusInternalServerError) return } } 

अनुरोध का परिणाम: लोकलहोस्ट / हल

 [{ "Key": "1604388558816", "Val": "[1,2,3,227]" }, { "Key": "545232916387", "Val": "[1,545232916387]" }, { "Key": "1786301239076", "Val": "[1,2]" }, { "Key": "698495534061", "Val": "[1,3,13,641,165331]" }] 

अब आप प्रलेखन और विशेष साहित्य में तल्लीन कर सकते हैं। आशा है कि लेख उपयोगी था।

मैं विशेषज्ञों से कहता हूं कि वे बहुत आलसी न हों और अपनी गलतियों को इंगित करें।