🦀 👊🏻 👩🏿‍🚀 एक RQ सेवा से डेटा को MQ कतार में स्ट्रीम करना ☀️ 🖼️ 👨🏿‍💼

नमस्कार, हेब्र!

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

स्पष्टता के लिए, मैं IBM WebSphere Liberty Server एप्लिकेशन सर्वर के लिए JEE7 पर विकसित सेवा कोड का उदाहरण दूंगा, और IBM MQ एक संदेश प्रणाली के रूप में कार्य करेगा।
हालांकि, वर्णित विधि अन्य समान प्लेटफार्मों के लिए उपयुक्त है, अर्थात। कोई भी JMS API प्रदाता एक संदेश प्रणाली के रूप में कार्य कर सकता है, और कोई JEE सर्वर (उदाहरण के लिए, Apache Tomcat) एक एप्लिकेशन सर्वर के रूप में कार्य कर सकता है।

समस्या का बयान

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

शीर्ष-स्तरीय समाधान में तीन घटक शामिल हैं:

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

इस लेख में, मैं REST सेवा को लागू करने के लिए एक विधि का वर्णन करता हूं, जिसके कार्यों में शामिल हैं:

एक ग्राहक से एक फ़ाइल प्राप्त करना।
प्राप्त फ़ाइल को MQ पर स्थानांतरित करें।
प्रतिक्रिया के रूप में MQ से क्लाइंट के लिए फ़ाइल स्थानांतरित करना।

समाधान विधि

प्रेषित फ़ाइल के बड़े आकार के कारण, इसे पूरी तरह से रैम में रखना संभव नहीं है, इसके अलावा, एमक्यू पक्ष पर भी प्रतिबंध है - एमक्यू में एक संदेश का अधिकतम आकार 100 एमबी से अधिक नहीं हो सकता है। इस प्रकार, मेरा निर्णय निम्नलिखित सिद्धांतों पर आधारित होगा:

एक फ़ाइल प्राप्त करना और इसे एमक्यू कतार में सहेजना, संपूर्ण फ़ाइल को मेमोरी में रखे बिना, स्ट्रीमिंग मोड में किया जाना चाहिए।
कतार में, MQ फ़ाइल को छोटे संदेशों के सेट के रूप में रखा जाएगा।

रेखांकन, क्लाइंट की ओर से फ़ाइल आवंटन, REST सेवा और MQ नीचे दिखाया गया है:

क्लाइंट पक्ष पर, फ़ाइल पूरी तरह से फ़ाइल सिस्टम पर स्थित है, REST सेवा में, फ़ाइल का केवल एक भाग रैम में संग्रहीत होता है, और एमक्यू पक्ष पर, फ़ाइल के प्रत्येक भाग को एक अलग संदेश के रूप में रखा जाता है।

एक आरईएस सेवा का विकास

प्रस्तावित समाधान पद्धति की स्पष्टता के लिए, डेमो रीस्ट सेवा को दो तरीकों से विकसित किया जाएगा:

अपलोड - क्लाइंट से एक फ़ाइल प्राप्त करता है और इसे MQ कतार में लिखता है, एक प्रतिक्रिया के रूप में संदेश समूह पहचानकर्ता (बेस 64 प्रारूप में) देता है।
डाउनलोड - ग्राहक से संदेश समूह पहचानकर्ता (बेस 64 प्रारूप में) प्राप्त करता है और एमक्यू कतार में संग्रहीत फ़ाइल लौटाता है।

क्लाइंट से फ़ाइल प्राप्त करने की विधि (अपलोड)

विधि का कार्य आने वाली फ़ाइल की धारा प्राप्त करना है और फिर इसे एमक्यू कतार में लिखना है।

आने वाली फ़ाइल को पुनः प्राप्त करना

क्लाइंट से एक इनपुट फ़ाइल प्राप्त करने के लिए, विधि इंटरफ़ेस com.ibm.websphere.jaxrs20.multipart.IMultipartBody इनपुट पैरामीटर के रूप में एक ऑब्जेक्ट की उम्मीद करती है, जो आने वाली फ़ाइल की धारा के लिए एक लिंक प्राप्त करने की क्षमता प्रदान करती है

@PUT @Path("upload") public Response upload(IMultipartBody body) { ... IAttachment attachment = body.getAttachment("file"); InputStream inputStream = attachment.getDataHandler().getInputStream(); ... }

