6

G'day everyone,

I have a system (the source) that needs to notify another system (the target) asynchronously whenever certain objects change. The twist is that the source system may mutate a single object many times in a short interval (updates are very "bursty"), and in that case it would be ideal to only notify the target system once, and with the final state of the object.

My thought was to use some kind of time-delayed, de-duping queue in front of a ThreadPoolExecutor for this. This queue would:

  1. keep items on the queue for a minimum amount of time (ideally configured to be just a smidgin longer than the duration of a typical burst of mutations)

  2. replace the existing object in the event that a duplicate (as defined by the object's identifier) is enqueued. The item should, however, retain its original place in the queue (to avoid any one item being perpetually bumped to the back of the queue - at some point we need to just send the notification even if another one will be coming along momentarily).

I haven't seen anything exactly like this in java.util, and my google-fu in this area appears to be particularly weak.

Has anyone implemented this before, know a BlockingQueue implementation that behaves this way, or have tips on how I might go about implementing one?

Thanks in advance!

Peter

PS. I know ESBs do this kind of thing, but that is too heavyweight an approach in this case - ideally I don't want to add any new library dependencies to the source system at all.

Peter
  • 519
  • 6
  • 15

2 Answers2

4

I think your best bet is to extend ArrayBlockingQueue and override offer and poll to add the time delay functionality. Particularly ArrayBlockingQueue because it has a contains method.

Another idea is a DelayQueue where you override offer to remove the old element and insert the new but preserve the old time delay, which will essentially preserve order. Then you need to wrap your queue items in a Delayed interface.

trutheality
  • 23,114
  • 6
  • 54
  • 68
  • Thanks trutheality. Any thoughts on just implementing `BlockingQueue` directly, but delegating to an internal `ArrayBlockingQueue` instance, with some extra logic in the `offer` and `poll` methods? I always get a bit leery of extending core JDK classes - so many of them weren't really designed with that in mind (particularly in java.util.concurrent, from my limited poking around in there). – Peter May 29 '11 at 06:41
  • Well the difference between extending and implementing+internal in this case would be the difference between using `this`/`super` and `myBlockingQueue` as your queue object, so it's a question of taste IMO. – trutheality May 29 '11 at 06:50
  • Now that I think about it, the `DelayQueue` idea might be easier. I think the trickiest part is going to be updating elements of the queue while they're in the queue, which, if you use an `ArrayBlockingQueue`, would probably involve either going in there with an `iterator` or keeping something like an `ArrayList` of references for easy access to the objects. With `DelayQueue` it's just `remove`, match expiration time, and `offer`. – trutheality May 29 '11 at 06:58
0

You could notify the system only that a change has occurred, and have the other system fetch the new state. Then don't notify of further changes until the state has been fetched.

Lawrence Dol
  • 63,018
  • 25
  • 139
  • 189
  • Thanks Software Monkey, but I think that converts the original problem into a bigger one - i.e. having to keep track of the individual "retrieval states" of potentially millions of objects. – Peter May 29 '11 at 06:44