Сам заказал - сам спляшу:
| Код |
import java.util.concurrent.ArrayBlockingQueue;
public class ThreadedBFSTest {
public static void main(String[] args) throws InterruptedException {
Thread[] t = new Thread[100];
Producer p = new Producer();
p.put(new Object());
for (int i = 0; i < t.length; i++)
t[i] = new Thread(new Consumer(p));
for (Thread tt : t)
tt.start();
for (Thread tt : t)
tt.join();
System.out.println("Done");
}
private static class Producer {
private ArrayBlockingQueue<Object> queue = new ArrayBlockingQueue<Object>(4096);
private int queued;
private final Object LOCK = new Object();
public boolean isDone() {
synchronized(LOCK) {
do {
System.out.println("Queued = " + queued + "; queue size = " + queue.size());
if (!queue.isEmpty())
return false;
if (queued == 0)
return true;
try {
LOCK.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
} while (true);
}
}
public boolean hasNext() {
return !queue.isEmpty();
}
public Object get() throws InterruptedException {
synchronized(LOCK) {
Object res = queue.take();
queued++;
LOCK.notifyAll();
return res;
}
}
public void put(Object o) {
if (queue.offer(o))
synchronized(LOCK) {
LOCK.notifyAll();
}
}
public void jobDone() {
synchronized(LOCK) {
queued--;
LOCK.notifyAll();
}
}
}
private static class Consumer implements Runnable {
private long startTime;
private Producer producer;
public Consumer(Producer producer) {
this.producer = producer;
startTime = System.currentTimeMillis();
}
public void run() {
while (!producer.isDone()) {
try {
System.out.println(producer.get());
System.out.println(Thread.currentThread().getId());
Thread.yield();
} catch (InterruptedException e) {
}
if (System.currentTimeMillis() <= startTime + 15000) {
//System.out.println("Putting object");
producer.put(new Object());
try {
Thread.sleep(50);
} catch (InterruptedException e) {
}
//System.out.println("Putting object");
producer.put(new Object());
}
producer.jobDone();
}
try {
Thread.sleep(100);
} catch (InterruptedException e) {
}
System.out.println("Finished");
}
}
}
|
Как обычно я не знаю как это работает (в смысле, синхронизация на додумках и соплях), но работает. Пугает меня метод isDone в классе Producer. Работает он верно, но ответ выдавать не сразу, а только когда поймет, завершена ли работа или нет. Правильно ли так делать? Если думать с точки зрения удобства, считаю, удобно, программисту, разработчику Consumer не надо думать и гадать, как сделать правильное ожидание разрешения ситуации.
И я что-то запамятовал, почему этот код работает?
| Код |
while (p.hasNext()) {
Object o = p.get();
}
|
hasNext и get синхронизированы. Но если подумать hasNext могут запросить 2 потока, сначала одному будет сказано, что есть следующий, потом второму будет сказано, что есть следующий, а затем они вдвоем помчатся ловить p.get(), один из них может навечно остаться ждать данных в p.get()
Добавлено через 12 минут и 54 секунды
Может правильней такая конструкция:
| Код |
while (true) {
synchronized(producer) {
if (producer.isDone()) break;
System.out.println(producer.get());
}
}
|
Тоже работает. |
)