@Test public void testCleanExpiredRequestInQueue() throws Exception { BrokerFastFailure brokerFastFailure = new BrokerFastFailure(null); BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(); brokerFastFailure.cleanExpiredRequestInQueue(queue, 1); assertThat(queue.size()).isZero(); Runnable runnable = new Runnable() { @Override public void run() { } }; queue.add(runnable); assertThat(queue.size()).isEqualTo(1); brokerFastFailure.cleanExpiredRequestInQueue(queue, 1); assertThat(queue.size()).isEqualTo(1); queue.clear(); RequestTask expiredRequest = new RequestTask(runnable, null, null); queue.add(new FutureTaskExt<>(expiredRequest, null)); TimeUnit.MILLISECONDS.sleep(100); RequestTask requestTask = new RequestTask(runnable, null, null); queue.add(new FutureTaskExt<>(requestTask, null)); assertThat(queue.size()).isEqualTo(2); brokerFastFailure.cleanExpiredRequestInQueue(queue, 100); assertThat(queue.size()).isEqualTo(1); assertThat(((FutureTaskExt) queue.peek()).getRunnable()).isEqualTo(requestTask); }
void cleanExpiredRequestInQueue(final BlockingQueue<Runnable> blockingQueue, final long maxWaitTimeMillsInQueue) { while (true) { try { if (!blockingQueue.isEmpty()) { final Runnable runnable = blockingQueue.peek(); if (null == runnable) { break; } final RequestTask rt = castRunnable(runnable); if (rt == null || rt.isStopRun()) { break; } final long behind = System.currentTimeMillis() - rt.getCreateTimestamp(); if (behind >= maxWaitTimeMillsInQueue) { if (blockingQueue.remove(runnable)) { rt.setStopRun(true); rt.returnResponse(RemotingSysResponseCode.SYSTEM_BUSY, String.format("[TIMEOUT_CLEAN_QUEUE]broker busy, start flow control for a while, period in queue: %sms, size of queue: %d", behind, blockingQueue.size())); } } else { break; } } else { break; } } catch (Throwable ignored) { } } }
BrokerFastFailure { void cleanExpiredRequestInQueue(final BlockingQueue<Runnable> blockingQueue, final long maxWaitTimeMillsInQueue) { while (true) { try { if (!blockingQueue.isEmpty()) { final Runnable runnable = blockingQueue.peek(); if (null == runnable) { break; } final RequestTask rt = castRunnable(runnable); if (rt == null || rt.isStopRun()) { break; } final long behind = System.currentTimeMillis() - rt.getCreateTimestamp(); if (behind >= maxWaitTimeMillsInQueue) { if (blockingQueue.remove(runnable)) { rt.setStopRun(true); rt.returnResponse(RemotingSysResponseCode.SYSTEM_BUSY, String.format("[TIMEOUT_CLEAN_QUEUE]broker busy, start flow control for a while, period in queue: %sms, size of queue: %d", behind, blockingQueue.size())); } } else { break; } } else { break; } } catch (Throwable ignored) { } } } }
BrokerFastFailure { void cleanExpiredRequestInQueue(final BlockingQueue<Runnable> blockingQueue, final long maxWaitTimeMillsInQueue) { while (true) { try { if (!blockingQueue.isEmpty()) { final Runnable runnable = blockingQueue.peek(); if (null == runnable) { break; } final RequestTask rt = castRunnable(runnable); if (rt == null || rt.isStopRun()) { break; } final long behind = System.currentTimeMillis() - rt.getCreateTimestamp(); if (behind >= maxWaitTimeMillsInQueue) { if (blockingQueue.remove(runnable)) { rt.setStopRun(true); rt.returnResponse(RemotingSysResponseCode.SYSTEM_BUSY, String.format("[TIMEOUT_CLEAN_QUEUE]broker busy, start flow control for a while, period in queue: %sms, size of queue: %d", behind, blockingQueue.size())); } } else { break; } } else { break; } } catch (Throwable ignored) { } } } BrokerFastFailure(final BrokerController brokerController); }
BrokerFastFailure { void cleanExpiredRequestInQueue(final BlockingQueue<Runnable> blockingQueue, final long maxWaitTimeMillsInQueue) { while (true) { try { if (!blockingQueue.isEmpty()) { final Runnable runnable = blockingQueue.peek(); if (null == runnable) { break; } final RequestTask rt = castRunnable(runnable); if (rt == null || rt.isStopRun()) { break; } final long behind = System.currentTimeMillis() - rt.getCreateTimestamp(); if (behind >= maxWaitTimeMillsInQueue) { if (blockingQueue.remove(runnable)) { rt.setStopRun(true); rt.returnResponse(RemotingSysResponseCode.SYSTEM_BUSY, String.format("[TIMEOUT_CLEAN_QUEUE]broker busy, start flow control for a while, period in queue: %sms, size of queue: %d", behind, blockingQueue.size())); } } else { break; } } else { break; } } catch (Throwable ignored) { } } } BrokerFastFailure(final BrokerController brokerController); static RequestTask castRunnable(final Runnable runnable); void start(); void shutdown(); }
BrokerFastFailure { void cleanExpiredRequestInQueue(final BlockingQueue<Runnable> blockingQueue, final long maxWaitTimeMillsInQueue) { while (true) { try { if (!blockingQueue.isEmpty()) { final Runnable runnable = blockingQueue.peek(); if (null == runnable) { break; } final RequestTask rt = castRunnable(runnable); if (rt == null || rt.isStopRun()) { break; } final long behind = System.currentTimeMillis() - rt.getCreateTimestamp(); if (behind >= maxWaitTimeMillsInQueue) { if (blockingQueue.remove(runnable)) { rt.setStopRun(true); rt.returnResponse(RemotingSysResponseCode.SYSTEM_BUSY, String.format("[TIMEOUT_CLEAN_QUEUE]broker busy, start flow control for a while, period in queue: %sms, size of queue: %d", behind, blockingQueue.size())); } } else { break; } } else { break; } } catch (Throwable ignored) { } } } BrokerFastFailure(final BrokerController brokerController); static RequestTask castRunnable(final Runnable runnable); void start(); void shutdown();  }