In order to implement service which limits of request count per time quantity I look for implementation of queue that reject adding elements if enqueue()/add() command exceeded some frequency rate?
So for example you can call enqueue()/add() no more that 1000 times per second and prevent high resource consuming service DDOS.
Also queue must reject enqueue()/add() if queue capacity were exceeded (until you call dequeue()/remove()).
One simple way to implement this would be to check the following whenever a new enqueue / add operation is requested:
- Check whether the queue is full
- Check whether the time since last enqueue operation is > 1/1000 seconds. 2a) You can actually modify point 2 above by keeping a time stamp of 1000 latest enqueue operations. Whenever an enqueue operation is requested, check the time since the first operation that was made in this queue. If it is more than 1 second, then pop out that operation from the time stamp queue and add this one.
keeping a time stamp of 1000 latest enqueue operations what if this count too large? iterate over 1e+6 elements to keep quota precision - too cost operation...
Do you want to reject with an Exception, or do you just want to ignore them.
A better approach might be to update a Map and ignore duplicate updates to a given key. This drops events you couldn't process but ensuring the latest update of a key is always taken.
This way you consume N new message per second and can gracefully drop duplicates.
e.g.
final ConcurrentMap<String, Message> map = new ConcurrentHashMap<>(); final BlockingQueue<String> queue = new ArrayBlockingQueue<>(10000); int updateCount = 0; long lastUpdate = 0; public void enqueue(String key, Message message) { map.put(key, message); queue.offer(key); } public Message dequeue() throws InterruptedException { long updateTime = System.currentTimeMillis(); long time = updateTime - lastUpdate; lastUpdate = updateTime; if (time > count) count = 0; else count -= time; if (count > 1000) Thread.sleep(1); count++; while(true) { String key = queue.take(); Message msg = map.remove(key); if (msg != null) return msg; // if the msg is null, we have already sent the latest update for that key } } This will give you an update rate of 1000 per second and always give you the latest update for any key.
message which I process but what is key? How I generate keys from (currentTime(), message) pair?