Creating and joining threads in clojure

create-join-threads.clj

(import '[java.util.concurrent Callable
                               CountDownLatch
                               Executors
                               Semaphore
                               TimeUnit])

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Examples of creating and joining threads in clojure
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(let [numthreads 5
      waiting-msg "Waiting on threads to finish..."
      done-msg    "All threads joined. Exiting."
      func #(do (Thread/sleep (+ 1000 (rand-int 4000)))
                (println (.getName (Thread/currentThread)) "finishing")
                (Thread/sleep 20))]


  (println "\n\n******* 1 - Clojure futures, and deref to join")
  (let [func #(future (func))
        threads (repeatedly numthreads func)]

    (dorun threads)      ; create and run threads

    (println waiting-msg)

    (run! deref threads) ; (doall (map deref threads)) to get values
    
    (println done-msg))



  (println "\n\n******* 2a - Java ExecutorService, invokeAll")
  (let [tp (Executors/newCachedThreadPool)
        threads (repeat numthreads func)]

    (println waiting-msg)
    
    (.invokeAll tp threads)
    
    (println done-msg))



  (println "\n\n******* 2b - Java ExecutorService, awaitTermination")
  (let [tp (Executors/newCachedThreadPool)]

    (dotimes [_ numthreads] (.submit tp ^Callable func))
    
    (println waiting-msg)
    
    (.shutdown tp)
    (.awaitTermination tp Long/MAX_VALUE TimeUnit/DAYS)
    
    (println done-msg))



  (println "\n\n******* 3 - Semaphore")
  (let [sema (Semaphore. 0)
        func #(future (try (func) (finally (.release sema))))]

    (dotimes [_ numthreads] (func))
    
    (println waiting-msg)
    
    (.acquire sema numthreads)
    
    (println done-msg))



  (println "\n\n******* 4 - CoundDownLatch")
  (let [latch (CountDownLatch. numthreads)
        func #(future (try (func) (finally (.countDown latch))))]

    (dotimes [_ numthreads] (func))
    
    (println waiting-msg)
    
    (.await latch)
    
    (println done-msg))



  (println "\n\n******* 5 - Manual thread creation and join")
  (let [threads (repeatedly numthreads #(Thread. func))]

    (run! #(.start %) threads)
    
    (println waiting-msg)
    
    (run! #(.join %) threads)
    
    (println "All threads joined. Exiting.")))