Multithreading in Java — Part 5
Sychronized Collections
Cyclic Barrier
A cyclic barrier allows a set of threads to wait for each other to reach a common barrier point. Once all the threads reach it, the barrier can be reused, as it resets automatically.
public class CyclicBarrierDemo {
private static final int NO_OF_THREADS = 4;
private static final int STAGE = 3;
private static final CyclicBarrier barrier = new CyclicBarrier(NO_OF_THREADS,() -> {
System.out.println("All threads reached the barrier. Proceeding...");
});
public static void main(String[] args) {
for(int i=0;i<NO_OF_THREADS;i++){
Thread thread = new Thread(new DemoThread(i));
thread.start();
}
}
static class DemoThread implements Runnable {
private final int threadId;
public DemoThread(int threadId) {
this.threadId = threadId;
}
@Override
public void run() {
for(int i=0;i<STAGE;i++){
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
System.out.println("Thread "+ threadId + " completed stage "+ (i+1));
try {
barrier.await();
System.out.println("Thread "+ threadId+ " continue execution");
} catch (InterruptedException e) {
throw new RuntimeException(e);
} catch (BrokenBarrierException e) {
throw new RuntimeException(e);
}
}
}
}
}Output
Here, each thread completes three stages. Once one thread reaches the barrier point after completing a stage, it waits for other threads to reach it. barrier.await() waiting for all threads to reach the barrier point. Once all threads complete the stage and reach the barrier point, release the barrier and continue the execution.
Exchanger
Get Gavindualwis’s stories in your inbox
Join Medium for free to get updates from this writer.
An exchanger is a synchronized point where two threads can exchange objects. Each thread presents an object to exchange() method. After two threads present their objects to exchange() method swap objects and continue the process.
public class ExchangerDemo {
public static void main(String[] args) {
Exchanger<Integer> exchanger = new Exchanger<>();
Thread one = new Thread(new FirstThread(exchanger));
Thread two = new Thread(new SecondThread(exchanger));
one.start();
two.start();
}
}
class FirstThread implements Runnable{
private final Exchanger<Integer> exchanger;
public FirstThread(Exchanger<Integer> exchanger){
this.exchanger = exchanger;
}
@Override
public void run() {
int dataToSend = 10;
System.out.println("First thread sending data: "+dataToSend);
try {
Integer receivedData = exchanger.exchange(dataToSend);
System.out.println("First thread received data: "+receivedData);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
class SecondThread implements Runnable{
private final Exchanger<Integer> exchanger;
public SecondThread(Exchanger<Integer> exchanger){
this.exchanger = exchanger;
}
@Override
public void run() {
try {
Thread.sleep(3000);
int dataToSend = 20;
System.out.println("Second thread sending data: "+dataToSend);
Integer receivedData = exchanger.exchange(dataToSend);
System.out.println("Second thread received data: "+receivedData);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}Output
Here thread one prints the data that is sent as 10. Then, thread one is presented to the exchange() method. At that time thread two is in the sleep state for 3 seconds. After thread one presents to exchange() method, it waits until thread two presents to exchange() method. After 3 seconds start executing thread two. Thread two print data to send as 20. Then present to exchange() method. Now as both threads present to exchange() method happen the swapping. The output is given as thread one receives data as 20 and thread two receives data as 10 after swapping.
CopyOnWriteArray
Here creates a new copy of the underlying array every time you modify it. This is safe for concurrent access of reads and writes.
public class CopyOnWriteArrayDemo {
public static void main(String[] args) {
Simulation simulation = new Simulation();
simulation.simulate();
}
}
class Simulation{
private List<Integer> list;
public Simulation(){
this.list = new ArrayList<Integer>();
this.list.addAll(Arrays.asList(0,0,0,0,0));
}
public void simulate() {
Thread one = new Thread(new WriteTask(list));
Thread two = new Thread(new WriteTask(list));
Thread three = new Thread(new WriteTask(list));
Thread four = new Thread(new ReadTask(list));
one.start();
two.start();
three.start();
four.start();
}
}
class ReadTask implements Runnable{
private final List<Integer> list;
public ReadTask(List<Integer> list){
this.list = list;
}
@Override
public void run() {
while (true){
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
System.out.println(list);
}
}
}
class WriteTask implements Runnable{
private final List<Integer> list;
private Random random;
public WriteTask(List<Integer> list){
this.list = list;
this.random = new Random();
}
@Override
public void run() {
while (true) {
try {
Thread.sleep(1200);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
list.set(random.nextInt(list.size()), random.nextInt(10));
}
}
}Output
[0, 0, 0, 0, 0]
[0, 4, 0, 4, 0]
[9, 4, 7, 4, 9]
[9, 4, 7, 4, 2]
[5, 4, 7, 8, 2]
[5, 4, 7, 8, 5]
[5, 4, 7, 8, 2]
[5, 6, 7, 1, 2]
[1, 6, 1, 1, 2]
[8, 6, 3, 1, 2]…
The process that happens here is when a thread needs to read from the array it creates the snapshot of the array and uses that to read. If a thread wants to write to the array it creates a snapshot and performs the write operation. Once the write operation is completed changed array is considered as the latest version of the array.
Next Chapter — Part 6 — Locks
https://medium.com/@gavindualwis99/multithreading-in-java-part-6-124753af7ed4
