Synchronization mechanism for an observable object

synchronization

observable object

concurrency

thread safety

data consistency

Synchronization mechanism for an observable object

Interview Questions practice on Codemia

Over 8,000 real interview questions from top companies, searchable by company and role.

Browse interview questions

Concurrency

Introduction

An observable object allows multiple observers to subscribe to state changes and receive notifications when the object is updated. In concurrent environments, multiple threads may modify the observable's state or subscribe/unsubscribe observers simultaneously, leading to race conditions, missed notifications, or ConcurrentModificationException errors. Synchronization mechanisms — locks, concurrent collections, copy-on-write patterns, and reactive frameworks — ensure thread-safe access to both the observer list and the observable's internal state.

Thread-Safe Observable in Java

java

1import java.util.List;
2import java.util.concurrent.CopyOnWriteArrayList;
3
4public class ObservableValue<T> {
5    private volatile T value;
6    private final List<Observer<T>> observers = new CopyOnWriteArrayList<>();
7
8    public interface Observer<T> {
9        void onChanged(T newValue);
10    }
11
12    public void addObserver(Observer<T> observer) {
13        observers.add(observer);
14    }
15
16    public void removeObserver(Observer<T> observer) {
17        observers.remove(observer);
18    }
19
20    public void setValue(T newValue) {
21        this.value = newValue;
22        // CopyOnWriteArrayList is safe to iterate during modification
23        for (Observer<T> observer : observers) {
24            observer.onChanged(newValue);
25        }
26    }
27
28    public T getValue() {
29        return value;
30    }
31}

CopyOnWriteArrayList creates a new copy of the internal array on every write (add/remove). Iteration is lock-free and safe even if observers are added or removed during notification. This is ideal when reads (notifications) far outnumber writes (subscribe/unsubscribe).

Synchronized Blocks Approach

java

1import java.util.ArrayList;
2import java.util.List;
3
4public class SynchronizedObservable<T> {
5    private T value;
6    private final List<Observer<T>> observers = new ArrayList<>();
7    private final Object lock = new Object();
8
9    public interface Observer<T> {
10        void onChanged(T newValue);
11    }
12
13    public void addObserver(Observer<T> observer) {
14        synchronized (lock) {
15            observers.add(observer);
16        }
17    }
18
19    public void removeObserver(Observer<T> observer) {
20        synchronized (lock) {
21            observers.remove(observer);
22        }
23    }
24
25    public void setValue(T newValue) {
26        List<Observer<T>> snapshot;
27        synchronized (lock) {
28            this.value = newValue;
29            snapshot = new ArrayList<>(observers);  // snapshot under lock
30        }
31        // Notify outside the lock to prevent deadlocks
32        for (Observer<T> observer : snapshot) {
33            observer.onChanged(newValue);
34        }
35    }
36}

The key pattern: take a snapshot of the observer list inside the synchronized block, then notify outside the lock. This prevents deadlocks where an observer's callback tries to modify the observable.

ReadWriteLock for Read-Heavy Workloads

java

1import java.util.ArrayList;
2import java.util.List;
3import java.util.concurrent.locks.ReadWriteLock;
4import java.util.concurrent.locks.ReentrantReadWriteLock;
5
6public class RWLockObservable<T> {
7    private T value;
8    private final List<Observer<T>> observers = new ArrayList<>();
9    private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
10
11    public interface Observer<T> {
12        void onChanged(T newValue);
13    }
14
15    public T getValue() {
16        rwLock.readLock().lock();
17        try {
18            return value;
19        } finally {
20            rwLock.readLock().unlock();
21        }
22    }
23
24    public void addObserver(Observer<T> observer) {
25        rwLock.writeLock().lock();
26        try {
27            observers.add(observer);
28        } finally {
29            rwLock.writeLock().unlock();
30        }
31    }
32
33    public void setValue(T newValue) {
34        List<Observer<T>> snapshot;
35        rwLock.writeLock().lock();
36        try {
37            this.value = newValue;
38            snapshot = new ArrayList<>(observers);
39        } finally {
40            rwLock.writeLock().unlock();
41        }
42        for (Observer<T> obs : snapshot) {
43            obs.onChanged(newValue);
44        }
45    }
46}

ReadWriteLock allows multiple threads to read the value concurrently while ensuring exclusive access for writes. This improves throughput when getValue() is called much more frequently than setValue().

