Managing Complexity with Concurrent Programming - Clojure Data Analysis - page 94

Managing large inputs with sized queues
When we work with very large datasets, we often talk about structuring our program
concurrently. One big problem when dealing with very large datasets concurrently
is coordinating and managing the low of data between different parts of our program.
If one part produces data too quickly, or another part processes it too slowly (depending
on how you look at it), the message queue between the two can get backed up. If that
happens, the memory will ill up with messages and data waiting to be processed.
How to do it…
The solution for this in Clojure is quite simple: use seque . This uses an instance of
java.util.concurrent.LinkedBlockingQueue to pull values from a lazy sequence.
It works ahead of where we're pulling values out of the queue, but not too far ahead. And
once we've wrapped a sequence with seque , we can treat it just like any other sequence:
user=> (take 20 (seque 5 (range Integer/MAX_VALUE)))
(0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19)
How it works...
The seque function reads ahead a bit (usually a little more than we specify). It then waits
until some of the items that it has read have been consumed, and then it reads ahead a
little more. This ensures that the rest of our system always has input to process, but its
memory won't get illed by input waiting to be processed. This is an easy solution to
balancing input and processing, and this function's simplicity helps keep an often
complex problem from introducing incidental complexity into our processing system.