Столкнулся с интересной проблемой, нужна помощь:)
Задача с рабочего проекта(веб). Смысл вот в чем - имеется визард, на каждой странице пользователь заполняет поля, жмет Некст, в это время посылается запрос на веб-сервис на их сохранение и после этого рендерится страница. Вызов веб-сервиса долгий, не хочется мучать юзера, поэтому появилась мысль ложить таск по сохранению атрибутов в некое хранилище и выполнять его асинхронно.
Доп требования -
1) Каждый юзер имеет свой уникальный entityNumber
2) Нельзя дать выполняться двум таскам одновременно для одинакового entityNumber(более старый может перетереть более новый, если новый выполнится быстрее)
3) Допустим выполняется таск для entityNumber =1. Пока он выполняется, юзер успел нажать два раза Некст, имеет смысл посылать только последний таск для этого entityNumber(приложение устроено так, что Некст с 4 страницы содержит все атрибуты с 1 по 4 страницы). То есть если выполняется таск для 1 страницы, в это время пришли со 2 и с 3, надо сделать так чтобы 2 вообще не выполнялся, а сразу 3(когда закончится таки 1).
4) Вот тут косяк :) Предположим юзер по ошибке нажал Ф5. Я в контроллере хочу убедиться, что в хранилище нет выполняющихся для entityNumber юзера. Если выполняются - подождать, пока они выполнятся.
Моя идея была такова - Сохранять таски в ArrayBlockingQueue. Запустить поток, который лочит очередь, берет первый таск, если в очереди есть еще таск с данным entityNumber - реджектаем его, если нет - то добавляет в список выполняющихся entityNumber и посылает его в экзекьютор. Если уже выполняется таск с таким entityNumber, то помещаем таск в конец очереди.
Скажу сразу, я знаю что один лок - это потенциальное проседание производительности, но меня устраивает подождать 0.1с, вместо нынешних 15 на веб-сервис:) Все работало неплохо, пока не появилось требование 4) - моя имплементация покрывает 1-3, а 4) никак не пойму как сделать.
Вот мой код -
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.
package test;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class Pool {
static Pool pool;
ArrayBlockingQueue<Task> queue = new ArrayBlockingQueue<Task>(1000);
Lock lock = new ReentrantLock();
volatile boolean stopped;
Watcher watcher = null;
List<Long> runned = new ArrayList<Long>();
private Pool() {
}
public static synchronized Pool getInstance() {
if (pool == null) {
pool = new Pool();
pool.init();
}
return pool;
}
public void waitForMemberComplete(long entity){
for (;;) {
try {
lock.lock();
if (runned.contains(entity)) {
Thread.sleep(100);
System.out.println("Member task runned, sleep");
continue;
}
boolean needSleep = false;
for (Task t : queue) {
if (t.getEntityNumber() == entity) {
System.out.println("Member task in queue, sleep");
needSleep = true;
break;
}
}
if (needSleep) {
Thread.sleep(100);
continue;
}
System.out.println("Member has no pending tasks");
return;
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
public void put(Task task) {
if (stopped) {
return;
}
try {
lock.lock();
queue.add(task);
System.out.println("Task added for " + task.getEntityNumber());
} finally {
lock.unlock();
}
}
public void destroy() {
stopped = true;
watcher.shutdown();
}
ArrayBlockingQueue<Task> getQueue() {
return queue;
}
private void init() {
watcher = new Watcher(pool, lock, runned);
watcher.start();
}
boolean isStopped() {
return stopped;
}
public static void main(String[] args) throws InterruptedException {
Task t = new Task(0);
Task t2 = new Task(0);
Task t3 = new Task(0);
Task t4 = new Task(1);
Task t5 = new Task(2);
Pool.getInstance().put(t);
Thread.sleep(100);
Pool.getInstance().put(t2);
Thread.sleep(100);
Pool.getInstance().put(t3);
Pool.getInstance().put(t4);
Pool.getInstance().put(t5);
Pool.getInstance().waitForMemberComplete(0);
}
}
class Task implements Runnable {
long entity;
public Task (long entity) {
this.entity = entity;
}
public long getEntityNumber() {
return entity;
}
@Override
public void run() {
try {
Thread.sleep(300);
System.out.println("Task done");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
class Watcher extends Thread {
Pool pool;
Lock lock;
ExecutorService executor;
List<Long> runned;
public Watcher(Pool pool, Lock lock, List<Long> runned) {
this.pool = pool;
this.lock = lock;
this.runned = runned;
executor = new CustomExecutor(runned, lock);
}
public void shutdown() {
List<Runnable> notRunned = executor.shutdownNow();
// Maybe save in logs?
}
public void run() {
for (;;) {
try {
Thread.sleep(20);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
if (pool.isStopped()) {
return;
}
Task toRun = null;
try {
toRun = pool.getQueue().take();
} catch (InterruptedException e) {
return;
}
try {
lock.lock();
boolean hasNewest = false;
for (Task t : pool.getQueue()) {
if (toRun.getEntityNumber() == t.getEntityNumber()) {
System.out.println("Member has newest task " + t.getEntityNumber());
hasNewest = true;
break;
}
}
if (hasNewest) {
continue;
}
if (runned.contains(toRun.getEntityNumber())) {
pool.getQueue().add(toRun);
System.out.println("Member task running, put in end queue " + toRun.getEntityNumber());
continue;
}
runned.add(toRun.getEntityNumber());
executor.execute(toRun);
System.out.println("Task scheduled for member" + toRun.getEntityNumber());
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
}
final class CustomExecutor extends ThreadPoolExecutor {
List<Long> runned;
Lock lock;
public CustomExecutor(List<Long> runned, Lock lock) {
super(5, 10, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
this.runned = runned;
this.lock = lock;
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
super.afterExecute(r, t);
try {
lock.lock();
Task runnable = (Task) r;
runned.remove(runnable.getEntityNumber());
System.out.println("Task completed, executor can accept new tasks for " + runnable.getEntityNumber());
} finally {
lock.unlock();
}
}
}
Собственно проблема понятна, она вот тут - toRun = pool.getQueue().take();
То есть может случиться так, что не выполняется ни одного таска с энтити, и его нет в очереди, но его мы только что вытащили из очереди, поэтому waitForMemberComplete() работает неправильно. Ложить его под Лок тоже нельзя по понятным причинам. Что посоветуете? Можно и абсолютно другие варианты, если знаете что может помочь. Ну и попутно гляньте, может есть еще ошибки, замечания?
