Geekly articles weekly

Transmisión de datos desde un servicio REST a una cola MQ

Hola Habr!

En el artículo, describiré una forma de desarrollar un servicio REST que le permite recibir archivos y guardarlos en el sistema de mensajería en modo de transmisión sin la necesidad de almacenar el archivo completo en el lado del servicio. También se describirá el escenario inverso en el que el cliente recibirá como respuesta un archivo ubicado en el sistema de mensajería.

Para mayor claridad, daré ejemplos del código de servicio desarrollado en JEE7 para el servidor de aplicaciones IBM WebSphere Liberty Server, e IBM MQ actuará como un sistema de mensajería.
Sin embargo, el método descrito es adecuado para otras plataformas similares, es decir. cualquier proveedor de API JMS puede actuar como un sistema de mensajería, y cualquier servidor JEE (por ejemplo, Apache Tomcat) puede actuar como un servidor de aplicaciones.

Declaración del problema.


Era necesario implementar una solución que permitiera recibir archivos grandes (> 100 Mb) del cliente y transferirlos a otro sistema geográficamente remoto, y en la dirección opuesta: transferir archivos de este sistema al cliente como respuesta. En vista del canal de red poco confiable entre la red del cliente y la red de la aplicación, se utiliza un sistema de mensajería que garantiza la entrega garantizada entre ellos.

La solución de nivel superior incluye tres componentes:

  1. Servicio REST: la tarea consiste en proporcionar al cliente la oportunidad de transferir el archivo (o solicitud).
  2. MQ: es responsable de la transmisión de mensajes entre diferentes redes.
  3. Aplicación: una aplicación responsable de almacenar archivos y emitirlos a pedido.

imagen

En este artículo, describo un método para implementar un servicio REST, cuyas tareas incluyen:

  • Recibir un archivo de un cliente.
  • Transfiera el archivo recibido a MQ.
  • Transferencia de un archivo de MQ al cliente como respuesta.

Método de solución


Debido al gran tamaño del archivo transmitido, no es posible colocarlo completamente en la RAM, además, también hay una restricción en el lado de MQ: el tamaño máximo de un mensaje en MQ no puede exceder los 100 Mb. Por lo tanto, mi decisión se basará en los siguientes principios:

  • Recibir un archivo y guardarlo en la cola MQ debe realizarse en modo de transmisión, sin colocar todo el archivo en la memoria.
  • En la cola, el archivo MQ se colocará como un conjunto de mensajes pequeños.

Gráficamente, la asignación de archivos en el lado del cliente, el servicio REST y MQ se muestra a continuación:

imagen

En el lado del cliente, el archivo está completamente ubicado en el sistema de archivos, en el servicio REST, solo una parte del archivo se almacena en la RAM, y en el lado de MQ, cada parte del archivo se coloca como un mensaje separado.

Desarrollo de un servicio REST.


Para mayor claridad del método de solución propuesto, se desarrollará un servicio REST de demostración que contiene dos métodos:

  • upload: recibe un archivo del cliente y lo escribe en la cola MQ, devuelve el identificador del grupo de mensajes (en formato base64) como respuesta.
  • descargar: recibe el identificador del grupo de mensajes (en formato base64) del cliente y devuelve el archivo almacenado en la cola MQ.

Método para recibir un archivo de un cliente (cargar)


La tarea del método es obtener la secuencia del archivo entrante y luego escribirlo en la cola MQ.

Recibiendo la secuencia del archivo entrante


Para recibir un archivo de entrada del cliente, el método espera un objeto con la interfaz com.ibm.websphere.jaxrs20.multipart.IMultipartBody como parámetro de entrada, que proporciona la capacidad de obtener un enlace a la secuencia del archivo entrante

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

Esta interfaz (IMultipartBody) se encuentra en el archivo JAR com.ibm.websphere.appserver.api.jaxrs20_1.0.21.jar, se incluye con el servidor IBM Liberty y se encuentra en la carpeta: < WLP_INSTALLATION_PATH > / dev / api / ibm.

