Concurrent programming

Concurrently Programming Challenges

1. Race conditions: When two or more threads access shared data simultaneously, the final value of the data might depend on the order in which the threads are executed. This can lead to unpredictable results and hard-to-reproduce bugs.

2. Deadlocks: A situation where two or more threads are stuck waiting for each other to release a resource (e.g., a lock), causing the program to hang indefinitely. Deadlocks usually occur when multiple threads acquire locks in an inconsistent order.

3. Livelocks: A livelock is similar to a deadlock, but in this case, threads are not completely blocked; they are just unable to make progress due to the constant change of states. A livelock can occur when two or more threads are stuck in a loop, continuously responding to each other's actions but not making any progress.

4. Starvation: A situation where a thread is unable to make progress because it is constantly waiting for a shared resource that is always in use by other threads. Starvation can lead to reduced performance or even application crashes.

5. Resource contention: When multiple threads compete for the same limited resources (e.g., memory, CPU time, or I/O bandwidth), it can lead to contention and significantly impact the performance of the application.

6. Correctness and consistency: Ensuring that concurrent operations maintain the correctness and consistency of shared data structures can be challenging, as different threads may have different views of the shared state at any given time.

7. Complexity: Concurrent programming often introduces additional complexity to the code, making it harder to read, understand, and maintain. Debugging and testing concurrent code can also be more challenging due to the non-deterministic nature of thread execution.

Context switching

CAS

• CAS takes three arguments: memory location, expected value and new value. The algorithm compare the value at the memory location with the expected value; if they match, the new value is written to the memory location. The entire operation is atomic, meaning it cannot be interrupted by other threads, ensuring consistency and correctness in concurrent environments.

• Here's a high-level description of the CAS algorithm:
1. Read the current value at the memory location.
2. Compare the current value with the expected value.
3. If the values match, update the memory location with the new value.
4. Return a boolean indicating whether the operation was successful.

Deadlock Demo

public class DeadLockDemo {

    private static final Object HAIR_A = new Object();
    private static final Object HAIR_B = new Object();

