How do I use BlockingQueue to pass data between threads?

To use a BlockingQueue to pass data between threads in Java, you can follow these steps:

1. Understand BlockingQueue

BlockingQueue is part of the java.util.concurrent package and is designed for thread-safe communication between producer and consumer threads. It provides methods such as put() and take(), which handle blocking behavior:

  • put(E e): Blocks if the queue is full until space becomes available.
  • take(): Blocks if the queue is empty until an element becomes available.

Common implementations of BlockingQueue include:

  • ArrayBlockingQueue: A fixed-capacity, bounded queue.
  • LinkedBlockingQueue: A linked-node queue, optionally bounded.
  • PriorityBlockingQueue: A priority-based queue (does not block on offer/add).
  • SynchronousQueue: A queue with no capacity, where put blocks until a take occurs (and vice versa).

2. Example Setup for Producer-Consumer Pattern

Here’s an example to show how to use a BlockingQueue to pass data between producer and consumer threads:

Code Example

package org.kodejava.util.concurrent;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class BlockingQueueDemo {
    private static final int QUEUE_CAPACITY = 5;

    public static void main(String[] args) {
        // Instantiate a BlockingQueue with a capacity of 5
        BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY);

        // Create producer and consumer threads
        Thread producer = new Thread(new Producer(queue));
        Thread consumer = new Thread(new Consumer(queue));

        // Start threads
        producer.start();
        consumer.start();
    }
}

class Producer implements Runnable {
    private final BlockingQueue<Integer> queue;

