public void start ( ) { scheduledExecutorService = Executors . newScheduledThreadPool ( 2 , new ThreadFactory ( ) { @ Override public Thread newThread ( Runnable runnable ) { Thread thread = new Thread ( runnable ) ; thread . setName ( " DataQueue Process " + source ) ; return thread ; } } ) ; future = scheduledExecutorService . scheduleAtFixedRate ( new Runnable ( ) { @ Override public void run ( ) { if ( stop . get ( ) ) { return ; } try { processWriteQueue ( ) ; } catch ( InterruptedException ex ) { //let it restart or stop } catch ( Exception ex ) { logger . fatal ( ex ) ; } } } , 0 , dataStoreConfig . threadErrorResumeTimeMS ( ) , TimeUnit . MILLISECONDS ) ; future = scheduledExecutorService . scheduleAtFixedRate ( new Runnable ( ) { @ Override public void run ( ) { if ( stop . get ( ) ) { return ; } try { processReadQueue ( ) ; } catch ( InterruptedException ex ) { //let it restart or stop } catch ( Exception ex ) { logger . fatal ( ex , "Problem with base data store running scheduled job" ) ; } } } , 0 , dataStoreConfig . threadErrorResumeTimeMS ( ) , TimeUnit . MILLISECONDS ) ; }
Start up the queue handlers .