Nota:

  • WLP_INSTALLATION_PATH : ruta al directorio de WebSphere Liberty Profile.
  • Se espera que el cliente transfiera el archivo en el parámetro llamado "archivo".
  • Si está utilizando un servidor de aplicaciones diferente, puede usar la biblioteca alternativa de Apache CXF.

Stream guardando un archivo en MQ


El método recibe la secuencia del archivo de entrada, el nombre de la cola MQ donde se debe escribir el archivo y el identificador del grupo de mensajes que se utilizará para enlazar mensajes. El identificador de grupo se genera en el lado del servicio, por ejemplo, con la utilidad org.apache.commons.lang3.RandomStringUtils:

 String groupId = RandomStringUtils.randomAscii(24);

El algoritmo para guardar el archivo de entrada en MQ consta de los siguientes pasos:

  1. Inicialización de objetos de conexión MQ.
  2. Lectura cíclica de una parte del archivo entrante hasta que el archivo esté completamente leído:
    1. Un fragmento de datos de archivo se registra como un mensaje separado en MQ.
    2. Cada mensaje en el archivo tiene su propio número de serie (propiedad "JMSXGroupSeq").
    3. Todos los mensajes en el archivo tienen el mismo valor de grupo (propiedad "JMSXGroupID").
    4. El último mensaje tiene un signo que indica que este mensaje es final (propiedad "JMS_IBM_Last_Msg_In_Group").
    5. La constante SEGMENT_SIZE contiene el tamaño de la porción. Por ejemplo, 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)); } } } } }

Método para enviar un archivo al cliente (descargar)


El método obtiene el identificador de un grupo de mensajes en formato base64, mediante el cual lee mensajes de la cola MQ y los envía como respuesta en modo de transmisión.

Obtener la identificación del grupo de mensajes


El método recibe el identificador del grupo de mensajes como un parámetro de entrada.

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

Transmitir una respuesta a un cliente


Para transferir un archivo al cliente, almacenado como un conjunto de mensajes separados en MQ, en modo de transmisión, cree una clase con la interfaz 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(); } } }

En la clase, implementamos el método de escritura, que recibe una referencia de entrada a la secuencia saliente en la que se escribirán los mensajes de MQ. También agregué el nombre de la cola y el identificador del grupo cuyos mensajes serán leídos a la clase.

Se pasará un objeto de esta clase como parámetro para crear una respuesta al cliente:

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

Stream leyendo un archivo de MQ


El algoritmo para leer mensajes de MQ en la secuencia saliente consta de los siguientes pasos:

  1. Inicialización de objetos de conexión MQ.
  2. Lectura cíclica de mensajes de MQ hasta que se lea un mensaje con el signo de terminar en el grupo (propiedad "JMS_IBM_Last_Msg_In_Group"):
    1. Antes de leer cada mensaje de la cola, se establece un filtro (messageSelector), en el que se establece el identificador del grupo de mensajes y el número de serie del mensaje en el grupo.
    2. El contenido del mensaje leído se escribe en la secuencia saliente.


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

Llamada de servicio REST


Para probar el servicio, usaré la herramienta curl.

Carga de archivos


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

La respuesta será una cadena base64 que contiene el identificador del grupo de mensajes, que indicaremos en el siguiente método para obtener el archivo.

Recibiendo un archivo


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

Conclusión


El artículo examinó el enfoque para desarrollar un servicio REST que le permita transmitir y recibir datos de gran tamaño en la cola del sistema de mensajería, así como leerlo desde la cola para devolverlo como respuesta. Este método reduce el uso de recursos y, por lo tanto, aumenta el rendimiento de la solución.

Materiales adicionales


Más información sobre la interfaz IMultipartBody utilizada para recibir la secuencia de archivos entrantes es un enlace .

Una biblioteca alternativa para recibir archivos en modo de transmisión en servicios REST es Apache CXF .

La interfaz StreamingOutput para transmitir una respuesta REST a un cliente es un enlace .

Source: https://habr.com/ru/post/es424941/

More articles:


All Articles