001/* 002 * Licensed to the Apache Software Foundation (ASF) under one or more 003 * contributor license agreements. See the NOTICE file distributed with 004 * this work for additional information regarding copyright ownership. 005 * The ASF licenses this file to You under the Apache License, Version 2.0 006 * (the "License"); you may not use this file except in compliance with 007 * the License. You may obtain a copy of the License at 008 * 009 * http://www.apache.org/licenses/LICENSE-2.0 010 * 011 * Unless required by applicable law or agreed to in writing, software 012 * distributed under the License is distributed on an "AS IS" BASIS, 013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 014 * See the License for the specific language governing permissions and 015 * limitations under the License. 016 */ 017package org.apache.camel.util.concurrent; 018 019import java.util.concurrent.Callable; 020import java.util.concurrent.CompletionService; 021import java.util.concurrent.DelayQueue; 022import java.util.concurrent.Delayed; 023import java.util.concurrent.Executor; 024import java.util.concurrent.Future; 025import java.util.concurrent.FutureTask; 026import java.util.concurrent.TimeUnit; 027import java.util.concurrent.atomic.AtomicInteger; 028 029/** 030 * A {@link java.util.concurrent.CompletionService} that orders the completed tasks in the same order as they where 031 * submitted. 032 */ 033public class SubmitOrderedCompletionService<V> implements CompletionService<V> { 034 035 private final Executor executor; 036 037 // the idea to order the completed task in the same order as they where submitted is to leverage 038 // the delay queue. With the delay queue we can control the order by the getDelay and compareTo methods 039 // where we can order the tasks in the same order as they where submitted. 040 private final DelayQueue<SubmitOrderFutureTask> completionQueue = new DelayQueue<>(); 041 042 // id is the unique id that determines the order in which tasks was submitted (incrementing) 043 private final AtomicInteger id = new AtomicInteger(); 044 // index is the index of the next id that should expire and thus be ready to take from the delayed queue 045 private final AtomicInteger index = new AtomicInteger(); 046 047 private class SubmitOrderFutureTask extends FutureTask<V> implements Delayed { 048 049 // the id this task was assigned 050 private final long id; 051 052 SubmitOrderFutureTask(long id, Callable<V> voidCallable) { 053 super(voidCallable); 054 this.id = id; 055 } 056 057 SubmitOrderFutureTask(long id, Runnable runnable, V result) { 058 super(runnable, result); 059 this.id = id; 060 } 061 062 @Override 063 public long getDelay(TimeUnit unit) { 064 // if the answer is 0 then this task is ready to be taken 065 long answer = id - index.get(); 066 if (answer <= 0) { 067 return answer; 068 } 069 // okay this task is not ready yet, and we don't really know when it would be 070 // so we have to return a delay value of one time unit 071 if (TimeUnit.NANOSECONDS == unit) { 072 // okay this is too fast so use a little more delay to avoid CPU burning cycles 073 // To avoid align with java 11 impl of 074 // "java.util.concurrent.locks.AbstractQueuedSynchronizer.SPIN_FOR_TIMEOUT_THRESHOLD", otherwise 075 // no sleep with very high CPU usage 076 answer = 1001L; 077 } else { 078 answer = unit.convert(1, unit); 079 } 080 return answer; 081 } 082 083 @Override 084 @SuppressWarnings("unchecked") 085 public int compareTo(Delayed o) { 086 SubmitOrderFutureTask other = (SubmitOrderFutureTask) o; 087 return (int) (this.id - other.id); 088 } 089 090 @Override 091 protected void done() { 092 // when we are done add to the completion queue 093 completionQueue.add(this); 094 } 095 096 @Override 097 public String toString() { 098 // output using zero-based index 099 return "SubmitOrderedFutureTask[" + (id - 1) + "]"; 100 } 101 } 102 103 public SubmitOrderedCompletionService(Executor executor) { 104 this.executor = executor; 105 } 106 107 @Override 108 public Future<V> submit(Callable<V> task) { 109 if (task == null) { 110 throw new IllegalArgumentException("Task must be provided"); 111 } 112 SubmitOrderFutureTask f = new SubmitOrderFutureTask(id.incrementAndGet(), task); 113 executor.execute(f); 114 return f; 115 } 116 117 @Override 118 public Future<V> submit(Runnable task, Object result) { 119 if (task == null) { 120 throw new IllegalArgumentException("Task must be provided"); 121 } 122 SubmitOrderFutureTask f = new SubmitOrderFutureTask(id.incrementAndGet(), task, null); 123 executor.execute(f); 124 return f; 125 } 126 127 @Override 128 public Future<V> take() throws InterruptedException { 129 index.incrementAndGet(); 130 return completionQueue.take(); 131 } 132 133 @Override 134 public Future<V> poll() { 135 index.incrementAndGet(); 136 Future<V> answer = completionQueue.poll(); 137 if (answer == null) { 138 // decrease counter if we didnt get any data 139 index.decrementAndGet(); 140 } 141 return answer; 142 } 143 144 @Override 145 public Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException { 146 index.incrementAndGet(); 147 Future<V> answer = completionQueue.poll(timeout, unit); 148 if (answer == null) { 149 // decrease counter if we didnt get any data 150 index.decrementAndGet(); 151 } 152 return answer; 153 } 154 155 /** 156 * Marks the current task as timeout, which allows you to poll the next tasks which may already have been completed. 157 */ 158 public void timeoutTask() { 159 index.incrementAndGet(); 160 } 161 162}