Cómo leer registros de base de datos como un Stream de Java

He estado usando Spring desde la versión 1.0 y una de las clases que siempre me han resultado extremadamente útiles es JdbcTemplate. Esperaba que para la versión 5 integrara funcionalidad del API de stream de Java, pero no fue así.

Sin embargo a veces necesito realizar búsquedas en base de datos que devuelven miles o incluso millones de registros, y no puedo usar los métodos de JdbcTemplate que devuelven listas porque me quedo sin memoria. Los métodos que usan RowCallbackHandler son más apropiados, pero sería mucho más conveniente poder usar Streams de Java, particularmente si se pueden leer los resultados como objetos usando algún RowMapper.

Así que decidí hacer mi propio generados de Stream para usar con un JdbcTemplate. Al final, terminé con uno que es realmente más genérico y se puede usar con cualquier código que genere elementos ya sea de manera finita o infinita (aunque para streams infinitos existe una API mucho más simple). No es suficiente material como para generar una biblioteca, así que decidí publicarlo como un post.

El reto

Primero que nada, hay que considerar que los streams son lazy, es decir la evaluación se difiere hasta el momento en que realmente se tiene que hacer y aún así se van obteniendo elementos de una fuente conforme se necesiten. Al crear un stream y definir operaciones sobre el mismo, no ocurre actualmente nada, hasta que se realice alguna operación que requiera realmente recorrer los elementos del stream y aplicar las operaciones definidas. Hay operaciones que se aplican a un stream completo (como contar los elementos, o juntarlos en otra colección), y hay operaciones de corto circuito (como determinar si al menos un elemento del stream pasa un filtro).

Así que queremos crear un stream que va a obtener sus elementos de una consulta a base de datos, por lo que se necesita diferir dicha consulta hasta el momento en que realmente se necesita, ya que requiere mantener una conexión abierta.

La única manera en que pude hacer que esto funcionara fue usando dos hilos: un productor en el cual se ejecuta la consulta y los resultados alimentan al stream, y un consumidor donde se leen los elementos del stream.

Necesitamos además un buffer donde el productor pueda poner elementos para que el consumidor los vaya tomando. Una LinkedBlockingQueue es perfecta para esto.

Entonces, el código queda así:

Y así es como ese código se usa (aquí seguimos el ejemplo con JdbcTemplate):

Cuando ese código se ejecuta, el query no se ha realizado. Incluso se pueden hacer cosas como esto:

Y hasta ahí sigue sin pasar nada. Pero cuando se hace algo como esto:

Entonces ya se ejecuta el query, se leen ( y saltan ) los primeros cien mil registros, y luego se pasa cada registro por el filtro, hasta que uno pase el filtro o se hayan leído mil registros.

##Un gran poder conlleva gran responsabilidad

Es muy tentador usar esto para simplemente hacer   y filtrar en memoria. Por favor, por lo que más quieran, no lo hagan. Filtrar en memoria jamás será un reemplazo adecuado para un buen WHERE con índices bien diseñados en sus tablas.

Yo he utilizado esto principalmente para reportes, donde puedo concatenar streams de tipos disjuntos mapeando sus elementos a un tipo común para después continuar su procesamiento. Básicamente, estoy compensando por la falta de tipos unión en Java.

Ahora, habiendo dicho eso, está bastante chingón poder leer registros de una base de datos en modo de streaming. Tal vez un día veamos algo así ya integrado en Spring JdbcTemplate, o en jOOQ o incluso a nivel de JDBC...