java.util.concurrent.DelayQueue#take ( )源码实例Demo

/**
 * Delayed actions do not occur until their delay elapses
 */
public void testDelay() throws InterruptedException {
    DelayQueue<NanoDelay> q = new DelayQueue<>();
    for (int i = 0; i < SIZE; ++i)
        q.add(new NanoDelay(1000000L * (SIZE - i)));

    long last = 0;
    for (int i = 0; i < SIZE; ++i) {
        NanoDelay e = q.take();
        long tt = e.getTriggerTime();
        assertTrue(System.nanoTime() - tt >= 0);
        if (i != 0)
            assertTrue(tt >= last);
        last = tt;
    }
    assertTrue(q.isEmpty());
}
 

public static void monitorThread(final DelayQueue<Session> queue){
	Thread thread=new Thread(){
		public void run(){
			while(!queue.isEmpty()){
				try {
					Session dt=queue.take();
					String str=dt.toString2()+"=====已被清理"+"\t currentTime:"+System.currentTimeMillis();
					System.out.println(str);
					result.add(str);
					result.add(iteratorDelayQueue(queue));	//每次清理之后，都要保存一下队列的快照
				} catch (InterruptedException e) { System.out.println("清理中断...."); return; }
			}
			System.out.println("清理完所有缓存！！！！！");
			
			System.out.println("======================延迟对象生命周期运行时queue的快照========================");
			for(String tstr:result){
				System.out.println(tstr);
			}
			System.out.println("==============================================");
		}
	};
	thread.start();
}
 

/**
 * Delayed actions do not occur until their delay elapses
 */
public void testDelay() throws InterruptedException {
    DelayQueue<NanoDelay> q = new DelayQueue<NanoDelay>();
    for (int i = 0; i < SIZE; ++i)
        q.add(new NanoDelay(1000000L * (SIZE - i)));

    long last = 0;
    for (int i = 0; i < SIZE; ++i) {
        NanoDelay e = q.take();
        long tt = e.getTriggerTime();
        assertTrue(System.nanoTime() - tt >= 0);
        if (i != 0)
            assertTrue(tt >= last);
        last = tt;
    }
    assertTrue(q.isEmpty());
}
 

@Test
public void test() throws InterruptedException {
    DelayQueue<IntDelay> delayQueue = new DelayQueue<>();
    for (int i = 0; i < 10; i++) {
        delayQueue.add(new IntDelay(i));
    }
    while (!delayQueue.isEmpty()) {
        IntDelay delay = delayQueue.take();
        if (Objects.nonNull(delay)) {
            System.out.println(delay.num);
        }
    }
}
 

public void start(final Runnable task)
{
	semaphore = new Semaphore(0);
	shouldRun = true;
	Thread thread = new Thread(new Runnable() {
		
		@Override
		public void run() 
		{
			try{
				DelayQueue<Delayed> dq = new DelayQueue<Delayed>();
				MyDelayed dl = new MyDelayed();
				
				while(shouldRun)
				{
					task.run();
					dl.nextAt(period, TimeUnit.NANOSECONDS);
					dq.add(dl);
					try {
						dq.take();
					} catch (InterruptedException e) {
						// Thread interruption is fine.
					}
				}
			}finally{
				semaphore.release();
			}
			
		}
	});
	thread.setPriority(Thread.MAX_PRIORITY);
	thread.start();
}
 

public void start(final Runnable taskA, final Runnable taskB)
{
	dlA = new MyDelayed(taskA);
	dlB = new MyDelayed(taskB);

	semaphore = new Semaphore(0);
	shouldRun = true;
	Thread thread = new Thread(new Runnable() {
		
		@Override
		public void run() 
		{
			try{
				DelayQueue<MyDelayed> dq = new DelayQueue<MyDelayed>();
				
				dlA.nextAt(period, TimeUnit.NANOSECONDS);
				dq.add(dlA);
				dlB.nextAt(period, TimeUnit.NANOSECONDS);
				dq.add(dlB);
				
				while(shouldRun)
				{
					try {
						MyDelayed dl = dq.take();
						dl.task.run();
						dl.nextAt(period, TimeUnit.NANOSECONDS);
						dq.add(dl);
					} catch (InterruptedException e) {
						// Thread interruption is fine.
					}
				}
			}finally{
				semaphore.release();
			}
			
		}
	});
	thread.setPriority(Thread.MAX_PRIORITY);
	thread.start();
}
 

/**
 * @return null (if an interrupt) or an instance of E; resets interrupt on calling thread.
 */
public static <E extends Delayed> E takeWithoutInterrupt(final DelayQueue<E> queue) {
  try {
    return queue.take();
  } catch (InterruptedException e) {
    Thread.currentThread().interrupt();
    return null;
  }
}