Python Thread-Safe Observable

python

1import threading
2from typing import Callable, Any
3
4class Observable:
5    def __init__(self, initial_value=None):
6        self._value = initial_value
7        self._observers: list[Callable] = []
8        self._lock = threading.Lock()
9
10    @property
11    def value(self):
12        with self._lock:
13            return self._value
14
15    @value.setter
16    def value(self, new_value):
17        with self._lock:
18            self._value = new_value
19            observers = self._observers.copy()  # snapshot
20        # Notify outside lock
21        for callback in observers:
22            callback(new_value)
23
24    def subscribe(self, callback: Callable):
25        with self._lock:
26            self._observers.append(callback)
27
28    def unsubscribe(self, callback: Callable):
29        with self._lock:
30            self._observers.remove(callback)
31
32# Usage
33counter = Observable(0)
34counter.subscribe(lambda v: print(f"Observer A: {v}"))
35counter.subscribe(lambda v: print(f"Observer B: {v}"))
36counter.value = 42
37# Observer A: 42
38# Observer B: 42

Reactive Streams (Modern Approach)

java

1// Using RxJava BehaviorSubject (thread-safe by default)
2import io.reactivex.rxjava3.subjects.BehaviorSubject;
3
4BehaviorSubject<Integer> subject = BehaviorSubject.createDefault(0);
5
6// Subscribe from different threads — RxJava handles synchronization
7subject.observeOn(Schedulers.io())
8    .subscribe(value -> System.out.println("Observer: " + value));
9
10subject.onNext(42);  // thread-safe

python

1# Using RxPY
2from reactivex import subject
3
4s = subject.BehaviorSubject(0)
5s.subscribe(lambda v: print(f"Value: {v}"))
6s.on_next(42)  # Value: 42

Reactive frameworks like RxJava and RxPY encapsulate synchronization internally. Subscribers receive notifications on designated schedulers, eliminating manual lock management. This is the preferred approach for complex event-driven systems.

Common Pitfalls

Notifying observers inside a synchronized block: If an observer's callback modifies the observable (adds/removes observers or changes the value), notifying inside the lock causes a deadlock or ConcurrentModificationException. Always take a snapshot of the observer list inside the lock and notify outside it.
Using ArrayList without synchronization: Iterating over an ArrayList while another thread modifies it throws ConcurrentModificationException. Use CopyOnWriteArrayList, or synchronize and snapshot before iterating.
Holding locks during long-running callbacks: If observer callbacks perform I/O, network calls, or heavy computation, holding the lock during notification blocks all other threads from accessing the observable. Snapshot-and-notify-outside-lock is the standard pattern to avoid this.
Forgetting volatile or locks for the value field: Without volatile or synchronization, changes to the observable's value may not be visible to other threads due to CPU cache coherency. Mark the field volatile or always read/write it under a lock.
Observer memory leaks: If observers hold strong references and are never unsubscribed, the observable prevents garbage collection of observer objects. Use weak references (WeakReference in Java, weakref in Python) or ensure explicit unsubscription when observers are no longer needed.

Summary

Use CopyOnWriteArrayList when subscribe/unsubscribe is rare but notifications are frequent
Use synchronized blocks with snapshot-and-notify-outside-lock for general-purpose thread safety
Use ReadWriteLock when reads (getValue) far outnumber writes (setValue)
Reactive frameworks (RxJava, RxPY) handle synchronization internally and are preferred for complex scenarios
Always notify observers outside the lock to prevent deadlocks
Use volatile or locks for the observable's value field to ensure cross-thread visibility

Tackling System Design Interview Problems

A short course that equips you with the skills to approach system design interviews methodically.

Start the free course

Track what you have practised

A free account saves your progress, solutions and study plan across every problem on Codemia.

Interview Questions practice on Codemia

Over 8,000 real interview questions from top companies, searchable by company and role.

Browse interview questions

Synchronization mechanism for an observable object

Interview Questions practice on Codemia

Introduction

Thread-Safe Observable in Java

Synchronized Blocks Approach

ReadWriteLock for Read-Heavy Workloads

Python Thread-Safe Observable

Reactive Streams (Modern Approach)

Common Pitfalls

Summary

Related reading

Tackling System Design Interview Problems

Track what you have practised

Interview Questions practice on Codemia