A Real-Life Example - Getting Started with Storm - page 54

String sCount = jedis . hget ( buildRedisKey ( product ), categ );
if ( sCount == null || "nil" . equals ( sCount )) {
count = 0 ;
} else {
count = Integer . valueOf ( sCount );
}
}
return count ;
}
...
private void storeProductCategoryCount ( String categ , String product , int count ) {
String key = buildLocalKey ( categ , product );
counter . put ( key , count );
synchronized ( pendingToSave ) {
pendingToSave . put ( key , count );
}
}
...
}
The getProductCategoryCount method first looks in memory cache counter. If the in-
formation is not available there, it gets it from the Redis server.
The storeProductCategoryCount method updates the counter cache and the pending-
ToSave buffer. The buffer is persisted by the following background thread:
package storm . analytics ;
public class ProductCategoriesCounterBolt extends BaseRichBolt {
...
private void startDownloaderThread () {
TimerTask t = new TimerTask () {
@Override
public void run () {
HashMap < String , Integer > pendings ;
synchronized ( pendingToSave ) {
pendings = pendingToSave ;
pendingToSave = new HashMap < String , Integer >();
}
for ( String key : pendings . keySet ()) {
String [] keys = key . split ( ":" );
String product = keys [ 0 ];
String categ = keys [ 1 ];
Integer count = pendings . get ( key );
jedis . hset ( buildRedisKey ( product ), categ , count . toString ());
}
}
};
timer = new Timer ( "Item categories downloader" );
timer . scheduleAtFixedRate ( t , downloadTime , downloadTime );
}
...
}