[rabbitmq-discuss] Message Aggregating Queue

[Alexis Richardson]

Jason

Interesting stuff :-)

Prior discussions on this subject have tended to the view that the
exchange is the best place to cache a 'last value'.

In your example below, the way to identify that #1 and #3 refer to the
same stock would be to use the stock ticker (eg "UBS") as the routing
key in a direct exchange.

See for example this prototype: https://github.com/squaremo/rabbitmq-lvc-plugin

An aggregation capability is also doable but it's not obvious what the
best way to do it is.  It depends whether aggregate messages are the
same type as their constituents.  If they are not, as in your example,
then you might want to republish a new aggregate message whenever
something changes, and then cache aggregates using the LVC approach.

alexis



On Thu, Apr 28, 2011 at 9:28 AM, Jason Zaugg <jzaugg at gmail.com> wrote:
> Among other things, we're using RabbitMQ to distribute market data
> ticks. The interesting characteristic about this stream of messages is
> that a new tick for a obsoletes previous, unprocessed, ticks for the
> same stock.
>
> After a little brainstorming last night with Alvaro Videla, I'm
> curious to discuss how this could be modeled with an extension to
> RabbitMQ.
>
> It seems a similar problem to the Queue Based TTL [1], however TTL
> isn't quite right for this case, as the validity period of a tick
> depends on the liquidity of the stock. We really just want the
> *latest* tick.
>
> Assume a message stream:
>
> 1. Tick { stock = "UBS", bid = 12.3 }   RK="tick.UBS"
> 2. Tick { stock = "ABB", bid = 15.3 }  RK="tick.ABB"
> 3. Tick { stock = "UBS", bid = 12.28 } RK="tick.UBS"
>
> These are sent to a fanout exchange, and on to a queue(s) for a
> Consumer(s). Assume that consumer is slow, and the messages are not
> processed immediately. Rather than just en queuing message #3, I would
> like to insert it in place of message #1.
>
> This has the nice property that the broker won't overrun with messages
> if the consumer can't keep up; and that the consumer doesn't do work
> that is obsolete.
>
> What would be the suitable way to identify that #1 and #3 refer to the
> same stock? Is the routing key of message #1 retained after it has
> been en queued?
>
> An generalization of this would be to provide a function that takes
> the old and new message and combines them into an aggregated message.
> For example, we might want to track the latest, min and max:
>
>  AggregatedTick { stock = "UBS", latestBid = 12.28 minBid=12.28 maxBid=12.30 }
>
> So, does this sound sensible and possible?
>
> -jason
>
> [1] http://www.rabbitmq.com/extensions.html#queue-ttl
> _______________________________________________
> rabbitmq-discuss mailing list
> rabbitmq-discuss at lists.rabbitmq.com
> https://lists.rabbitmq.com/cgi-bin/mailman/listinfo/rabbitmq-discuss
>