The situation offers the next strategies:
public interface Situation {
void await() throws InterruptedException;
void awaitUninterruptibly();
lengthy awaitNanos(lengthy nanosTimeout) throws InterruptedException;
boolean await(very long time, TimeUnit unit) throws InterruptedException;
boolean awaitUntil(Date deadline) throws InterruptedException;
void sign();
void signalAll();
}
The closest we got here to that to date is the wait Object Monitor methodology.
A Situation is sure to a Lock and a thread can’t work together with a situation and its strategies if it doesn’t have a maintain on that lock.
Additionally Situation makes use of the underlying lock mechanisms, for instance sign and signalAll will use the underlying Queue of the threads that’s maintained by the Lock and can notify them to get up.
One of many apparent issues to implement utilizing Circumstances is a BlockingQueue. Employee threads processing information and writer threads dispatching information. Information are printed on a queue, employee threads will course of information from the queue after which they need to wait if there isn’t any information within the queue.
For a employee thread, if the situation is met the circulate is the next:
◉ Purchase the lock
◉ Verify the situation
◉ Course of Information
◉ Launch the lock
If the situation is just not met, the circulate would barely change to this:
◉ Purchase the lock
◉ Verify the situation
◉ Wait till the situation is met.
◉ Re-acquire the lock
◉ Course of Information
◉ Launch the lock
The writer thread every time it provides a message it ought to notify the threads ready on the situation.
The workflow could be like this.
◉ Purchase the lock
◉ Publish information
◉ Notify the employees
◉ Launch the lock
Clearly this performance already exists by way of the BlockingQueue interface and the LinkedBlockingDeque and ArrayBlockingQueue implementation.
We’ll proceed with an implementation for the shake of the instance.
Let’s see the Message Queue:
package deal com.gkatzioura.concurrency.lock.situation;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.locks.Situation;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class MessageQueue<T> {
non-public Queue<T> queue = new LinkedList<>();
non-public Lock lock = new ReentrantLock();
non-public Situation hasMessages = lock.newCondition();
public void publish(T message) {
lock.lock();
strive {
queue.provide(message);
hasMessages.sign();
} lastly {
lock.unlock();
}
}
public T obtain() throws InterruptedException {
lock.lock();
strive {
whereas (queue.isEmpty()) {
hasMessages.await();
}
return queue.ballot();
} lastly {
lock.unlock();
}
}
}
Now let’s put it into motion:
MessageQueue<String> messageQueue = new MessageQueue<>();
@Check
void testPublish() throws InterruptedException {
Thread writer = new Thread(() -> {
for (int i = 0; i < 10; i++) {
String message = “Sending message num: ” + i;
log.information(“Sending [{}]”, message);
messageQueue.publish(message);
strive {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
});
Thread worker1 = new Thread(() -> {
for (int i = 0; i < 5; i++) {
strive {
String message = messageQueue.obtain();
log.information(“Acquired: [{}]”, message);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
});
Thread worker2 = new Thread(() -> {
for (int i = 0; i < 5; i++) {
strive {
String message = messageQueue.obtain();
log.information(“Acquired: [{}]”, message);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
});
writer.begin();
worker1.begin();
worker2.begin();
writer.be part of();
worker1.be part of();
worker2.be part of();
}
That’s it! Our staff processed the anticipated messages and waited when the queue was empty.
Supply: javacodegeeks.com
