Creating a Thread-Safe Channel with CopyOnWriteArrayList

Creating a Thread-Safe Messaging Channel with CopyOnWriteArrayList

Welcome back! In our last lesson, we explored the use of ConcurrentSkipListMap for a game leaderboard. Today, we're taking another step forward in concurrency by building a thread-safe communication channel using CopyOnWriteArrayList . This approach is particularly useful when you need to manage frequent reads and occasional writes, making it a perfect fit for messaging systems or notification services.

What You'll Learn

By the end of this lesson, you will: Learn how to build a thread-safe messaging channel using concurrent collections. Understand how to efficiently manage concurrent reads and writes in your application. Implement a system where multiple threads can post and read messages without additional locking mechanisms. These concepts will help you build highly responsive and thread-safe communication systems that are essential in real-world applications like live feeds or notification systems.

Recap: Thread-Safe Collections

Before we dive into the implementation, let's quickly revisit CopyOnWriteArrayList . This collection is a thread-safe variant of ArrayList where each modification (like adding or removing an item) creates a new copy of the underlying array. It shines in scenarios where reads vastly outnumber writes because reads occur directly on the array, while writes involve a fresh copy. High Read Efficiency : Since reads happen on a stable snapshot of the array, they don’t require locking, making the collection ideal for cases where multiple threads frequently read data. Copy-on-Write Behavior : Modifications such as adding or removing items result in creating a new array, ensuring that readers don’t encounter data inconsistencies during concurrent writes. This behavior provides a safe and simple way to manage concurrent access in applications that prioritize read performance.

Implementing the Channel Class

Now that we're clear on how CopyOnWriteArrayList works, let’s implement the Channel class, which will act as our messaging platform where threads can post and read messages. Java import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; public class Channel {private final List<String> messages = new CopyOnWriteArrayList<>(); The Channel class uses a CopyOnWriteArrayList to store messages. This ensures that even if multiple threads post or read messages simultaneously, they won’t interfere with each other.

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

Java

    public void postMessage(String message) {
        messages.add(message);
        System.out.println(Thread.currentThread().getName() + " posted: " + message);
    }

    public void postMessage(String message) {
        messages.add(message);
        System.out.println(Thread.currentThread().getName() + " posted: " + message);
    }

Java

    public boolean removeMessage(String message) {
        boolean removed = messages.remove(message);
        if (removed) {
            System.out.println(Thread.currentThread().getName() + " removed: " + message);
        }
        return removed;
    }

    public boolean removeMessage(String message) {
        boolean removed = messages.remove(message);
        if (removed) {
            System.out.println(Thread.currentThread().getName() + " removed: " + message);
        }
        return removed;
    }

Java

    public List<String> getMessages() {
        return messages;
    }
}

    public List<String> getMessages() {
        return messages;
    }
}

Java

public class Main {
    public static void main(String[] args) throws InterruptedException {
        Channel channel = new Channel();

        // Publisher thread
        Thread publisher = new Thread(() -> {
            channel.postMessage("Welcome!");
            channel.postMessage("Today's topic is Java Concurrency.");
        }, "Publisher");

public class Main {
    public static void main(String[] args) throws InterruptedException {
        Channel channel = new Channel();

        // Publisher thread
        Thread publisher = new Thread(() -> {
            channel.postMessage("Welcome!");
            channel.postMessage("Today's topic is Java Concurrency.");
        }, "Publisher");

Java

        // Subscriber 1 thread
        Thread subscriber1 = new Thread(() -> {
            for (String message : channel.getMessages()) {
                System.out.println("Subscriber1 reads: " + message);
            }
        }, "Subscriber1");

        // Subscriber 2 thread
        Thread subscriber2 = new Thread(() -> {
            for (String message : channel.getMessages()) {
                System.out.println("Subscriber2 reads: " + message);
            }
        }, "Subscriber2");

        // Subscriber 1 thread
        Thread subscriber1 = new Thread(() -> {
            for (String message : channel.getMessages()) {
                System.out.println("Subscriber1 reads: " + message);
            }
        }, "Subscriber1");

        // Subscriber 2 thread
        Thread subscriber2 = new Thread(() -> {
            for (String message : channel.getMessages()) {
                System.out.println("Subscriber2 reads: " + message);
            }
        }, "Subscriber2");

Java

        // Start publisher thread
        publisher.start();
        publisher.join(); // Ensures the publisher posts messages before subscribers read

        // Start subscriber threads
        subscriber1.start();
        subscriber2.start();

        subscriber1.join();
        subscriber2.join();
    }
}

        // Start publisher thread
        publisher.start();
        publisher.join(); // Ensures the publisher posts messages before subscribers read

        // Start subscriber threads
        subscriber1.start();
        subscriber2.start();

        subscriber1.join();
        subscriber2.join();
    }
}

Java

    public void postMessage(String message) {
        messages.add(message);
        System.out.println(Thread.currentThread().getName() + " posted: " + message);
    }

Java

    public boolean removeMessage(String message) {
        boolean removed = messages.remove(message);
        if (removed) {
            System.out.println(Thread.currentThread().getName() + " removed: " + message);
        }
        return removed;
    }

Java

    public List<String> getMessages() {
        return messages;
    }
}

Java

public class Main {
    public static void main(String[] args) throws InterruptedException {
        Channel channel = new Channel();

        // Publisher thread
        Thread publisher = new Thread(() -> {
            channel.postMessage("Welcome!");
            channel.postMessage("Today's topic is Java Concurrency.");
        }, "Publisher");

Java

        // Subscriber 1 thread
        Thread subscriber1 = new Thread(() -> {
            for (String message : channel.getMessages()) {
                System.out.println("Subscriber1 reads: " + message);
            }
        }, "Subscriber1");

        // Subscriber 2 thread
        Thread subscriber2 = new Thread(() -> {
            for (String message : channel.getMessages()) {
                System.out.println("Subscriber2 reads: " + message);
            }
        }, "Subscriber2");

Java

        // Start publisher thread
        publisher.start();
        publisher.join(); // Ensures the publisher posts messages before subscribers read

        // Start subscriber threads
        subscriber1.start();
        subscriber2.start();

        subscriber1.join();
        subscriber2.join();
    }
}

Creating a Thread-Safe Messaging Channel with CopyOnWriteArrayList

What You'll Learn

Recap: Thread-Safe Collections

Implementing the Channel Class

Join the 1M+ learners on CodeSignal

Posting a Message

Removing a Message

Retrieving Messages

Building the Main Class with Threads

Subscriber Threads

Thread Coordination

Why It Matters