    public Producer(BlockingQueue<Integer> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            for (int i = 1; i <= 10; i++) {
                System.out.println("Produced: " + i);
                queue.put(i); // Add item to the queue, blocks if full
                Thread.sleep(500); // Simulate production time
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

class Consumer implements Runnable {
    private final BlockingQueue<Integer> queue;

    public Consumer(BlockingQueue<Integer> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            while (true) {
                Integer item = queue.take(); // Take item from the queue, blocks if empty
                System.out.println("Consumed: " + item);
                Thread.sleep(1000); // Simulate consumption time
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

3. How It Works

  1. BlockingQueue:
    • A LinkedBlockingQueue with a capacity of 5 is created.
    • The producer thread calls queue.put(i) to add elements to the queue. If the queue is full, the thread blocks until space becomes available.
    • The consumer thread calls queue.take() to retrieve elements. If the queue is empty, the thread blocks until an item is added.
  2. Producer:
    • Produces data (e.g., numbers) and adds them to the queue.
    • The put method ensures thread safety and blocks automatically when the queue is full.
  3. Consumer:
    • Retrieves data from the queue and processes it.
    • The take method ensures thread safety and blocks automatically when the queue is empty.

4. Key Points

  • Thread Safety: BlockingQueue handles all necessary synchronization internally.
  • Automatic Blocking:
    • put() in the producer blocks if the queue is full.
    • take() in the consumer blocks if the queue is empty.
  • Stopping Mechanism:
    • In real-world applications, implement a stopping mechanism such as adding a “poison pill” (special object) to signal termination.

Example of Poison Pill:

// Add poison pill to queue after all items are produced
queue.put(-1);

// Consumer stops processing when it encounters the poison pill
if (item == -1) break;

5. Execution Output

If you execute the above example, the producer and consumer will work concurrently, producing and consuming items in a thread-safe manner. Sample output:

Produced: 1
Consumed: 1
Produced: 2
Consumed: 2
Produced: 3
Produced: 4
Consumed: 3
Produced: 5

This demonstrates how BlockingQueue effectively facilitates communication between threads.

What is ConcurrentHashMap and how do I use it in Java?

The ConcurrentHashMap is a class in Java that implements the ConcurrentMap interface. It is part of the Java Collection Framework and extends the AbstractMap class.

ConcurrentHashMap is thread-safe, which means it is designed to support high concurrency levels by handling multiple threads concurrently without any inconsistencies. It allows multiple threads to perform retrieve (get) and update (insert & delete) operations. Internally, ConcurrentHashMap uses concepts of Segmentation to store data which allows higher degree of concurrency.

Here is an example of how to use ConcurrentHashMap in Java:

package org.kodejava.util;

import java.util.concurrent.ConcurrentHashMap;

public class ConcurrentHashMapExample {
    public static void main(String[] args) {
        // Create a ConcurrentHashMap instance
        ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();

        // Add elements
        map.put("One", 1);
        map.put("Two", 2);
        map.put("Three", 3);

        // Retrieve elements
        Integer one = map.get("One");
        System.out.println("Retrieved value for 'One': " + one);

        // Remove an element
        map.remove("Two");

        // Print all elements
        map.forEach((key, value) -> System.out.println(key + " = " + value));
    }
}

Output:

Retrieved value for 'One': 1
One = 1
Three = 3

In this example, we’re creating a ConcurrentHashMap, adding some elements to it, retrieving an element, removing an element, and finally printing all the elements.

One thing to note is that while ConcurrentHashMap allows multiple threads to read and write concurrently, a get() operation might not reflect the latest put() operation, since it might be looking at a previous segment. Further thread synchronization mechanisms might be necessary depending on your exact use case.

Also, worth mentioning, null values and null keys are not permitted in ConcurrentHashMap to prevent ambiguities and potential errors in multithreaded contexts. If you try to use null, ConcurrentHashMap will throw a NullPointerException.

Here’s an example demonstrating the usage of ConcurrentHashMap in a multithreaded context:

package org.kodejava.util;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class ConcurrentHashMapThreadDemo {
    public static void main(String[] args) throws InterruptedException {
        ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();

        // Create a ThreadPool with 5 threads
        try (ExecutorService executor = Executors.newFixedThreadPool(5)) {

            // Runnable task to increment a value in the map
            Runnable task = () -> {
                for (int i = 0; i < 10; i++) {
                    map.compute("TestKey", (key, value) -> {
                        if (value == null) {
                            return 1;
                        } else {
                            return value + 1;
                        }
                    });
                }
            };

            // Submit the task to each thread in the pool
            for (int i = 0; i < 5; i++) {
                executor.submit(task);
            }

            // Shut down the executor and wait for tasks to complete
            executor.shutdown();
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        }

        System.out.println("Final value for 'TestKey': " + map.get("TestKey"));
    }
}

Output:

Final value for 'TestKey': 50

In this example, we’re creating a ConcurrentHashMap and a thread pool with ExecutorService. We’re then defining a Runnable task, which increments the value of the “TestKey” key in the map 10 times.

The task uses ConcurrentHashMap‘s compute() method, which is atomic, meaning that the retrieval and update of the value is done as a single operation that cannot be interleaved with other operations. We then submit the task to each of the five threads in our thread pool. After all threads have completed their tasks, we retrieve and print the final value of “TestKey”.

If everything works correctly, the output should be “Final value for ‘TestKey’: 50”, because we have 5 threads each incrementing the value 10 times. This demonstrates the thread-safety of ConcurrentHashMap, as the compute() operation is done atomically and many threads were able to modify the map simultaneously without causing inconsistencies. If we were using a plain HashMap instead, we could not guarantee this would be the case.

How do I check whether a thread group has been destroyed?

You can use ThreadGroup.isDestroyed() method to check whether a thread group and its subgroups has been destroyed.

package org.kodejava.lang;

public class CheckGroupDestroy {
    public static void main(String[] args) {
        ThreadGroup grandParent = new ThreadGroup("GrandParent");
        ThreadGroup uncle = new ThreadGroup(grandParent, "Uncle");
        ThreadGroup parent = new ThreadGroup(grandParent, "Parent");
        ThreadGroup son = new ThreadGroup(parent, "Son");
        ThreadGroup daughter = new ThreadGroup(parent, "Daughter");
        ThreadGroup neighbour = new ThreadGroup("Neighbour");

        ThreadGroup[] groupArray = {
                grandParent, uncle, parent, son, daughter, neighbour
        };

        // Destroy 'parent' group and all its subgroups
        parent.destroy();

        // Check whether the group is destroyed. The result is,
        // GrandParent, Uncle, and Neighbour did not destroyed
        // because they are not Parent's subgroups
        for (ThreadGroup tg : groupArray) {
            if (tg.isDestroyed()) {
                System.out.println(tg.getName() + " is destroyed");
            } else {
                System.out.println(tg.getName() + " is not destroyed");
            }
        }
    }
}

The result is:

GrandParent is not destroyed
Uncle is not destroyed
Parent is destroyed
Son is destroyed
Daughter is destroyed
Neighbour is not destroyed

How do I destroy a thread group?

You can destroy a thread group by using destroy() method of ThreadGroup class. It will cleans up the thread group and removes it from the thread group hierarchy. It’s not only destroy the thread group, but also all its subgroups.

The destroy() method is of limited use: it can only be called if there are no threads presently in the thread group.

package org.kodejava.lang;

public class ThreadGroupDestroy {
    public static void main(String[] args) {
        ThreadGroup root = new ThreadGroup("Root");
        ThreadGroup server = new ThreadGroup(root, "ServerGroup");
        ThreadGroup client = new ThreadGroup(root, "ClientGroup");

        // Destroy 'root' thread groups and all its subgroup
        // ('server' & 'client')
        root.destroy();

        // Check if 'root' group and its subgroups already destroyed
        if (root.isDestroyed()) {
            System.out.println("Root group is destroyed");
        }

        if (server.isDestroyed()) {
            System.out.println("Server group is destroyed");
        }

        if (client.isDestroyed()) {
            System.out.println("Client group is destroyed");
        }
    }
}

How do I count active thread in a thread group?

This example describe how to get the number of active threads in this thread group. The result might not reflect concurrent activity, and might be affected by the presence of certain system threads.

Due to the inherently imprecise nature of the result, it is recommended that this method only be used for informational purposes.

package org.kodejava.lang;

public class ThreadGroupActiveThread {
    public static void main(String[] args) {
        ThreadGroup threadGroup = new ThreadGroup("TestThread");

        Thread t1 = new Thread(threadGroup, new Server(), "Server1");
        Thread t2 = new Thread(threadGroup, new Server(), "Server2");
        Thread t3 = new Thread(threadGroup, new Server(), "Server3");
        Thread t4 = new Thread(threadGroup, new Server(), "Server4");

        t1.start();
        t2.start();
        t3.start();
        t4.start();

        // Get an estimate number of active thread of the thread
        // group.
        int activeThread = threadGroup.activeCount();
        System.out.format("Number of active threads of %s is %d %n",
                threadGroup.getName(), activeThread);
    }
}

class Server implements Runnable {
    public void run() {
        System.out.println("Running..");
    }
}

How do I get number of active thread in current thread?

In this example you’ll see how to obtain the number of active thread in the current thread. You can use the Thread.activeCount() method to get this information.

package org.kodejava.lang;

public class CountActiveThread {
    public static void main(String[] args) {
        Thread t = new Thread(() -> System.out.println("Hello..."));
        t.start();

        // Get the number of active threads in the current thread's
        // thread group.
        int activeThread = Thread.activeCount();
        System.out.format("Number of active threads of %s is %d %n",
                Thread.currentThread().getName(), activeThread);
    }
}

How do I get number of active thread group?

Use method activeGroupCount() of ThreadGroup class to get estimate number of active groups in the thread group and use activeCount() to get estimate number of active threads in a thread group.

package org.kodejava.lang;

public class ActiveGroupCount {
    public static void main(String[] args) {
        ThreadGroup root = new ThreadGroup("RootGroup");
        ThreadGroup server = new ThreadGroup(root, "ServerGroup");
        ThreadGroup client = new ThreadGroup(root, "ClientGroup");

        Thread t1 = new Thread(server, new ServerThread(), "ServerThread");
        Thread t2 = new Thread(client, new ClientThread(), "ClientThread");

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

        // Get estimate active groups in 'root' thread group
        int activeGroup = root.activeGroupCount();
        System.out.format("Estimated active group in %s is %d%n",
                root.getName(), activeGroup);

        // Get estimate active threads in 'root' thread group
        int activeThread = root.activeCount();
        System.out.format("Estimated active thread in %s is %d%n",
                root.getName(), activeThread);
    }
}

class ServerThread implements Runnable {
    public void run() {
        System.out.println("Running - Server Thread..");
    }
}


class ClientThread implements Runnable {
    public void run() {
        System.out.println("Running - Client Thread..");
    }
}

The example above print the following example output:

Estimated active group in RootGroup is 2
Estimated active thread in RootGroup is 2
Running - Client Thread..
Running - Server Thread..

How do I get thread group of a thread?

Use the getThreadGroup() method of Thread class to get the thread group to which the thread belongs.

package org.kodejava.lang;

public class GetThreadGroup {
    public static void main(String[] args) {
        // Create thread groups
        ThreadGroup group = new ThreadGroup("ThreadGroup");
        ThreadGroup anotherGroup = new ThreadGroup(group, "AnotherGroup");

        // Create threads and placed into thread group
        Thread t1 = new Thread(group, new FirstThread(), "Thread1");
        Thread t2 = new Thread(anotherGroup, new FirstThread(), "Thread2");

        // Start the threads
        t1.start();
        t2.start();

        // Use getThreadGroup() method of Thread class to get the object
        // of ThreadGroup then use the getName() method to get the name
        // of thread group.
        System.out.format("%s is a member of %s%n", t1.getName(),
                t1.getThreadGroup().getName());
        System.out.format("%s is a member of %s%n", t2.getName(),
                t2.getThreadGroup().getName());
    }
}

class FirstThread implements Runnable {
    public void run() {
        System.out.println("Start..");
    }
}

How do I get the state of a thread?

To get the state of a thread use getState() method of a Thread class. One thing to be noted is that this method is designed for use in monitoring of the system state, not for synchronization control.

package org.kodejava.lang;

public class GetThreadState implements Runnable {
    public static void main(String[] args) {
        Thread thread = new Thread(new GetThreadState());
        thread.start();

        // Get the state of the thread.
        Thread.State state = thread.getState();
        System.out.println("State: " + state.name());
    }

    public void run() {
        System.out.println("Start..");
    }
}

How do I create a thread synchronized block?

The objective of thread synchronization is to ensure that when several threads want access to a single resource, only one thread can access it at any given time.

You can manage synchronization of your program at method level (synchronized method) or at block level (synchronized block). To make a block of code synchronized you can use the synchronized keyword.

The example below show how you can use the synchronized keyword.

  • Incrementor class
package org.kodejava.lang;

public class Incrementor {
    private int count;

    // A synchronized method example.
    public synchronized void increment(int value) {
        count += value;
        System.out.println(Thread.currentThread().getName() +
                ": inc >>> " + count);
    }

    public void decrement(int value) {
        // A synchronized block example the use the current object instance
        // as the monitor object.
        synchronized (this) {
            count -= value;
            System.out.println(Thread.currentThread().getName() +
                    ": dec >>> " + count);
        }
    }
}
  • IncrementThread class
package org.kodejava.lang;

public class IncrementThread implements Runnable {
    private final Incrementor incrementor;

    public IncrementThread(Incrementor incrementor) {
        this.incrementor = incrementor;
    }

    public void run() {
        for (int i = 1; i <= 5; i++) {
            incrementor.increment(i * 10);
            incrementor.decrement(i * 2);

            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
  • IncrementorDemo class
package org.kodejava.lang;

public class IncrementorDemo {
    public static void main(String[] args) {
        Incrementor incrementor = new Incrementor();

        Thread t1 = new Thread(new IncrementThread(incrementor), "T1");
        Thread t2 = new Thread(new IncrementThread(incrementor), "T2");

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

Here an example result printed by the program:

T1: inc >>> 10
T1: dec >>> 8
T2: inc >>> 18
T2: dec >>> 16
T1: inc >>> 36
T1: dec >>> 32
T2: inc >>> 52
T2: dec >>> 48
T1: inc >>> 78
T1: dec >>> 72
T2: inc >>> 102
T2: dec >>> 96
T1: inc >>> 136
T1: dec >>> 128
T2: inc >>> 168
T2: dec >>> 160
T1: inc >>> 210
T1: dec >>> 200
T2: inc >>> 250
T2: dec >>> 240