Beforehand we checked on ReentRantLock and its equity. One of many issues we are able to come upon is the creation of a Situation. Through the use of Situation we are able to create mechanisms that enable threads to attend for particular circumstances to be met earlier than continuing with their execution.
The situation supplies 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 thus far is the wait Object Monitor methodology.
A Situation is sure to a Lock and a thread can not 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. Knowledge are revealed on a queue, employee threads will course of information from the queue after which they need to wait if there isn’t a information within the queue.
For a employee thread, if the situation is met the circulate is the next:
- Purchase the lock
- Test the situation
- Course of Knowledge
- Launch the lock
If the situation shouldn’t be met, the circulate would barely change to this:
- Purchase the lock
- Test the situation
- Wait till the situation is met.
- Re-acquire the lock
- Course of Knowledge
- Launch the lock
The writer thread at any time when 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 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:
bundle 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> {
personal Queue<T> queue = new LinkedList<>();
personal Lock lock = new ReentrantLock();
personal 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<>();
@Take a look at
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 a part of();
worker1.be a part of();
worker2.be a part of();
}
That’s it! Our staff processed the anticipated messages and waited when the queue was empty.
|
Printed on Java Code Geeks with permission by Emmanouil Gkatziouras, companion at our JCG program. See the unique article right here: Java Concurrency: Situation Opinions expressed by Java Code Geeks contributors are their very own. |