    public static void main(String[] args) {
        new Thread(() -> {
            synchronized (HAIR_A) {
                try {
                    Thread.sleep(50L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                synchronized (HAIR_B) {
                    System.out.println("A successfully grabbed B's hair");
                }
            }
        }).start();

        new Thread(() -> {
            synchronized (HAIR_B) {
                synchronized (HAIR_A) {
                    System.out.println("B successfully grabbed A's hair");
                }
            }
        }).start();
    }
}

1. Thread 1 acquires the lock on HAIR_A by entering the synchronized (HAIR_A) block.

2. Thread 1 sleeps for 50 milliseconds to simulate some processing time.

3. Meanwhile, Thread 2 acquires the lock on HAIR_B by entering the synchronized (HAIR_B) block.

4. Thread 1 wakes up from sleep and tries to acquire the lock on HAIR_B by entering the synchronized (HAIR_B) block. However, it can't proceed because Thread 2 holds the lock on HAIR_B.

5. At the same time, Thread 2 tries to acquire the lock on HAIR_A by entering the synchronized (HAIR_A) block. It can't proceed because Thread 1 holds the lock on HAIR_A.

6. As a result, both threads are stuck waiting for each other to release the respective locks, and neither can proceed. This situation leads to a deadlock.

Thread Safety

import java.util.concurrent.CountDownLatch;

public class UnSafeThread {
    private static int num = 0;
    private static CountDownLatch countDownLatch = new CountDownLatch(10);

    /**
     * Increment num by 1
     */
    public static void increment() {
        num++;
    }

    public static void main(String[] args) {
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                for (int j = 0; j < 100; j++) {
                    increment();
                    try {
                        Thread.sleep(10);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                countDownLatch.countDown();
            }).start();
        }

        while (true) {
            if (countDownLatch.getCount() == 0) {
                System.out.println(num);
                break;
            }
        }
    }
}

1. The main method creates 10 threads in a loop.

2. Each thread acquires a "virtual" lock on the shared num variable by entering the increment() method (although no actual synchronization mechanism is used).

3. Each thread increments the num variable by 1.

4. Each thread sleeps for 10 milliseconds to simulate some processing time.

5. Meanwhile, other threads are being created and run concurrently, also incrementing the num variable and sleeping for 10 milliseconds.

6. As the threads access the shared num variable without any synchronization, race conditions occur, causing the final value of num to be unpredictable.

7. After each thread completes its execution, it calls the countDown() method on the CountDownLatch.

8. The main thread is continuously checking the count of the CountDownLatch in a while loop.

9. When the count of the CountDownLatch reaches 0, indicating that all threads have completed their execution, the main thread breaks out of the while loop.

10. The main thread prints the final value of the num variable, which is the result of all threads' concurrent operations.

Basic Thread knowledge

Process vs Thread

Process:

Thread:

Summary

Thread states

1. New: The thread is created but not yet started.

2. Runnable (or Running): The thread might be running or ready to run at any instant of time. It's the responsibility of the thread scheduler to give the thread time to run.

3. Blocked (or Waiting): The thread is alive, but currently not eligible to run due to awaiting some external processing such as file I/O to complete.

4. Waiting: A thread is in a blocked state while it waits for another thread to perform a particular action.

5. Timed Waiting: A thread is in the timed waiting state due to calling a method with a specified time limit.

6. Terminated (or Dead): A thread is in a terminated state when its run() method exits.

stateDiagram
    [*] --> New: Thread created
    New --> Runnable: Thread started
    state fork_state <<fork>>
    Runnable --> fork_state
    fork_state --> Running: Scheduler selects thread
    fork_state --> Blocked: Awaits resource
    fork_state --> Waiting: Awaits another thread
    fork_state --> TimedWaiting: Thread sleeps
    Running --> Dead: run() method completes
    Dead --> [*]
    Blocked --> Runnable: Resource available
    Waiting --> Runnable: Other thread signals
    TimedWaiting --> Runnable: Sleep time expires or gets interrupted

Creating new thread

public class MyRunnable implements Runnable {
    @Override
    public void run() {
        // code to execute in the new thread
        for (int i = 0; i < 5; i++) {
            System.out.println(Thread.currentThread().getId() +" Value "+ i);
            try {
                // Sleep for a while
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

public class Main {
    public static void main(String[] args) {
        Thread thread1 = new Thread(new MyRunnable());
        thread1.start(); // starts the thread

        Thread thread2 = new Thread(new MyRunnable());
        thread2.start(); // starts the second thread
    }
}

// Thread pool method
public class Main {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(2); // creating a pool of 2 threads
        Runnable worker = new MyRunnable();
        executor.execute(worker); // calling execute method of ExecutorService
        executor.execute(worker);
        executor.shutdown(); // shut down executor
        while (!executor.isTerminated()) {
        }
        System.out.println("Finished all threads");
    }
}

1. In the main method of the Main class, we create an ExecutorService with a fixed thread pool of 2 threads using Executors.newFixedThreadPool(2).

2. Then we create an instance of MyRunnable, which we pass to the execute method of the ExecutorService. This method starts a new thread and calls the run method of MyRunnable in that thread.

3. The shutdown method is called on the ExecutorService to stop accepting new tasks and to allow previously submitted tasks to complete.

4. The while loop checks if all tasks have completed execution.

5. Once all threads have completed execution, it prints "Finished all threads".

Thread suspension and recover

public class SharedObject {
    private String data = null;

    public synchronized void produce(String data) {
        while (this.data != null) {
            try {
                wait(); // suspend the thread if data already exists
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
        this.data = data;
        notifyAll(); // notify all waiting threads
    }

    public synchronized String consume() {
        while (this.data == null) {
            try {
                wait(); // suspend the thread if no data to consume
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
        String data = this.data;
        this.data = null;
        notifyAll(); // notify all waiting threads
        return data;
    }
}

public class Producer implements Runnable {
    private final SharedObject sharedObject;

    public Producer(SharedObject sharedObject) {
        this.sharedObject = sharedObject;
    }

    @Override
    public void run() {
        for (int i = 0; i < 10; i++) {
            sharedObject.produce("Data " + i);
        }
    }
}

public class Consumer implements Runnable {
    private final SharedObject sharedObject;

    public Consumer(SharedObject sharedObject) {
        this.sharedObject = sharedObject;
    }

    @Override
    public void run() {
        for (int i = 0; i < 10; i++) {
            System.out.println("Consumed: " + sharedObject.consume());
        }
    }
}

public class Main {
    public static void main(String[] args) {
        SharedObject sharedObject = new SharedObject();
        new Thread(new Producer(sharedObject)).start();
        new Thread(new Consumer(sharedObject)).start();
    }
}

Thread Interruption

public class MyRunnable implements Runnable {
    @Override
    public void run() {
        while (!Thread.currentThread().isInterrupted()) {
            // Long running operation here.
            System.out.println("Thread is running...");
            try {
                Thread.sleep(1000); // Simulating a long-running operation.
            } catch (InterruptedException e) {
                // Thread was interrupted during sleep.
                Thread.currentThread().interrupt(); // Set the interrupt flag.
            }
        }
        System.out.println("Thread was interrupted and is stopping...");
    }
}

public class Main {
    public static void main(String[] args) {
        Thread thread = new Thread(new MyRunnable());
        thread.start(); // Start the thread.

        try {
            Thread.sleep(5000); // Wait for 5 seconds
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        thread.interrupt(); // Interrupt the thread.
    }
}

Thread Priority

public class HighPriorityThread extends Thread {
    public HighPriorityThread(String name) {
        super(name);
        setPriority(MAX_PRIORITY); // Set this thread to high priority
    }

    @Override
    public void run() {
        for (int i = 0; i < 50; i++) {
            System.out.println(getName() + " is running, count: " + i);
        }
    }
}

public class LowPriorityThread extends Thread {
    public LowPriorityThread(String name) {
        super(name);
        setPriority(MIN_PRIORITY); // Set this thread to low priority
    }

    @Override
    public void run() {
        for (int i = 0; i < 50; i++) {
            System.out.println(getName() + " is running, count: " + i);
        }
    }
}

public class Main {
    public static void main(String[] args) {
        HighPriorityThread highPriorityThread = new HighPriorityThread("HighPriorityThread");
        LowPriorityThread lowPriorityThread = new LowPriorityThread("LowPriorityThread");

        lowPriorityThread.start();
        highPriorityThread.start();
    }
}

Thread Safety

1. Read the current value of num.

2. Increment the value by one.

3. Write the new value back to num.

1. Thread A reads the value of num (let's say it's 0).

2. Thread B reads the value of num (still 0).

3. Thread A increments the value and writes it back to num (now it's 1).

4. Thread B increments the value it read earlier (0) and writes it back to num.

public class Counter {
    private int num = 0;

    public synchronized void increment() {
        num++;
    }
}

Synchronized In-depth

1. Synchronized Method: When a method is declared with the synchronized keyword, it's known as a synchronized method. Only one thread can access a synchronized method at a time. If a thread is in a synchronized method, all other threads that want to call any of the object's synchronized methods have to wait.

public synchronized void increment() {
    num++;
}

2. Synchronized Block: Synchronized blocks in Java are marked with the synchronized keyword. A synchronized block can be used to mark a specific section of the method as synchronized, rather than the whole method. This can help to improve performance because only the necessary part of the code is synchronized.

public void increment() {
    synchronized(this) {
        num++;
    }
}

In this example, the synchronized keyword is followed by an object in parentheses. This object is called a monitor object. The code inside the synchronized block is only accessible to a single thread that has obtained the lock on the monitor object.

• Granularity: Synchronized methods synchronize the entire method and prevent multiple threads from executing the method concurrently. On the other hand, synchronized blocks provide more granular control over the code. Only the code present inside the block is synchronized.

• Performance: Since synchronized blocks have a smaller scope compared to synchronized methods, they can lead to better performance in multithreaded applications.

• Flexibility: Synchronized blocks are more flexible than synchronized methods because they allow you to choose which object's lock will be used for synchronization. This way, you can control the level of concurrency in your program more precisely. When you synchronize on this, you are using the object's intrinsic lock. But you can also use other objects for synchronization. This can help to avoid unnecessary contention if different synchronized blocks are independent and don't need to be locked by the same object.

Thread and Singleton

Hunger Singleton

Lazy Singleton

public class Singleton {
    private static Singleton instance;
    
    private Singleton() {}

    public static Singleton getInstance() {
        if (instance == null) {
            instance = new Singleton();
        }
        return instance;
    }
}

public class Singleton {
    private static Singleton instance;
    
    private Singleton() {}

    public static synchronized Singleton getInstance() {
        if (instance == null) {
            instance = new Singleton();
        }
        return instance;
    }
}

public class Singleton {
    private static volatile Singleton instance;
    
    private Singleton() {}

    public static Singleton getInstance() {
        if (instance == null) {
            synchronized (Singleton.class) {
                if (instance == null) {
                    instance = new Singleton();
                }
            }
        }
        return instance;
    }
}

How to prevent thread safety problem

Main cause for thread safety problems

• multi-threading environment

• multiple thread accessing same resources

• perform non-atomic operation to shared resources

How to prevent

• multi-threading environment
• Change from multi-threading to single-thread —> add lock to access

• multiple thread accessing same resources
• Do not share resources

• perform non-atomic operation to shared resources
• change non-atomic operation to atomic —> Use lock or JDK’s own atomic packages

Lock

Different type of locks

1. Spinlock: When the time to access shared resources is short, and the frequency of context switching is high, it is not worth suspending and resuming threads. JVM implements a method to prevent the thread from being suspended when it does not acquire the lock and instead executes an empty loop. After looping a few times, if the lock is still not obtained, the thread is suspended.

2. Blocking lock: This type of lock changes the running state of the thread, causing the thread to enter a blocked state and wait. When the corresponding signal is received (either a wake-up signal or a timer signal), the thread can enter a ready state. All threads in the ready state compete to enter the running state.

3. Reentrant lock: A lock that supports a thread re-entering a lock, similar to having a key to a room that allows multiple entries.

4. Read-write lock: Two locks, a read lock and a write lock. Write-write operations are mutually exclusive, read-write operations are mutually exclusive, and read-read operations can be shared.

5. Mutex lock: Similar to going to the bathroom, once you enter the door, you close the door and do not allow others to enter.

6. Pessimistic lock: Always assumes the worst case scenario. Each time data is accessed, it is assumed that others will modify it, so a lock is applied each time data is accessed. This way, others wanting to access this data will be blocked until they acquire the lock.

7. Optimistic lock: Each time data is accessed, it is assumed that others will not modify it, so a lock is not applied. However, when updating, it checks whether others have updated this data in the meantime. This can be achieved using mechanisms such as versioning.

8. Fair lock: Everyone lines up in an orderly manner, which is fair to everyone.

9. Non-fair lock: Some people are in line, but newcomers may cut in line.

10. Biased lock: Biased locks use a mechanism that releases the lock only when competition occurs. Therefore, when other threads try to compete for the biased lock, the thread holding the biased lock will release the lock.

11. Exclusive lock: In the exclusive lock mode, only one thread can hold the lock at a time.

12. Shared lock: Allows multiple threads to acquire the lock simultaneously, allowing concurrent access to shared resources.

Simple implementation of Reentrant lock

public class SimpleReentrantLock {

    private Thread owner = null;
    private int lockCount = 0;

    public synchronized void lock() throws InterruptedException {
        Thread currentThread = Thread.currentThread();

        while (owner != null && owner != currentThread) {
            wait();
        }

        owner = currentThread;
        lockCount++;
    }

    public synchronized void unlock() {
        Thread currentThread = Thread.currentThread();

        if (currentThread == owner) {
            lockCount--;

            if (lockCount == 0) {
                owner = null;
                notify();
            }
        }
    }

    public static void main(String[] args) {
        SimpleReentrantLock lock = new SimpleReentrantLock();

        Runnable r = () -> {
            try {
                System.out.println("Thread starting: " + Thread.currentThread().getName());
                lock.lock();
                lock.lock(); // reentrant, should not block
                System.out.println("Thread inside critical section: " + Thread.currentThread().getName());
                Thread.sleep(1000);
                lock.unlock();
                lock.unlock(); // reentrant, should not throw an exception
                System.out.println("Thread ending: " + Thread.currentThread().getName());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };

        Thread t1 = new Thread(r);
        Thread t2 = new Thread(r);

        t1.start();
        t2.start();
    }
}

Brief Introduction to AQS

1. State: AQS uses an integer to represent the state of the synchronizer (for a lock, this could represent whether it's locked or not). You can change the state using methods like setState(int newState), compareAndSetState(int expect, int update), and getState().

2. Acquire: When a thread wants to acquire a lock or change the state of the synchronizer, it calls one of the acquire methods (like acquire(int arg)). If the state change isn't immediately available (for example, if the lock is held by another thread), the acquire method adds the thread to the wait queue and parks it.

3. Release: When a thread is done with the state or lock, it calls one of the release methods (like release(int arg)). The release method will change the state and if there are any waiting threads, it will unpark the first one in the queue.

4. Condition Objects: AQS also provides a ConditionObject inner class that you can use to implement condition variables, which allow a thread to voluntarily give up a lock until a certain condition is met.

ReentrantReadWriteLock

import java.util.concurrent.locks.ReentrantReadWriteLock;

public class Counter {
    private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
    private int count = 0;

    // use read lock for getting the value
    public int getCount() {
        lock.readLock().lock();
        try {
            return count;
        } finally {
            lock.readLock().unlock();
        }
    }

    // use write lock for updating the value
    public void incrementCount() {
        lock.writeLock().lock();
        try {
            count++;
        } finally {
            lock.writeLock().unlock();
        }
    }
}

• ReentrantReadWriteLock is "reentrant" because each lock that it provides, the read lock and the write lock, can be held by the same thread more than once. That is, the thread can acquire the lock, perform some work, and then acquire the lock again (re-enter the lock) if it needs to perform more work. When the thread is done, it must release the lock as many times as it acquired it.

• Also, ReentrantReadWriteLock provides a fairness policy, which can be either 'fair' or 'unfair'. A fair lock favors granting access to the longest-waiting thread. In contrast, an unfair lock makes no guarantees about the order in which threads are granted access. By default, ReentrantReadWriteLock is unfair, which generally has higher throughput but less predictability.

Thread Communication

Wait, notify, notifyAll

public class Shared {
    private String data;
    private boolean empty = true;

    public synchronized void produce(String data) {
        while (!empty) {
            try {
                wait();
            } catch (InterruptedException e) { /* handle exception */ }
        }
        this.data = data;
        empty = false;
        notifyAll();
    }

    public synchronized String consume() {
        while (empty) {
            try {
                wait();
            } catch (InterruptedException e) { /* handle exception */ }
        }
        empty = true;
        notifyAll();
        return data;
    }
}

public class Producer implements Runnable {
    private Shared shared;

    public Producer(Shared shared) {
        this.shared = shared;
    }

    @Override
    public void run() {
        for (int i = 0; i < 10; i++) {
            shared.produce("Data-" + i);
        }
    }
}

public class Consumer implements Runnable {
    private Shared shared;

    public Consumer(Shared shared) {
        this.shared = shared;
    }

    @Override
    public void run() {
        for (int i = 0; i < 10; i++) {
            System.out.println(shared.consume());
        }
    }
}

public class Main {
    public static void main(String[] args) {
        Shared shared = new Shared();

        Thread producerThread = new Thread(new Producer(shared));
        Thread consumerThread = new Thread(new Consumer(shared));

        producerThread.start();
        consumerThread.start();
    }
}

• The Shared class represents a shared object that holds the data being passed between the producer and the consumer. It has two methods, produce and consume, which are synchronized to ensure that they can't be called concurrently by different threads.

• The produce method waits until the shared object is empty, then it produces some data, stores it in the shared object, and calls notifyAll to wake up any threads that are waiting for the shared object to be filled.

• The consume method waits until the shared object is filled, then it consumes the data, empties the shared object, and calls notifyAll to wake up any threads that are waiting for the shared object to be emptied.

• The Producer and Consumer classes implement Runnable and represent the producer and consumer threads, respectively. Each thread has a reference to the shared object and repeatedly calls produce or consume.

• In the main method, a Shared object, a Producer thread and a Consumer thread are created, and the threads are started. The producer and consumer work concurrently, passing data back and forth through the shared object.

Pipelining

import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;

public class PipeExample {
    public static void main(String[] args) throws IOException {
        final PipedOutputStream output = new PipedOutputStream();
        final PipedInputStream input = new PipedInputStream(output);

        Thread writerThread = new Thread(() -> {
            try {
                output.write("Hello, world!".getBytes());
                output.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        });

        Thread readerThread = new Thread(() -> {
            try {
                int data = input.read();
                while (data != -1) {
                    System.out.print((char) data);
                    data = input.read();
                }
                input.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        });

        writerThread.start();
        readerThread.start();
    }
}

1. We create a PipedOutputStream (output) and a PipedInputStream (input) connected to the output.

2. The writerThread writes a string into the output stream.

3. The readerThread reads from the input stream, which is connected to the output stream. It reads byte by byte until it gets -1, which signifies the end of the stream, and prints each byte as a char to the console.

4. We start both threads. Depending on the scheduler and the timing, you may sometimes see no output because the readerThread starts before writerThread has had a chance to write anything. To avoid this, you could add some kind of synchronization, or you could make readerThread wait for writerThread to complete using Thread.join().

PipedInputSteam and PipedOutputStream Limitations

1. Blocking behavior: Both reading and writing to a pipe are blocking operations. If you try to read from a PipedInputStream and there's no data, the thread will block until some data is written. Similarly, if the PipedOutputStream's internal buffer is full (because the data hasn't been read yet), a thread trying to write will block until some data is read and space becomes available in the buffer.

2. Buffer size limitation: PipedInputStream/PipedOutputStream uses a relatively small fixed-size buffer (the default size is 1KB in Java 8). If you need to pass large amounts of data between threads, or if the producer thread can generate data much faster than the consumer thread can process it, you might run into problems with the buffer filling up.

3. Need for proper synchronization: These classes require careful synchronization to avoid data corruption or other problems. For example, you should not attempt to use the same PipedInputStream or PipedOutputStream from multiple threads without proper synchronization.

4. Deadlock risk: There's a risk of deadlock if a thread attempts to read and write the same pipe without proper control over when and where data is being read or written.

5. IOException on Thread Interruption: A thread blocked on PipedInputStream.read() or PipedOutputStream.write() will throw an IOException if interrupted. So if your code uses interruption to cancel tasks or shut down threads, you'll need to handle this properly.

6. One-to-one pairing: Each PipedOutputStream must be connected to exactly one PipedInputStream and vice versa. If you need to distribute data from one thread to multiple consumers, or collect data from multiple producers into one thread, you'll need to set up multiple pipes and manage them carefully.

Thread.join

Thread thread1 = new Thread(() -> {
    System.out.println("Start thread1.");
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("End thread1.");
});

Thread thread2 = new Thread(() -> {
    System.out.println("Start thread2.");
    try {
        thread1.join();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("End thread2.");
});

thread1.start();
thread2.start();

ThreadLocal

public class Example {
    private static ThreadLocal<Integer> threadId
       = new ThreadLocal<Integer>() {
           @Override protected Integer initialValue() {
               return (int) (Math.random() * 100);
           }
    };

    public static void main(String[] args) {
        for(int i=0; i<10; i++){
            new Thread(() -> {
                try {
                    System.out.println("Thread's ID: " + threadId.get());
                } finally {
                    threadId.remove();
                }
            }).start();
        }
    }
}

Condition

import java.util.LinkedList;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class ProducerConsumerExample {
    LinkedList<Integer> buffer = new LinkedList<>();
    int capacity = 4;
    ReentrantLock lock = new ReentrantLock();
    Condition notEmpty = lock.newCondition();
    Condition notFull = lock.newCondition();

    // Producer class
    class Producer implements Runnable {
        public void run() {
            try {
                int value = 0;
                while (true) {
                    lock.lock();
                    try {
                        while (buffer.size() == capacity) {
                            notFull.await();
                        }
                        System.out.println("Producer produced-" + value);
                        buffer.add(value++);
                        notEmpty.signal();
                    } finally {
                        lock.unlock();
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    // Consumer class
    class Consumer implements Runnable {
        public void run() {
            try {
                while (true) {
                    lock.lock();
                    try {
                        while (buffer.isEmpty()) {
                            notEmpty.await();
                        }
                        int value = buffer.removeFirst();
                        System.out.println("Consumer consumed-" + value);
                        notFull.signal();
                    } finally {
                        lock.unlock();
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    public void start() {
        new Thread(new Producer()).start();
        new Thread(new Consumer()).start();
    }

    public static void main(String[] args) {
        new ProducerConsumerExample().start();
    }
}

• The Condition instances notEmpty and notFull are used to manage the state of the buffer.

• When the buffer is full, the producer calls notFull.await() to wait until space is available. When the consumer removes an item from the buffer, it calls notFull.signal() to notify the producer that space is available.

• Similarly, when the buffer is empty, the consumer calls notEmpty.await() to wait until an item is available. When the producer adds an item to the buffer, it calls notEmpty.signal() to notify the consumer that an item is available.

Atomic

import java.util.concurrent.atomic.AtomicInteger;

public class AtomicExample {
    private AtomicInteger counter = new AtomicInteger(0);

    public void incrementCounter() {
        counter.incrementAndGet();
    }

    public int getCounter() {
        return counter.get();
    }
}

import java.util.concurrent.atomic.LongAccumulator;

public class AccumulatorExample {
    private LongAccumulator accumulator = new LongAccumulator(Long::max, Long.MIN_VALUE);

    public void accumulate(long value) {
        accumulator.accumulate(value);
    }

    public long getAccumulator() {
        return accumulator.get();
    }
}