Java SQL Resultado para InputStream

Question

Necesito una función Java que devuelva los resultados de una consulta SQL SELECT como un parámetro InputStream para otro sistema que envía el resultado a través de una red.

Sin embargo, el InputStream debe ser de String con delimitadores personalizados (es decir, a menudo, pero no siempre, CSV).

Mientras que puedo crear fácilmente una función para recuperar el resultado, crear un delimitado String, y finalmente convertir ese String a un InputStream, el resultado de SQL menudo será demasiado grande para procesar en la memoria. Además, el procesamiento de todo el conjunto de resultados antes de devolver el resultado incurrirá en un tiempo de espera no deseado.

¿Cómo puedo devolver un InputStream para iterar sobre el resultado de SQL y enviar los datos procesados ​​(delimitados) a medida que se devuelven desde la base de datos?

Answer 1

Publicación (no probado) fragmento de código, que debe dar idea básica:

/** 
* Implementors of this interface should only convert current row to byte array and return it. 
* 
* @author yura 
*/ 
public interface RowToByteArrayConverter { 
    byte[] rowToByteArray(ResultSet resultSet); 
} 

public class ResultSetAsInputStream extends InputStream { 

    private final RowToByteArrayConverter converter; 
    private final PreparedStatement statement; 
    private final ResultSet resultSet; 

    private byte[] buffer; 
    private int position; 

    public ResultSetAsInputStream(final RowToByteArrayConverter converter, final Connection connection, final String sql, final Object... parameters) throws SQLException { 
     this.converter = converter; 
     statement = createStatement(connection, sql, parameters); 
     resultSet = statement.executeQuery(); 
    } 

    private static PreparedStatement createStatement(final Connection connection, final String sql, final Object[] parameters) { 
     // PreparedStatement should be created here from passed connection, sql and parameters 
     return null; 
    } 

    @Override 
    public int read() throws IOException { 
     try { 
      if(buffer == null) { 
       // first call of read method 
       if(!resultSet.next()) { 
        return -1; // no rows - empty input stream 
       } else { 
        buffer = converter.rowToByteArray(resultSet); 
        position = 0; 
        return buffer[position++] & (0xff); 
       } 
      } else { 
       // not first call of read method 
       if(position < buffer.length) { 
        // buffer already has some data in, which hasn't been read yet - returning it 
        return buffer[position++] & (0xff); 
       } else { 
        // all data from buffer was read - checking whether there is next row and re-filling buffer 
        if(!resultSet.next()) { 
         return -1; // the buffer was read to the end and there is no rows - end of input stream 
        } else { 
         // there is next row - converting it to byte array and re-filling buffer 
         buffer = converter.rowToByteArray(resultSet); 
         position = 0; 
         return buffer[position++] & (0xff); 
        } 
       } 
      } 
     } catch(final SQLException ex) { 
      throw new IOException(ex); 
     } 
    } 



    @Override 
    public void close() throws IOException { 
     try { 
      statement.close(); 
     } catch(final SQLException ex) { 
      throw new IOException(ex); 
     } 
    } 
} 

Ésta es la aplicación muy sencillo y se puede mejorar en las formas siguientes:

• código la duplicación entre if y else en el método de lectura se puede eliminar - se publicó solo para aclarar
• en lugar de volver a crear el búfer de la matriz de bytes para cada fila (new byte[] es ópera costosa ción), se puede implementar una lógica más sofisticada para usar la memoria intermedia de matriz de bytes que se inicializa solo una vez y luego se vuelve a llenar. Entonces uno debe cambiar la firma del método RowToByteArrayConverter.rowToByteArray a int fillByteArrayFromRow(ResultSet rs, byte[] array), que debe devolver el número de bytes rellenos y llenar el conjunto de bytes pasados.

Debido matriz de bytes contiene bytes firmado puede contener -1 (que en realidad es 255 byte como sin signo) y por lo tanto indicar el final incorrecto de la corriente, por lo & (0xff) se utiliza para convertir byte firmado a bytes sin signo como valores enteros. Para más detalles, consulte How does Java convert int into byte?.

Tenga en cuenta también que si la velocidad de transferencia de la red es lenta, esto puede mantener abiertos los conjuntos de resultados para durante un tiempo prolongado, lo que plantea problemas para la base de datos.

Espero que esto ayude ...

Answer 2

que podrían mejorar la respuesta sugerida por @Yura, introduciendo la siguiente:
Uso DataOutputStream que se inicializa con un ByteArrayOutputStream el fin de escribir los datos convenientemente a la matriz de bytes, dentro de una implementación de RowToByteArrayConverter.
De hecho, sugeriría tener una jerarquía de convertidores, todos ellos se extienden la misma clase abstracta (este es un fragmento de código de mi idea - no podría compilar desde primera vez)

public abstract class RowToByteArrayConverter { 
    public byte[] rowToByteArray(ResultSet resultSet) { 
     parseResultSet(dataOutputStream, resultSet); 
     return byteArrayOutputSteam.toByteArray(); 
    } 

    public RowToByteArrayConverter() { 
    dataOutputStream = new DataOutputStream(byteArrayOutputStream); 
    } 

    protected DataOutputStream dataOutputStream; 
    protected ByteArrayOutputStream byteArrayOutputStream; 

    protected abstract void parseResultSet(DataOutputStream dataOutputStresm, ResultSet rs); 
} 

Ahora, puede anular esta clase simplemente anulando el método parseResultSet,
, por ejemplo, escribe el código que obtiene como cadena un nombre de una columna "nombre" en el registro. y realiza writeUTF8 en DataOputputStream.

Answer 3

Las respuestas anteriores proporcionan una solución útil al problema de que se excede un stringbuilder de tamaño limitado.También son eficientes en la memoria. Sin embargo, mis pruebas sugiere que son más lentos que acaba de escribir datos en un StringBuilder, y llamando

nueva ByteArrayInputStream (data.getBytes ("UTF-8"))

para obtener un InputStream.

Lo que me pareció ser mucho más performante es para rebanar los datos entrantes mediante el uso de una función de partición y luego utilizando varios subprocesos para cada uno:

1. consulta la base de datos de origen para un subconjunto de los datos
2. Escriba los datos en el destino

Esto también evita el problema por el que los datos totales pueden exceder el tamaño máximo de un búfer de cadenas.

Por ejemplo, tengo 6m registros con una columna llamada "RecordDate" en una tabla de SQL Server. Los valores en Fecha de registro varían entre 2013 y 2016. Por lo tanto, configuro cada hilo para cada solicitud, los datos para 2013,14,15,16 respectivamente. Luego, cada subproceso escribe los datos transcodificados en un StringBuilder y cada carga masiva en el destino mediante la conversión a un Inputstream utilizando getBytes() como se indicó anteriormente.

Esto dio como resultado una velocidad de 2x.

¿Por qué? Debido a que las bases de datos de origen y de destino pueden manejar múltiples solicitudes simultáneas, por lo que la carga de trabajo general se distribuye en varios subprocesos en los tres procesos: base de datos de origen, transcodificador, base de datos de destino.