Databases
StackOverflow2013
Postsdbo
POClojure agents consuming from a queue
 

Note that there are some explanatory texts on larger screens.

plurals
  1. POClojure agents consuming from a queue
    primarykey
    data
    text
    <p>I'm trying to figure out the best way to use agents to consume items from a Message Queue (Amazon SQS). Right now I have a function (process-queue-item) that grabs an items from the queue, and processes it.</p> <p>I want to process these items concurrently, but I can't wrap my head around how to control the agents. Basically I want to keep all of the agents busy as much as possible without pulling to many items from the Queue and developing a backlog (I'll have this running on a couple of machines, so items need to be left in the queue until they are really needed).</p> <p>Can anyone give me some pointers on improving my implementation?</p> <pre><code>(def active-agents (ref 0)) (defn process-queue-item [_] (dosync (alter active-agents inc)) ;retrieve item from Message Queue (Amazon SQS) and process (dosync (alter active-agents dec))) (defn -main [] (def agents (for [x (range 20)] (agent x))) (loop [loop-count 0] (if (&lt; @active-agents 20) (doseq [agent agents] (if (agent-errors agent) (clear-agent-errors agent)) ;should skip this agent until later if it is still busy processing (not sure how) (send-off agent process-queue-item))) ;(apply await-for (* 10 1000) agents) (Thread/sleep 10000) (logging/info (str "ACTIVE AGENTS " @active-agents)) (if (&gt; 10 loop-count) (do (logging/info (str "done, let's cleanup " count)) (doseq [agent agents] (if (agent-errors agent) (clear-agent-errors agent))) (apply await agents) (shutdown-agents)) (recur (inc count))))) </code></pre>
    singulars
    1. This table or related slice is empty.
    1. This table or related slice is empty.
    1. PTQuestion
    1. This table or related slice is empty.
    1. USerikcw
    plurals
    1. PL
      singulars
      1. LTLinked
    1. PL
      singulars
      1. LTLinked
    2. PL
      singulars
      1. LTLinked
    1. This table or related slice is empty.
    1. PO
      singulars
      1. PTAnswer
    2. PO
      singulars
      1. PTAnswer
    3. PO
      singulars
      1. PTAnswer
    1. VO
      singulars
      1. POClojure agents consuming from a queue
      1. This table or related slice is empty.
      1. VTUpMod
    2. VO
      singulars
      1. POClojure agents consuming from a queue
      1. This table or related slice is empty.
      1. VTUpMod
    3. VO
      singulars
      1. POClojure agents consuming from a queue
      1. This table or related slice is empty.
      1. VTUpMod
    1. COIs there some way you could treat the Message Queue as a seq and then just use pmap to get the parallelization?
      singulars
      1. POClojure agents consuming from a queue
      1. USAlex Stoddard
    2. CO@Alex Stoddard: In my case, process-queue-item actually blocks on network IO, so I don't think pmap is the right choice since it only uses as many threads as the machine has cores.
      singulars
      1. POClojure agents consuming from a queue
      1. USerikcw
    3. CO@erikw: Sure, but that is just a pmap implementation detail (threads = #cores + 2). No reason you couldn't write a version of pmap with a parameterized number of threads. See the first line of the pmap source: (let [n (+ 2 (.. Runtime getRuntime availableProcessors))
      singulars
      1. POClojure agents consuming from a queue
      1. USAlex Stoddard
 

Querying!

 
Guidance
A row detail

Detail views are divided into sections. All the information in the data section comes from columns in the selected row. The other sections display data from other, related rows.

Related data can be related in a to-one or a to-many fashion. Captions of data related in a to-many fashion link to a list view showing a filtered view of the table.

Try moving around until you find a non-empty to-many entry and click on the label to get to one. You can move back to the root by clicking on the database name in the header.

SQuiL has stopped working due to an internal error.

If you are curious you may find further information in the browser console, which is accessible through the devtools (F12).

Reload