دفق البيانات من خدمة REST إلى قائمة انتظار MQ

مرحبا يا هبر!

في المقالة ، سوف أصف طريقة لتطوير خدمة REST التي تسمح لك بتلقي الملفات وحفظها في نظام المراسلة في وضع الدفق دون الحاجة إلى تخزين الملف بأكمله على جانب الخدمة. كما سيتم وصف السيناريو العكسي الذي سيتلقى فيه العميل كرد ملف موجود في نظام المراسلة.

للتوضيح ، سأعطي أمثلة لرمز الخدمة المطور على JEE7 لخادم تطبيق IBM WebSphere Liberty Server ، وسيعمل IBM MQ كنظام مراسلة.
ومع ذلك ، فإن الطريقة الموصوفة مناسبة لمنصات أخرى مماثلة ، أي يمكن لأي مزود JMS API أن يعمل كنظام مراسلة ، وأي خادم JEE (على سبيل المثال ، Apache Tomcat) يمكنه العمل كخادم تطبيق.

بيان المشكلة

كانت هناك حاجة لتطبيق حل يسمح باستقبال الملفات الكبيرة (> 100 ميجا بايت) من العميل ونقلها إلى نظام بعيد جغرافيًا آخر ، وفي الاتجاه المعاكس - نقل الملفات من هذا النظام إلى العميل كإجابة. في ضوء قناة الشبكة غير الموثوقة بين شبكة العميل وشبكة التطبيق ، يتم استخدام نظام مراسلة يضمن التسليم المضمون بينهما.

يتضمن حل المستوى الأعلى ثلاثة مكونات:

1. خدمة REST - تتمثل مهمتها في تزويد العميل بفرصة نقل الملف (أو الطلب).
2. MQ - مسؤول عن نقل الرسائل بين الشبكات المختلفة.
3. التطبيق - تطبيق مسؤول عن تخزين الملفات وإصدارها عند الطلب.

في هذه المقالة ، أصف طريقة لتنفيذ خدمة REST ، تتضمن مهامها:

• تلقي ملف من العميل.
• انقل الملف المستلم إلى MQ.
• نقل ملف من MQ إلى العميل كرد.

طريقة الحل

نظرًا للحجم الكبير للملف المرسل ، لا يمكن وضعه بالكامل في ذاكرة الوصول العشوائي ، علاوة على ذلك ، هناك أيضًا قيود على جانب MQ - لا يمكن أن يتجاوز الحجم الأقصى لرسالة واحدة في MQ 100 ميجا بايت. وبالتالي ، سوف يستند قراري إلى المبادئ التالية:

• يجب أن يتم استلام ملف وحفظه في قائمة انتظار MQ في وضع الدفق ، دون وضع الملف بأكمله في الذاكرة.
• في قائمة الانتظار ، سيتم وضع ملف MQ كمجموعة من الرسائل الصغيرة.

بيانياً ، يتم عرض تخصيص الملفات من جانب العميل وخدمة REST و MQ أدناه:


من جانب العميل ، يقع الملف بالكامل على نظام الملفات ، في خدمة REST ، يتم تخزين جزء فقط من الملف في ذاكرة الوصول العشوائي ، وعلى جانب MQ ، يتم وضع كل جزء من الملف كرسالة منفصلة.

تطوير خدمة REST

لتوضيح طريقة الحل المقترحة ، سيتم تطوير خدمة REST تجريبية تحتوي على طريقتين:

• تحميل - يتلقى ملفًا من العميل ويكتبه في قائمة انتظار MQ ، ويعيد معرف مجموعة الرسائل (بتنسيق base64) كرد.
• تنزيل - يستقبل معرف مجموعة الرسائل (بتنسيق base64) من العميل ويعيد الملف المخزن في قائمة انتظار MQ.

طريقة استلام ملف من عميل (تحميل)

تتمثل مهمة الطريقة في الحصول على دفق الملف الوارد ثم كتابته إلى قائمة انتظار MQ.

استقبال دفق الملف الوارد

لتلقي ملف إدخال من العميل ، تتوقع الطريقة وجود كائن بواجهة 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 Liberty Server ويقع في المجلد: < WLP_INSTALLATION_PATH > / dev / api / ibm.