यह इंटरफ़ेस (IMultipartBody) com.ibm.websphere.appserver.api.jaxrs20_1.0.21.jar JAR संग्रह में स्थित है, IBM लिबर्टी सर्वर के साथ शामिल है और फ़ोल्डर में स्थित है: < WLP_INSTALLATION_PATH > / dev / api / ibm।

नोट:

WLP_INSTALLATION_PATH - वेबसर्फ़ लिबर्टी प्रोफ़ाइल निर्देशिका का पथ।
यह उम्मीद की जाती है कि क्लाइंट फ़ाइल को "फ़ाइल" नाम के पैरामीटर में स्थानांतरित करेगा।
यदि आप एक अलग एप्लिकेशन सर्वर का उपयोग कर रहे हैं, तो आप Apache CXF से वैकल्पिक लाइब्रेरी का उपयोग कर सकते हैं।

MQ में किसी फाइल को सेव करने की स्ट्रीम

विधि इनपुट फ़ाइल स्ट्रीम, MQ कतार का नाम जहां फ़ाइल लिखी जानी चाहिए, और संदेश समूह की पहचानकर्ता जिसका उपयोग संदेशों को बांधने के लिए किया जाएगा। समूह पहचानकर्ता सेवा पक्ष पर उत्पन्न होता है, उदाहरण के लिए, उपयोगिता org.apache.commons.lang3.RandomStringUtils के साथ:

 String groupId = RandomStringUtils.randomAscii(24);

MQ में इनपुट फ़ाइल को बचाने के लिए एल्गोरिथ्म में निम्नलिखित चरण हैं:

एमक्यू कनेक्शन वस्तुओं का प्रारंभिककरण।
फ़ाइल के पूरी तरह से पढ़ने तक आने वाली फ़ाइल के एक हिस्से का चक्रीय पढ़ना:
1. फ़ाइल डेटा का एक हिस्सा MQ में एक अलग संदेश के रूप में दर्ज किया गया है।
2. फ़ाइल के प्रत्येक संदेश का अपना सीरियल नंबर (संपत्ति "JMSXGroupSeq") है।
3. फ़ाइल के सभी संदेशों का समूह मान समान है (संपत्ति "JMSXGroupID")।
4. अंतिम संदेश में संकेत है कि यह संदेश अंतिम है (संपत्ति "JMS_IBM_Last_Msg_In_Group")।
5. SEGMENT_SIZE स्थिरांक में सेवारत आकार होता है। उदाहरण के लिए, 1Mb।

 public void write(InputStream inputStream, String queueName, String groupId) throws IOException, JMSException { try ( Connection connection = connectionFactory.createConnection(); Session session = connection.createSession(); MessageProducer producer = session.createProducer(session.createQueue(queueName)); ) { byte[] buffer = new byte[SEGMENT_SIZE]; BytesMessage message = null; for(int readBytesSize = 1, sequenceNumber = 1; readBytesSize > 0; sequenceNumber++) { readBytesSize = inputStream.read(buffer); if (message != null) { if (readBytesSize < 1) { message.setBooleanProperty("JMS_IBM_Last_Msg_In_Group", true); } producer.send(message); } if (readBytesSize > 0) { message = session.createBytesMessage(); message.setStringProperty("JMSXGroupID", groupId); message.setIntProperty("JMSXGroupSeq", sequenceNumber); if (readBytesSize == SEGMENT_SIZE) { message.writeBytes(buffer); } else { message.writeBytes(Arrays.copyOf(buffer, readBytesSize)); } } } } }

ग्राहक को फ़ाइल भेजने की विधि (डाउनलोड)

विधि बेस 64 प्रारूप में संदेशों के एक समूह के पहचानकर्ता को प्राप्त करती है, जिसके द्वारा यह एमक्यू कतार से संदेश पढ़ता है और इसे स्ट्रीमिंग मोड में प्रतिक्रिया के रूप में भेजता है।

संदेश समूह आईडी प्राप्त करना

विधि इनपुट पैरामीटर के रूप में संदेश समूह पहचानकर्ता प्राप्त करता है।

 @PUT @Path("download") public Response download(@QueryParam("groupId") String groupId) { ... }

एक ग्राहक को एक प्रतिक्रिया स्ट्रीमिंग

