How to implement a callback/producer/consumers scheme?

Question

I'm warming up with Clojure and started to write a few simple functions.

I'm realizing how the language is clearly well-suited for parallel computation and this got me thinking. I've got an app (written in Java but whatever) that works the following way:

one thread waits for input to come in (filesystem in my case but it could be network or whatever) and puts that input once it arrives on a queue
several consumers fetch data from that queue and process the data in parallel

The code that puts the input to be parallely processed may look like this (it's just an example):

asynchFetchInput( new MyCallBack() {
    public void handle( Input input ) {
        queue.put(input)
    }
})

Where asynchFetchInput would spawn a Thread and then call the callback.

It's really just an example but if someone could explain how to do something similar using Clojure it would greatly help me understand the "bigger picture".

@wieczo: not yet... I had already read quite some and watched a few videos, but now from seeing your link I realize this may be one of the "missing piece" in my puzzle : ) Thanks a lot for that link — NoozNooz42, Jan 14 '12 at 22:36
You're welcome! [Refs](http://clojure.org/Refs) might help you with the queue. — Wieczo, Jan 14 '12 at 23:32
Perhaps these old questions might be helpful: [Clojure agents consuming from a queue](http://stackoverflow.com/questions/2602791/clojure-agents-consuming-from-a-queue), [Producer consumer with qualifications](http://stackoverflow.com/questions/2760017/producer-consumer-with-qualifications). — Michał Marczyk, Jan 16 '12 at 01:21

score 3 · Answer 1 · answered Jan 14 '12 at 15:52

If you have to transform data, you can make it into a seq, then you can feed it to either map or pmap The latter will process it in parallel. filter and reduce are also both really useful; so you might want to see if you can express your logic in those terms.

You might also want to look into the concurrency utilities in basic Java rather than spawning your own threads.

How to implement a callback/producer/consumers scheme?

1 Answers1