ØMQ और JZMQ का उपयोग करके फ़ाइलों को जावा में स्थानांतरित करना

अभिवादन, हम समान विचारधारा वाले लोगों की एक छोटी सी कंपनी है जो डेटा का प्रबंधन करने के लिए डिज़ाइन किए गए उत्पाद को विकसित करते हैं, भले ही उनके प्रारूप और भंडारण पद्धति की परवाह किए बिना - ArkStore , हमारे ब्लॉग में हम उस अनुभव को साझा करने का प्रयास करेंगे जो हमने इसके लगभग दो-वर्षीय विकास के दौरान प्राप्त किया था। मैंने एक IO परत और (MQ (या ZeroMQ) नामक उत्पाद के साथ पहले लेख को उजागर करने का निर्णय लिया। मैं यह बताने की कोशिश करूंगा कि tryMQ का उपयोग कैसे शुरू करें और इसका उपयोग डेटा की एक बड़ी मात्रा को स्थानांतरित करने के लिए कैसे किया जा सकता है।

क्यों QMQ

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

अंत में, हमने संदेश दलालों की ओर देखने का फैसला किया। ActiveMQ , RabbitMQ और QMQ । सिद्धांत रूप में, सभी दलालों ने हमारे लिए निर्धारित कार्य को हल किया, लेकिन हम .MQ पर बस गए। ActiveMQ बहुत भारी लग रहा था, और RabbitMQ ने मास्टर नोड को शुरू में वितरित वास्तुकला (एक स्पष्ट नेता के बिना) में पेश किया।

ØMQ तीन बुनियादी डेटा अंतरण पैटर्न का समर्थन करता है, ये हैं:

• अनुरोध-उत्तर सबसे सरल सामान्य टेम्पलेट है, हमने सर्वर को एक अनुरोध भेजा और एक प्रतिक्रिया प्राप्त की। क्लासिक क्लाइंट-सर्वर मॉडल।
• प्रकाशन-सदस्यता - इस मामले में, सर्वर (प्रकाशक) समय-समय पर अपने ग्राहकों के लिए सूचना प्रकाशित करता है। एक उदाहरण एक सेंसर है जो सभी इच्छुक ग्राहकों के लिए लगातार माप डेटा प्रकाशित करता है।
• पुश-पुल - या समानांतर पाइपलाइन , हमें समानांतर कंप्यूटिंग करने की अनुमति देता है। सर्वर समान रूप से काम (पुश) वाले संदेश भेजता है, मशीनों के लिए जो गणना (कार्यकर्ता) कर सकते हैं, मशीन इन संदेशों (पुल) को उठाते हैं, गणना करते हैं और ग्राहक (पुश) को परिणाम देते हैं, जो गणना (पुल) के परिणामों में रुचि रखते हैं। चित्र

आप आधिकारिक वेबसाइट पर उपलब्ध अद्भुत गाइड से इन टेम्पलेट्स और theseMQ के बारे में अधिक जान सकते हैं।

इस लेख में, हम केवल पहले टेम्पलेट (अनुरोध-उत्तर) पर विचार करेंगे।

हम ØMQ और JZMQ एकत्र करते हैं

कोड में आने से पहले, हमें codeMQ और Java बाइंडिंग बनाने की आवश्यकता है।

लिनक्स

सेंटोस के लिए, प्रक्रिया इस तरह दिखाई देगी (यह अन्य * निक्स ओएस के लिए थोड़ा अलग होना चाहिए)।

सुनिश्चित करें कि हमारे पास सभी आवश्यक पुस्तकालय हैं:

yum install libtool autoconf automake gcc-c++ make e2fsprogs 

हम डेवलपर की साइट से .MQ लाइब्रेरी के नवीनतम स्थिर संस्करण को ले लेंगे और अनज़िप करेंगे (3.2.4 लिखने के समय)।

 wget http://download.zeromq.org/zeromq-3.2.4.tar.gz tar -xzvf zeromq-3.2.4.tar.gz cd zeromq-3.2.4 

हम QMQ एकत्र करते हैं और स्थापित करते हैं, लाइब्रेरीज़ / usr / लोकल / लेबर डायरेक्टरी में होंगे, हमें भविष्य में इसकी आवश्यकता होगी।

 ./configure make sudo make install sudo ldconfig 

हम JMQ इकट्ठे होने के बाद हमें JZMQ को इकट्ठा करने की जरूरत है। ऐसा करने के लिए, हम जीआईटी रिपॉजिटरी (मास्टर या टैग, लेखन के समय अंतिम टैग 2.2.2) से नवीनतम संस्करण को पंप करते हैं।

 wget https://github.com/zeromq/jzmq/archive/v2.2.2.zip unzip jzmq-2.2.2.zip cd jzmq-2.2.2 ./autogen.sh ./configure make sudo make install sudo ldconfig 

पुस्तकालय भी / usr / lib / स्थानीय निर्देशिका में होंगे। यह महत्वपूर्ण क्यों है? तथ्य यह है कि देशी पुस्तकालयों का उपयोग करने के लिए, जावा को पता होना चाहिए कि हम उन्हें कहां पा सकते हैं, इसके लिए, कार्यक्रम शुरू करते समय, हमें java.library.path पैरामीटर निर्दिष्ट करना होगा। ऐसा करने के कई तरीके हैं, हम इसे एप्लिकेशन -Djava.library.path = "/ usr / lib / स्थानीय" लॉन्च करने के समय निर्दिष्ट कर सकते हैं, या प्रोग्राम चलने के दौरान इसे सीधे इंस्टॉल कर सकते हैं। हम java.library.path मान का उपयोग भी कर सकते हैं, जो डिफ़ॉल्ट रूप से सेट किया गया है। डिफ़ॉल्ट रूप से क्या मान निर्धारित किए गए हैं, यह जानने के लिए, आपको निम्नलिखित कमांड चलाने की आवश्यकता है:

 java -XshowSettings:properties 

मेरे मामले में, यह है:

 java.library.path = /usr/java/packages/lib/amd64 /usr/lib64 /lib64 /lib /usr/lib 

जावा के लिए देशी पुस्तकालयों को खोजने में सक्षम होने के लिए, यह हमारे पुस्तकालयों को इनमें से किसी एक पते पर स्थानांतरित करने या उन्हें लिंक करने के लिए पर्याप्त है। चुनने के लिए कौन सा दृष्टिकोण, आप व्यक्तिगत रूप से चुनते हैं।

यह पता लगाने के लिए कि लाइब्रेरियों को स्थापित करने के बाद कहां स्थापित किया गया था, बस कमांड चलाएं:

 whereis libzmq whereis libjzmq 

विंडोज

Libzmq के लिए एकत्रित dll फाइलें आधिकारिक साइट से डाउनलोड की जा सकती हैं, और यहां आप विंडोज के लिए JZMQ को संकलित करने के लिए एक गाइड पा सकते हैं। दुर्भाग्य से, मैंने CMAKE का उपयोग करके पुस्तकालयों का निर्माण करने का प्रबंधन नहीं किया, मुझे Visual Studio 2013 का उपयोग करके libzmq और jzmq का निर्माण करना पड़ा। साथ ही, यह महत्वपूर्ण है कि आपके स्वयं के VVM (32 या 64 बिट) के लिए उपयुक्त आर्किटेक्चर के लिए पुस्तकालय स्वयं निर्मित हों,

यदि libzmq.dll और jzmq.dll को PATH में जोड़ा जाता है, तो JVM को उन्हें स्वचालित रूप से ढूंढना चाहिए।

कार्यक्रम

वाह, हम अंत में हमारे कंप्यूटर पर andMQ और JZMQ स्थापित और कॉन्फ़िगर करने में सक्षम थे! इसे अमल में लाने का समय आ गया है। एक उदाहरण के रूप में, हम मैनुअल में वर्णित फ़ाइल स्थानांतरण प्रोटोकॉल का उपयोग करने की कोशिश करेंगे और इसे थोड़ा सुधारेंगे।

सबसे पहले, हम अपने प्रोटोकॉल के लिए आवश्यकताओं का वर्णन करते हैं:

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

"बेतरतीब ढंग से" ™ डेटा की एक गीगाबाइट तैयार करें।

 dd if=/dev/urandom of=testdata bs=1M count=1024 

आइए मापें कि ओएस को डेटा की प्रतिलिपि बनाने में कितना समय लगता है। ये आंकड़े काफी अनुमानित हैं, लेकिन कम से कम हमारे पास किसी तरह का तुलनात्मक बिंदु होगा।

 echo 3 > /proc/sys/vm/drop_caches time cp testdata testdata2 real 0m7.745s user 0m0.011s sys 0m1.223s 

कोड के लिए नीचे आते हैं। ग्राहक:

 package com.coldsnipe.example; import org.zeromq.ZFrame; import org.zeromq.ZMQ; public class ZMQClient implements Runnable { //      "" private static final int PIPELINE = 10; //  1  , 250. private static final int CHUNK_SIZE = 250000; private final ZMQ.Context zmqContext; public ZMQClient(ZMQ.Context zmqContext) { this.zmqContext = zmqContext; } @Override public void run() { //  ZMQ.Socket,     DEALER socket try (ZMQ.Socket socket = zmqContext.socket(ZMQ.DEALER)) { //            //  ,   1      (drop)   socket.setLinger(1000); //     TCP  socket.connect("tcp://127.0.0.1:6000"); //  ,  Chunks     int credit = PIPELINE; //    long total = 0; //    (Chunks) int chunks = 0; //      long offset = 0; while (!Thread.currentThread().isInterrupted()) { //     "",     //   while (credit > 0) { socket.sendMore("fetch"); socket.sendMore(Long.toString(offset)); socket.send(Integer.toString(CHUNK_SIZE)); offset += CHUNK_SIZE; credit--; } //       ZFrame zFrame = ZFrame.recvFrame(socket); //   ,   if (zFrame == null) { break; } chunks++; credit++; int size = zFrame.getData().length; total += size; zFrame.destroy(); //         //       if (size < CHUNK_SIZE) { break; } } System.out.format("%d chunks received, %d bytes %n", chunks, total); } } } 

क्लाइंट सर्वर से जुड़ता है और डेटा के लिए अनुरोध भेजता है। प्रत्येक संदेश में कई भाग होते हैं।

1. कमांड जो हम सर्वर से चाहते हैं, इस मामले में केवल एक कमांड है, जिसमें डेटा प्राप्त करना है।
2. कमांड पैरामीटर (यदि मौजूद है) - भ्रूण के मामले में , यह फ़ाइल की शुरुआत और डेटा के टुकड़े के आकार से इंडेंट है।

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

 package com.coldsnipe.example; import org.zeromq.ZFrame; import org.zeromq.ZMQ; import java.io.File; import java.io.IOException; import java.nio.ByteBuffer; import java.nio.channels.FileChannel; import java.nio.file.Path; public class ZMQServer implements Runnable { private final ZMQ.Context zmqContext; //         private final Path filePath; public ZMQServer(ZMQ.Context zmqContext, Path filePath) { this.zmqContext = zmqContext; this.filePath = filePath; } @Override public void run() { try { File file = filePath.toFile(); if (!file.exists()) { throw new RuntimeException("File does not exists: " + filePath); } //    , Router   identity , //         try (FileChannel fileChannel = FileChannel.open(filePath)) { try (ZMQ.Socket socket = zmqContext.socket(ZMQ.ROUTER)) { //      localhost  6000  socket.bind("tcp://*:6000"); socket.setLinger(1000); while (!Thread.currentThread().isInterrupted()) { //   -  identity frame ZFrame identity = ZFrame.recvFrame(socket); assert identity != null; //    String command = socket.recvStr(); if (command.equals("fetch")) { //     fetch,   offset     String offsetString = socket.recvStr(); long offset = Long.parseLong(offsetString); String chunkSizeString = socket.recvStr(); int chunkSize = Integer.parseInt(chunkSizeString); int currentChunkSize = chunkSize; //   offset +       //     ,      offset    if (file.length() < (offset + chunkSize)) { currentChunkSize = (int) (file.length() - offset); } if (currentChunkSize > 0) { ByteBuffer byteBuffer = ByteBuffer.allocate(currentChunkSize); fileChannel.read(byteBuffer, offset); byteBuffer.flip(); byte[] bytes = new byte[currentChunkSize]; byteBuffer.get(bytes); //     ZFrame frameToSend = new ZFrame(bytes); //        identity // ,     identity.send(socket, ZFrame.MORE); frameToSend.send(socket, 0); } } } } } } catch (IOException e) { throw new RuntimeException(e); } } } 

हमारा सर्वर केवल एक फ़ाइल (लिंक जो इसे स्टार्टअप पर प्राप्त हुआ है) को स्थानांतरित कर सकता है और केवल एक ही कमांड का जवाब दे सकता है - प्राप्त करें । वह जानता है कि ग्राहकों को कैसे अलग करना है, लेकिन ग्राहक केवल एक ही फाइल प्राप्त कर सकते हैं। इसे कैसे सुधारें मैं थोड़ा कम लिखूंगा, लेकिन अब परीक्षण और माप के परिणाम के लिए।

 package com.coldsnipe.example; import org.junit.Test; import org.zeromq.ZMQ; import java.net.URL; import java.nio.file.Path; import java.nio.file.Paths; public class ZMQExampleTest { @Test public void testDataExchange() throws Exception { //  ZMQ.Context,    1!      //          ZMQ.Context zmqContext = ZMQ.context(1); //     final URL fileUrl = ZMQExampleTest.class.getClassLoader().getResource("testdata"); assert fileUrl != null; Path filePath = Paths.get(fileUrl.toURI()); //        ,     //       long startTime = System.nanoTime(); Thread clientThread = new Thread(new ZMQClient(zmqContext)); Thread serverThread = new Thread(new ZMQServer(zmqContext, filePath)); clientThread.start(); serverThread.start(); clientThread.join(); long estimatedTime = System.nanoTime() - startTime; float timeForTest = estimatedTime / 1000000000F; System.out.format("Elapsed time: %fs%n", timeForTest); //    ,     //          . zmqContext.term(); } } 

हमारा परीक्षण ZMQ.Context को लॉन्च करता है, क्लाइंट और सर्वर को शुरू करता है, और डेटा ट्रांसफर करने में लगने वाले समय को मापता है। मैं संदर्भ के बारे में अलग से कहना चाहता हूं। यह संदर्भ है कि छिपा कंडक्टर है जो प्रक्रिया के भीतर हमारी सॉकेट्स को नियंत्रित करता है और यह तय करता है कि डेटा कैसे और कब भेजा जाए। इसलिए, एक सरल नियम यहां से है - प्रति प्रक्रिया एक संदर्भ ।

परीक्षण चलाएं और परिणाम देखें।

 4295 chunks received, 1073741824 bytes Elapsed time: 1.429522s 

एक गीगाबाइट पढ़ने में 1.42 सेकंड लगे। मेरे लिए यह कहना मुश्किल है कि यह संकेतक कितना अच्छा है, लेकिन समान स्प्रे IO की तुलना में, QMQ 30-40% तेजी से पूरा होता है, जबकि IO पर लोड 100% (स्प्रे 85-90) के करीब है, CPU लोड लगभग एक तिहाई कम है ।

प्रोटोकॉल में सुधार

जबकि हमारा प्रोटोकॉल केवल एक कमांड जानता है, यह परीक्षण के लिए पर्याप्त है, लेकिन वास्तविक परिस्थितियों में हम चाहते हैं कि हमारा सर्वर कई अलग-अलग फ़ाइलों को स्थानांतरित करने और क्लाइंट के लिए सेवा की जानकारी प्रदान करने में सक्षम हो। ऐसा करने के लिए, हम दो नए कमांड पेश करते हैं:

• get - संदेश जो क्लाइंट डेटा प्राप्त करना चाहता है, उसमें वह डेटा होना चाहिए जो हम प्राप्त करना चाहते हैं
• अंत - क्लाइंट द्वारा डेटा प्राप्त करने के बाद भेजा गया एक संदेश, यह दर्शाता है कि सर्वर हस्तांतरित फ़ाइल से जुड़े संसाधनों को मुक्त कर सकता है।

इस मामले में संदेश हैंडलर इस तरह लग सकते हैं

 else if(command.equals("get")) { // ID  String id = socket.recvStr(); //       BigInteger identityString = new BigInteger(identity.getData()); DataBlob dataBlob = blobMap.get(identityString); if (dataBlob!= null){ //      ,   DataBlob dataBlob.closeSource(); blobMap.remove(identityString); } //     DataBlob dataBlob = dataProvider.getBlob(id); if (dataBlob == null) { log.error("Received wrong get call on server socket, ID [{}]", id); } else { //   ,        DataHeader dataHeader = new DataHeader(id, dataBlob.getSize()); byte[] bytesToSend = FrameHeaderEncoder.encode(dataHeader); ZFrame frameToSend = new ZFrame(bytesToSend); //   DataBlob   blobMap.put(new BigInteger(identity.getData()), dataBlob); //    identity.send(socket, ZFrame.MORE); frameToSend.send(socket, 0); } } else if (command.equals("end")){ BigInteger identityString = new BigInteger(identity.getData()); //  DataBlob    DataBlob dataBlob = blobMap.remove(identityString); if (dataBlob != null) { dataBlob.closeSource(); } } 

कई फ़ाइलों के साथ काम करने के लिए, हमने सर्वर में नई वस्तुओं को जोड़ा।

 public interface DataProvider { public DataBlob getBlob(String dataId); } public abstract class DataBlob { private final long size; private final String dataId; protected DataBlob(long size, String dataId) { this.size = size; this.dataId= dataId; } public abstract byte[] getData(long position, int length); public abstract void closeSource(); public long getSize() { return size; } public String getDataId() { return getDataId; } } 

DataProvider को लागू करने वाला वर्ग डेटा की प्राप्ति का प्रबंधन करता है, GetBlob विधि हमें एक नया DataBlob देता है जो अनिवार्य रूप से संसाधन का एक संदर्भ है।

फ़ाइल के लिए DataBlob कार्यान्वयन इस तरह दिखाई दे सकता है:

 public class FileDataBlob extends DataBlob { private final FileChannel fileChannel; public FileDataBlob(long size, String dataId, Path filePath) { super(size, dataId); try { this.fileChannel = FileChannel.open(filePath); } catch (IOException e) { throw new DataBlobException(e); } } @Override public byte[] getData(long position, int length) { ByteBuffer byteBuffer = ByteBuffer.allocate(length); try { fileChannel.read(byteBuffer, position); byteBuffer.flip(); byte[] bytes = new byte[length]; byteBuffer.get(bytes); return bytes; } catch (IOException e) { throw new DataBlobException(e); } } @Override public void closeSource() { try { fileChannel.close(); } catch (IOException e) { throw new DataBlobException(e); } } } 

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

निष्कर्ष

इस लेख में, मैं यह दिखाना चाहता था कि हम जावा में inMQ का उपयोग कैसे कर सकते हैं। फिलहाल, हमारी परियोजना में, यह momentMQ है जो मुख्य संदेश ब्रोकर है, न केवल फाइलों के लिए बल्कि मेटाडेटा के लिए भी, बल्कि अच्छे परिणाम दिखा रहा है (इसके साथ कोई समस्या अभी तक नहीं देखी गई है)।

निम्नलिखित लेखों में मैं अर्कस्टोर में इस्तेमाल की जाने वाली अन्य तकनीकों के बारे में बात करने की कोशिश करूंगा , जबकि अक्का और सिमेंटिक वेब बारी-बारी से हैं। आपका ध्यान के लिए धन्यवाद, मुझे उम्मीद है कि पढ़ा कम से कम किसी के लिए उपयोगी था!