एमक्यू में अलग-अलग संदेशों के एक सेट के रूप में, क्लाइंट को एक फाइल ट्रांसफर करने के लिए, स्ट्रीमिंग मोड में, javax.ws.rs.core.StreamingOutput इंटरफ़ेस के साथ एक क्लास बनाएं:

 public class MQStreamingOutput implements StreamingOutput { private String groupId; private String queueName; public MQStreamingOutput(String groupId, String queueName) { super(); this.groupId = groupId; this.queueName = queueName; } @Override public void write(OutputStream outputStream) throws IOException, WebApplicationException { try { new MQWorker().read(outputStream, queueName, groupId); } catch(NamingException | JMSException e) { e.printStackTrace(); new IOException(e); } finally { outputStream.flush(); outputStream.close(); } } }

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

ग्राहक की प्रतिक्रिया बनाने के लिए इस वर्ग की एक वस्तु को एक पैरामीटर के रूप में पारित किया जाएगा:

 @GET @Path("download") public Response download(@QueryParam("groupId") String groupId) { ResponseBuilder responseBuilder = null; try { MQStreamingOutput streamingOutput = new MQStreamingOutput(new String(Utils.decodeBase64(groupId)), Utils.QUEUE_NAME); responseBuilder = Response.ok(streamingOutput); } catch(Exception e) { e.printStackTrace(); responseBuilder.status(Status.INTERNAL_SERVER_ERROR).entity(e.getMessage()); } return responseBuilder.build(); }

MQ से किसी फाइल को पढ़ने की स्ट्रीम

MQ से आउटगोइंग स्ट्रीम पर संदेश पढ़ने के लिए एल्गोरिथ्म में निम्नलिखित चरण होते हैं:

एमक्यू कनेक्शन वस्तुओं का प्रारंभिककरण।
समूह में समाप्ति के संकेत के साथ एक संदेश पढ़ने तक MQ से संदेशों की चक्रीय रीडिंग पढ़ी जाती है (संपत्ति "JMS_IBM_Last_Msg_In_Group"):
1. प्रत्येक संदेश को कतार से पढ़े जाने से पहले, एक फ़िल्टर (messageSelector) सेट किया जाता है, जिसमें संदेश समूह पहचानकर्ता और समूह में संदेश क्रमांक सेट किए जाते हैं।
2. पठन संदेश की सामग्री आउटगोइंग स्ट्रीम को लिखी जाती है।

 public void read(OutputStream outputStream, String queueName, String groupId) throws IOException, JMSException { try( Connection connection = connectionFactory.createConnection(); Session session = connection.createSession(); ) { connection.start(); Queue queue = session.createQueue(queueName); int sequenceNumber = 1; for(boolean isMessageExist = true; isMessageExist == true; ) { String messageSelector = "JMSXGroupID='" + groupId.replaceAll("'", "''") + "' AND JMSXGroupSeq=" + sequenceNumber++; try( MessageConsumer consumer = session.createConsumer(queue, messageSelector); ) { BytesMessage message = (BytesMessage) consumer.receiveNoWait(); if (message == null) { isMessageExist = false; } else { byte[] buffer = new byte[(int) message.getBodyLength()]; message.readBytes(buffer); outputStream.write(buffer); if (message.getBooleanProperty("JMS_IBM_Last_Msg_In_Group")) { isMessageExist = false; } } } } } }

अन्य सेवा कॉल

सेवा का परीक्षण करने के लिए, मैं कर्ल टूल का उपयोग करूंगा।

फ़ाइल अपलोड करें

 curl -X PUT -F file=@<__> http://localhost:9080/Demo/rest/service/upload

प्रतिक्रिया एक बेस 64 स्ट्रिंग होगी जिसमें संदेश समूह के पहचानकर्ता होंगे, जिसे हम फ़ाइल प्राप्त करने के लिए अगली विधि में इंगित करेंगे।

एक फ़ाइल प्राप्त करना

 curl -X GET http://localhost:9080/Demo/rest/service/download?groupId=<base64____> -o <_____>

निष्कर्ष

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

अतिरिक्त सामग्री

आने वाली फ़ाइल स्ट्रीम प्राप्त करने के लिए उपयोग किए जाने वाले IMultipartBody इंटरफ़ेस के बारे में अधिक जानकारी एक लिंक है ।

REST सेवाओं में स्ट्रीमिंग मोड में फ़ाइलें प्राप्त करने के लिए एक वैकल्पिक पुस्तकालय Apache CXF है ।

किसी क्लाइंट के लिए REST प्रतिक्रिया को स्ट्रीम करने के लिए StreamingOutput इंटरफ़ेस एक लिंक है ।

एक RQ सेवा से डेटा को MQ कतार में स्ट्रीम करना