забыл ник ,
Ну вообще это достаточно серьезный косяк. Решений несколько, и все они разрливаются через мануальное создание пула.
1) Использовать ThreadPoolExecutor.CallerRunsPolicy. Это самый простой вариант - если не получается выполнить асинхронно, то выполнить синхронно. Но не факт, что он вас устроит по двум причинам:
- Долгий ответ клиенту, от которого хотелось избавиться.
- Starvation никуда не делся, а просто переместился в другое место - теперь у вас может застарвиться один клиент, если он в первом окне браузера начал синхронное выполнение, а во втором окне постоянно тыкает вперед-назад. В этом случае, вы рискуете в первом окне браузера так и не получить ответ :-)
В общем, сомнительное решение.
2) Более заморочный вариант:
2.1) Создать руками BlockingQueue.
2.2) Передать его в ThreadPoolExecutor.
2.3) Класть задачи не в ThreadPoolExecutor, а напрямую в BlockingQueue.
2.4) Если какой-то TaskRunnable слишком долго выполняется, то форсировать его завершение, и класть в конец этой очереди.
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
60.
61.
62.
63.
64.
65.
66.
67.
68.
69.
70.
71.
72.
73.
74.
75.
76.
77.
78.
79.
80.
81.
82.
83.
84.
85.
86.
87.
88.
89.
90.
91.
92.
93.
94.
95.
96.
97.
98.
99.
100.
101.
102.
103.
104.
105.
106.
107.
108.
109.
110.
111.
112.
113.
114.
115.
116.
117.
118.
119.
120.
121.
122.
123.
124.
125.
126.
127.
128.
129.
130.
131.
132.
133.
134.
135.
136.
137.
138.
139.
140.
141.
142.
143.
144.
145.
146.
147.
148.
149.
150.
151.
152.
153.
154.
155.
156.
157.
158.
159.
160.
161.
162.
163.
164.
165.
166.
167.
168.
169.
170.
171.
172.
173.
174.
175.
176.
177.
178.
179.
180.
181.
182.
183.
184.
185.
186.
187.
188.
189.
190.
191.
192.
193.
194.
195.
196.
197.
198.
199.
200.
201.
202.
203.
204.
205.
206.
207.
208.
209.
210.
211.
212.
213.
214.
215.
216.
217.
218.
219.
220.
221.
222.
223.
224.
225.
226.
227.
228.
229.
230.
public class TaskBlockingAndQueued {
/** Монитор. */
private static final Object mux = new Object();
/** Задачи. */
private static final Map<Integer, TaskRunnable> tasks = new HashMap<Integer, TaskRunnable>();
/** Очередь задач. */
private static final BlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<>();
/** Пул. */
private static final ExecutorService e = new ThreadPoolExecutor(8, 8, 1000L, TimeUnit.MILLISECONDS, taskQueue);
public static void main(String[] args) throws Exception {
addTask(1, new Task(1));
//Thread.sleep(200); // Расскомментируйте, и увидите, что выполнится и первый, и второй.
addTask(1, new Task(2));
e.shutdown();
}
/**
* Добавить задачу.
*
* @param id Идентификатор.
* @param task Задача.
*/
public static void addTask(Integer id, Task task) {
boolean isNew = false;
TaskRunnable taskRun;
synchronized (mux) {
taskRun = tasks.get(id);
if (taskRun != null)
taskRun.addTask(task);
else {
System.out.println("Put new task");
taskRun = new TaskRunnable(id, task);
TaskRunnable oldTaskRun = tasks.put(id, taskRun);
assert oldTaskRun == null;
isNew = true;
}
}
if (isNew)
submit(taskRun);
}
/**
* Подождать окончания выполнения задач для данного идентификатора.
*
* @param id Идентификатор.
* @throws InterruptedException В случае прерывания.
*/
public static void awaitCompletion(Integer id) throws InterruptedException {
TaskRunnable taskRun;
synchronized (mux) {
taskRun = tasks.get(id);
if (taskRun == null)
return; // Нет активных задач.
}
taskRun.awaitCompletion();
}
/**
* Отдать TaskRunnable в Executor.
*
* @param taskRun TaskRunnable.
*/
private static void submit(TaskRunnable taskRun) {
try {
taskQueue.put(taskRun);
}
catch (InterruptedException e) {
// Handle.
}
}
/**
* Класс, выполняющий задачи.
*/
private static class TaskRunnable implements Runnable {
/** Идентификатор. */
private final int id;
/** Latch, означающий окончание выполнение задач для текушего id. */
private final CountDownLatch completeLatch = new CountDownLatch(1);
/** Текущая задача. */
private Task task;
/**
* Конструктор.
*
* @param id Идентификатор.
* @param task Задача.
*/
public TaskRunnable(int id, Task task) {
assert Thread.holdsLock(mux); // Должны быть внутри synchronized для видимости.
this.id = id;
this.task = task;
}
/**
* Добавить следующую задачу.
*
* @param task Следующая задача.
*/
public void addTask(Task task) {
synchronized (mux) {
this.task = task;
}
}
/** {@inheritDoc} */
@Override
public void run() {
while (true) {
boolean proceed = false;
Task task0;
synchronized (mux) {
task0 = task;
if (task0 == null) {
// Все, больше нет задач в очереди.
tasks.remove(id);
completeLatch.countDown();
return;
}
else {
if (canProceed()) {
proceed = true;
task = null; // Занулим следующую задачу.
}
}
}
if (proceed) {
try {
task0.run(); // Выполнить задачу.
}
catch (Exception e) {
// Обработка на ваше усмотрение.
}
}
else {
returnToQueue();
return;
}
}
}
/**
* Подождать окончания выполнения задачи.
*
* @throws InterruptedException В случае прерывания.
*/
public void awaitCompletion() throws InterruptedException {
completeLatch.await();
}
/**
* Проверить, можно ли продолжать выполнение задач, или надо вернуть объект в конец очереди.
*
* @return {@code true}, если
*/
private boolean canProceed() {
// Ваша имплементация.
return true;
}
/**
* Вернуть задачу в очередь.
*/
private void returnToQueue() {
// Ваша исплементация, напр.:
try {
taskQueue.put(this);
}
catch (InterruptedException e) {
// Handle.
}
}
}
/**
* Ваша задача.
*/
private static class Task implements Runnable {
private final int id;
private Task(int id) {
this.id = id;
}
/** {@inheritDoc} */
@Override
public void run() {
System.out.println("Task started [id=" + id + "]");
try {
Thread.sleep(2000);
}
catch (InterruptedException e) {
// Handle.
}
System.out.println("Task finished [id=" + id + "]");
}
}
}
P.S.: Ну и помните дисклеймер про наличие багов. Это не production code, а демонстрация идеи.