ملحوظة:

• WLP_INSTALLATION_PATH - المسار الى دليل ملف مواصفات WebSphere Liberty.
• من المتوقع أن يقوم العميل بنقل الملف في المعلمة المسماة "file".
• إذا كنت تستخدم خادم تطبيقات مختلفًا ، فيمكنك استخدام المكتبة البديلة من Apache CXF.

دفق حفظ ملف في MQ

تستقبل الطريقة دفق ملف الإدخال ، واسم قائمة انتظار MQ حيث يجب كتابة الملف ، ومعرف مجموعة الرسائل التي سيتم استخدامها لربط الرسائل. يتم إنشاء معرف المجموعة على جانب الخدمة ، على سبيل المثال ، باستخدام الأداة المساعدة org.apache.commons.lang3.RandomStringUtils:

 String groupId = RandomStringUtils.randomAscii(24); 

تتكون خوارزمية حفظ ملف الإدخال في MQ من الخطوات التالية:

1. تهيئة كائنات اتصال MQ.
2. قراءة دورية لجزء من الملف الوارد حتى تتم قراءة الملف بالكامل:
   1. يتم تسجيل جزء من بيانات الملف كرسالة منفصلة في MQ.
   2. كل رسالة في الملف لها الرقم التسلسلي الخاص بها (الخاصية "JMSXGroupSeq").
   3. جميع الرسائل في الملف لها نفس قيمة المجموعة (الخاصية "JMSXGroupID").
   4. تحتوي الرسالة الأخيرة على علامة تشير إلى أن هذه الرسالة نهائية (الخاصية "JMS_IBM_Last_Msg_In_Group").
   5. يحتوي ثابت SEGMENT_SIZE على حجم الحصة. على سبيل المثال ، 1 ميجا بايت.

 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)); } } } } } 

طريقة إرسال ملف إلى العميل (تنزيل)

تحصل الطريقة على معرف مجموعة من الرسائل بتنسيق base64 ، والتي تقوم من خلالها بقراءة الرسائل من قائمة انتظار MQ وإرسالها كرد في وضع الدفق.

الحصول على معرف مجموعة الرسائل

تستقبل الطريقة معرف مجموعة الرسائل كمعلمة إدخال.

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

تدفق رد على العميل

لنقل ملف إلى العميل ، تم تخزينه كمجموعة من الرسائل المنفصلة في MQ ، في وضع الدفق ، قم بإنشاء فئة باستخدام واجهة 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(); } } } 

في الفصل ، نقوم بتطبيق طريقة الكتابة ، التي تتلقى مرجع إدخال إلى الدفق الصادر الذي ستتم كتابة رسائل MQ فيه. أضفت أيضًا اسم قائمة الانتظار ومعرف المجموعة التي سيتم قراءة رسائلها إلى الفصل.

سيتم تمرير كائن من هذه الفئة كمعلمة لإنشاء استجابة للعميل:

 @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 إلى الدفق الصادر من الخطوات التالية:

1. تهيئة كائنات اتصال MQ.
2. القراءة الدورية للرسائل من 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; } } } } } } 

مكالمة خدمة REST

لاختبار الخدمة ، سأستخدم أداة curl.

تحميل ملف

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

ستكون الاستجابة عبارة عن سلسلة base64 تحتوي على معرف مجموعة الرسائل ، والذي سنشير إليه في الطريقة التالية للحصول على الملف.

استلام ملف

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

الخلاصة

فحصت المقالة نهج تطوير خدمة REST ، والتي تسمح بالدفق لتلقي وحفظ البيانات الكبيرة في قائمة انتظار نظام المراسلة ، وكذلك قراءتها من قائمة الانتظار للعودة كرد. تقلل هذه الطريقة من استخدام الموارد ، وبالتالي تزيد من إنتاجية الحل.

مواد إضافية

مزيد من المعلومات حول واجهة IMultipartBody المستخدمة لتلقي دفق الملفات الواردة هو ارتباط .

مكتبة بديلة لاستقبال الملفات في وضع الدفق في خدمات REST هي Apache CXF .

تعد واجهة StreamingOutput لتدفق استجابة REST للعميل ارتباطًا .