In days gone by businesses proclaimed “we’re gonna do Hadoop Jase!”, no word of a lie, they used to phone me up and tell me so…… my response was fairly standard.
Now the world has gone streaming-based-full-on-kooki-mad because, “Jase we need the real time, we’re all about real time now, our customers demand realtime*”
* They probably don’t but hey….
The Problem With Kafka….
Well, in reality it’s not really a problem you just have to appreciate what Kafka was designed for. Kafka was designed for high volume message processing. It’s the Unix pipe on steroids, more steroids and a few more on top of that. When I say “high volume” I mean Linkedin billions of messages high volume.
And to use Kafka properly you need machines, plural. Three beefy servers for each of the Kafka brokers (you need at least three for leader election if the master broker pops off for some reason), then another three for Zookeeper because the last thing you need on earth is Zookeeper dying on you otherwise everything dies. And really if you’re on pager duty that’s the last thing you need.
Call me psychic, I know exactly what you’re thinking…..
Why not use Kinesis, Jase! Yer big ninny!
Well yeah, but no. There are two reasons, firstly I’m tight and I hate spending money whether it be mine or other people’s. “But we got VC funding so we run $5k/month on servers”, it doesn’t wash with me. You’ll run out of money. My other issue is more technical.
Kinesis is fine for things but my heart just doesn’t settle for the whole five consumers per shard malarky. Kinesis has some nice things but performance and consumer scaling are not two of them.
So, what to do?
Something a little more, manageable oh and smaller.
So I asked this question to the wider community (ie everyone) and I got one response from a Mrs Trellis of North Wales, well actually it was Bruce Durling, suggesting Factual’s Durable Queue. I wasn’t aware of it’s existence….
The disk based bit is important, very important. If the stream dies at any point I need to know that it will pick up messages from the point it died. Kafka does it, RabbitMQ does it and the others do it, so I need this to happen.
Durable Queue uses core.async to do it’s bidding. And it’s easy to put a stream or a number of streams together.
A Quick Demo Project
Add the Dependencies
First of all we need to add a few dependencies to a project.
[com.taoensso/timbre "4.10.0"] [factual/durable-queue "0.1.5"] [org.clojure/core.async "0.4.474"]
Adding the required namespaces to the namespace.
Let’s create a simple queue. Add the required dependencies in the namespace. You’ll need the Durable Queue and Core.Async namespaces, I’ve added the Taoensso Timbre logging as it’s my preferred way of logging.
(:require [durable-queue :refer :all]
[clojure.core.async :refer [go-loop <! timeout]]
[taoensso.timbre :as log])
Create a Disk Based Queue
Our actual queue is easy to create, pick a directory to write your queue to and you’re done. I see this in the same way as a Kafka topic, each file is basically your topic queue.
(def my-test-queue (queues "/tmp/" {}))
Define the Function The Does The Processing of Each Message
Think of this as your streaming api app, where the magic happens so to speak. This is the function that processes every message that’s been taken from the queue.
(defn the-function-that-does-the-work [m] (log/info "Got the message! - " m))
Put a Message On the Queue
We need to be able to add messages to the queue. The key is the key of the queue, further on down when we look at the queue loop itself, the reason why will become clear). The value, well I like passing Clojure maps of information but in reality it can be whatever you wish.
(defn send-something-to-queue [k v] (put! my-test-queue k v))
Taking Messages From the Queue
Now we’ve got something to put messages on the queue, we need something to take them off. Basically this is take the message from the queue with the same key as we passed in with the send-something-to-queue function. The messsage needs dereferencing so we get the true payload.
(defn take-from-queue [k]
(let [msgkey (take! my-test-queue k)]
(do
(the-function-that-does-the-work (deref msgkey))
(complete! msgkey))))
Also worth noting that we have to manually say that we’ve completed the work on that message with the complete! function. If this doesn’t happen then the message is still classed as queued and if you restart the queue (it dies for example) then that same message will get processed again.
The Queue Loop
I’m going to run a core.async go-loop against this function. If there are messages in the queue to be processed then go and get the next one.
(defn run-queue-loop [] (if (not= 0 (:enqueued (stats my-test-queue))) (take-from-queue :id)))
And finally a core async go-loop to keep everything going. A timeout of one second just to keep things sane.
(go-loop []
(do
(run-queue-loop)
(<! (timeout 1000))
(recur)))
Testing the Demo
You could either run this from the REPL (it works a treat) or run it standalone, though I didn’t provide a -main function in the repo, you’re happy to add if you wish. I’ve created two topic queues, q1 and q2, and will send a number of messages (nrange) to them. I’m just doling out randomly decided topic destination at this point. Mainly doing this to prove that multiple topic queues do work.
(defn run-queue-demo [nrange]
(map (fn [i]
(if (= 0 (rand-nth [0 1]))
(q1/send-something-to-queue :id
{:uuid (str (java.util.UUID/randomUUID))})
(q2/send-something-to-queue :id
{:uuid (str (java.util.UUID/randomUUID))}))) (range 1 nrange)))
When I run this in the REPL with 10 messages, I get the following output. You can see messages going to each queue and as we can see the output of the processing function we know the stream is working.
fractal.queue.demo> (run-queue-demo 10)
(true true true true true true true true true)
18-11-05 07:35:12 Jasons-MacBook-Pro.local INFO [fractal.queue.secondqueue:10] - Got the message on second queue! - {:uuid "c55cc0f8-54ff-47ca-81a2-858af68f47b2"}
18-11-05 07:35:12 Jasons-MacBook-Pro.local INFO [fractal.queue.mainqueue:10] - Got the message! - {:uuid "56d83928-0019-4d58-bfc0-af2dbbf625b0"}
18-11-05 07:35:13 Jasons-MacBook-Pro.local INFO [fractal.queue.secondqueue:10] - Got the message on second queue! - {:uuid "875dc578-b32c-4548-ac06-51ab9ef93d41"}
18-11-05 07:35:13 Jasons-MacBook-Pro.local INFO [fractal.queue.mainqueue:10] - Got the message! - {:uuid "0315c4e3-ba6b-4533-ac41-62293038da30"}
18-11-05 07:35:14 Jasons-MacBook-Pro.local INFO [fractal.queue.secondqueue:10] - Got the message on second queue! - {:uuid "f522deb6-968c-44a1-a042-2f312f4d314b"}
18-11-05 07:35:14 Jasons-MacBook-Pro.local INFO [fractal.queue.mainqueue:10] - Got the message! - {:uuid "bc7fc7e5-aaf2-4226-871b-31f31199356c"}
18-11-05 07:35:15 Jasons-MacBook-Pro.local INFO [fractal.queue.secondqueue:10] - Got the message on second queue! - {:uuid "b5a892fd-f516-4133-991b-539f8b8477be"}
18-11-05 07:35:16 Jasons-MacBook-Pro.local INFO [fractal.queue.secondqueue:10] - Got the message on second queue! - {:uuid "6ae167b2-c7a3-42ec-87eb-14b629348a21"}
18-11-05 07:35:17 Jasons-MacBook-Pro.local INFO [fractal.queue.secondqueue:10] - Got the message on second queue! - {:uuid "2f97c442-31ca-495e-b1b6-b7019917d504"}
Another nice feature of the durable queue is that you pull up stats on the queue. The num-slabs is the number of files the queue uses.
fractal.queue.demo> (durable-queue/stats q1/my-test-queue)
{"id" {:num-slabs 1, :num-active-slabs 1, :enqueued 3, :retried 0, :completed 3, :in-progress 0}}
Conclusion
A basic topic based queue system, a bit like Kafka but without all the overheads. From a Clojure perspective it’s a great use of core.async and looking at the throughput figures on Factual’s github page it’s a system I’m happy in using until I’m at a point where I really do need Kafka (and all those machines).
There’s a Github repo of the demo code here.
Update
So, I was half asleep when I wrote this, it’s Factual not Fractal. Thanks to Mrs Trellis for the headsup…..
It's All a Bet 
Leave a comment