Production-grade guide to concurrent programming patterns covering architecture patterns, implementation strategies, testing approaches, and operational best practices for enterprise engineering teams.
Concurrent programming patterns are reusable solutions for managing multiple computations executing at the same time, especially when they share resources or depend on each other. These patterns solve common problems in systems where tasks overlap in time: handling I/O-bound workloads, coordinating distributed services, managing state across threads, and reacting to events in real time. They matter when performance, scalability, and responsiveness are critical—such as in web servers, message brokers, real-time dashboards, or data processing pipelines.
When multiple threads or processes access a shared variable or data structure, race conditions can corrupt data. The goal is to ensure only one thread modifies the data at a time, or to coordinate updates efficiently.
Use a mutex (mutual exclusion lock) to protect a block of code that accesses shared state.
// Pseudocode: C-style
int counter = 0;
mutex_t lock;
void increment() {
mutex_lock(&lock);
counter += 1; // This block is atomic
mutex_unlock(&lock);
}The mutex_lock and mutex_unlock functions are typically provided by the runtime (e.g., pthread_mutex_lock, std::mutex::lock). The key failure mode is lock contention: when many threads arrive at the same time, they queue behind the mutex. The lock is not free: it has memory overhead and can be lost in context switches.
A common mistake is holding the lock too long. For example, doing a disk I/O or network call inside the critical section:
void write_to_database() {
mutex_lock(&lock);
// Write to disk: 10 ms delay
write_to_file("log.txt", "event received");
// But other threads can't enter!
mutex_unlock(&lock);
}This reduces concurrency. The solution is to split the critical section: acquire the lock, do fast work, release, then do expensive I/O.
When reads are frequent and writes are less common, use a read-write lock. It allows multiple readers to hold the lock simultaneously, but blocks all readers when a writer is active.
rwlock_t data_rwlock;
void read_data() {
rwlock_rdlock(&data_rwlock);
// Read data without blocking others
int value = data[0];
rwlock_rdunlock(&data_rwlock);
}
void update_data() {
rwlock_wrlock(&data_rwlock);
data[0] = 42;
rwlock_wrunlock(&data_rwlock);
}A silent failure occurs when a writer is blocked, but readers keep arriving. If the writer takes a long time, readers can starve. Also, writers can be writer-preferential: once a writer arrives, no new readers are allowed until the writer finishes. The pattern can be reader-preferential, where readers can enter freely, but a writer must wait for all current readers to finish.
When a system needs to execute several independent tasks and wait for all to complete, or for any one to complete, use task coordination patterns.
Launch multiple tasks and collect their results asynchronously.
// Go example: using channels and goroutines
func fetchUsers(userIDs []int) []User {
results := make(chan User, len(userIDs))
var wg sync.WaitGroup
for _, id := range userIDs {
wg.Add(1)
go func(uid int) {
defer wg.Done()
user, err := fetchUserFromAPI(uid)
if err != nil {
results <- User{ID: uid, Error: err.Error()}
} else {
results <- user
}
}(id)
}
go func() {
wg.Wait()
close(results)
}()
var users []User
for user := range results {
users = append(users, user)
}
return users
}The fetchUserFromAPI function is a remote call—each call can take hundreds of milliseconds. The chan User channel is the future: a placeholder for a value that will be available later.
A subtle failure is channel leakage: if the results channel is not closed, the for range loop in the main function never terminates. The go wg.Wait() and close(results) are essential. Also, if a task panics, it must be recovered or the main loop stops early.
For a large number of tasks, use a fixed-size pool of worker threads.
// Pseudocode: worker pool using a queue and threads
typedef struct {
task_t* tasks;
int head, tail, size;
pthread_mutex_t mutex;
pthread_cond_t not_empty;
} task_queue_t;
task_queue_t task_queue = { .size = 100 };
void* worker_thread(void* arg) {
while (1) {
pthread_mutex_lock(&task_queue.mutex);
while (task_queue.head == task_queue.tail) {
pthread_cond_wait(&task_queue.not_empty, &task_queue.mutex);
}
task_t task = task_queue.tasks[task_queue.head];
task_queue.head = (task_queue.head + 1) % task_queue.size;
pthread_mutex_unlock(&task_queue.mutex);
execute_task(task);
}
return NULL;
}
void queue_task(task_t task) {
pthread_mutex_lock(&task_queue.mutex);
task_queue.tasks[task_queue.tail] = task;
task_queue.tail = (task_queue.tail + 1) % task_queue.size;
pthread_cond_signal(&task_queue.not_empty);
pthread_mutex_unlock(&task_queue.mutex);
}The key is the condition variable (pthread_cond_t), which allows threads to wait until a task is available. If the queue is empty, pthread_cond_wait blocks the worker until a new task is queued.
A common failure: spurious wakeups. A thread wakes up from pthread_cond_wait even when no signal was sent. The solution is to wrap the while loop around pthread_cond_wait. Also, broadcast vs. signal: pthread_cond_signal wakes one thread, pthread_cond_broadcast wakes all. Use broadcast when multiple workers can process the same task type.
When a system reacts to asynchronous events—such as messages, file uploads, or HTTP requests—use event-driven coordination.
Decouple producers and consumers using a message broker.
// Message format: JSON
{
"event": "user.created",
"timestamp": "2024-07-15T10:00:00Z",
"payload": {
"id": 123,
"name": "Alice",
"email": "[email protected]"
}
}Producers publish events to a central event bus. Consumers subscribe to specific event types.
def on_user_created(event):
user = event['payload']
send_welcome_email(user['email'])
record_user_in_analytics(user['id'])
# Subscribe to all 'user.created' events
consumer = KafkaConsumer('user.created', bootstrap_servers=['kafka:9092'])
for message in consumer:
event = json.loads(message.value)
on_user_created(event)The event processing loop must be idempotent: processing the same message multiple times must have the same result. A message might be delivered multiple times due to network failures. The consumer must track offsets or message IDs.
A silent failure: offset commit failure. After processing a message, the consumer commits its offset (the position in the stream), but the commit fails before the next message is processed. If the system crashes, it restarts from the last committed offset, but the latest message is processed twice.
To handle duplicate or delayed events, design handlers to be idempotent.
-- Ensure user creation is idempotent
INSERT INTO users (id, name, email, created_at)
VALUES (123, 'Alice', '[email protected]', '2024-07-15T10:00:00Z')
ON CONFLICT (id) DO NOTHING;The ON CONFLICT (id) DO NOTHING clause ensures that reprocessing the same event does not create duplicate users. Alternatively, use UPDATE instead:
UPDATE users
SET name = 'Alice', email = '[email protected]', updated_at = NOW()
WHERE id = 123;Even better: use a deduplication key in the database.
CREATE TABLE user_events (
id SERIAL PRIMARY KEY,
event_id VARCHAR(64) UNIQUE, -- from the message ID
event_type VARCHAR(32),
payload JSONB,
processed BOOLEAN DEFAULT FALSE
);Before processing, check if the event has already been processed:
SELECT processed FROM user_events WHERE event_id = 'event-123';If processed = TRUE, skip the work. Otherwise, process and mark processed = TRUE.
When data arrives continuously—such as log entries, sensor readings, or real-time telemetry—use stream processing patterns.
When producers generate data faster than consumers can process, use a bounded queue to prevent memory overflow.
// Go: buffered channel with backpressure
type LogEntry struct {
Message string
Timestamp time.Time
}
var logQueue = make(chan LogEntry, 1000) // 1000 items max
func logWorker() {
for entry := range logQueue {
processLogEntry(entry)
// Simulate slow processing
time.Sleep(10 * time.Millisecond)
}
}
func main() {
// Start worker
go logWorker()
// Generate logs faster than processing
for i := 0; i < 10000; i++ {
logQueue <- LogEntry{
Message: fmt.Sprintf("Event %d", i),
Timestamp: time.Now(),
}
}
}The make(chan LogEntry, 1000) creates a channel with a buffer of 1000 elements. When the buffer is full, logQueue <- entry blocks until space becomes available. This is backpressure: the producer slows down as the consumer lags.
A common failure: deadlock. If the worker is not started before the producer, the logWorker goroutine may never run, and the logQueue <- entry calls block forever. The solution: start the worker before the loop, or use go logWorker().
Another failure: message loss. If the producer adds to the channel after close(logQueue), the last few messages are lost because the channel is closed but the worker is still running. The pattern must include both close(logQueue) and close on the worker’s for range.
For real-time analytics, group events into time windows.
// Stream of sensor readings
[
{ "sensor": "temp", "value": 22.1, "ts": "2024-07-15T10:00:00Z" },
{ "sensor": "temp", "value": 23.0, "ts": "2024-07-15T10:00:05Z" },
{ "sensor": "humidity", "value": 65.2, "ts": "2024-07-15T10:00:03Z" }
]Compute statistics over 10-second windows:
def process_stream(stream):
window_start = None
window_data = []
for event in stream:
if window_start is None:
window_start = event['ts']
# Check if event belongs to current window
if parse_timestamp(event['ts']) - parse_timestamp(window_start) >= 10:
# Compute average temperature
temp_values = [e['value'] for e in window_data if e['sensor'] == 'temp']
avg_temp = sum(temp_values) / len(temp_values)
emit_result(
window=window_start,
avg_temp=avg_temp
)
# Start new window
window_start = event['ts']
window_data = []
window_data.append(event)The key failure is window alignment: events may arrive at irregular intervals, and the window boundaries must be exact. A window may start at 10:00:00, but the first event is at 10:00:01, so it takes 9 seconds to fill the first 10 seconds. The solution is to use sliding windows or tumbling windows.
When multiple processes modify the same entity and must resolve conflicts, use state management patterns.
Track changes using version numbers or timestamps.
-- Table: orders
CREATE TABLE orders (
id BIGSERIAL PRIMARY KEY,
status VARCHAR(20),
version INT DEFAULT 1,
updated_at TIMESTAMPTZ
);
-- Update order: optimistic locking
UPDATE orders
SET status = 'shipped', version = version + 1, updated_at = NOW()
WHERE id = 123 AND version = 5;
-- If no rows updated, version mismatch
-- The client must retry with the new versionA common failure: lost update. Two clients read the same order (version 5), both modify it, and both update the database. The second update overwrites the first, but the version is not incremented. The first client’s update is lost.
The solution is retry-on-version-mismatch:
func updateOrder(orderID int, newStatus string) error {
for attempt := 0; attempt < 3; attempt++ {
tx, err := db.Begin()
if err != nil {
return err
}
var version int
err = tx.QueryRow(
"SELECT version FROM orders WHERE id = $1 FOR UPDATE",
orderID,
).Scan(&version)
if err != nil {
tx.Rollback()
return err
}
_, err = tx.Exec(
"UPDATE orders SET status = $1, version = version + 1, updated_at = NOW() WHERE id = $2 AND version = $3",
newStatus, orderID, version,
)
if err != nil {
tx.Rollback()
return err
}
if rowsAffected, _ := tx.RowsAffected(); rowsAffected == 0 {
tx.Rollback()
// Version mismatch: retry
continue
}
return tx.Commit()
}
return errors.New("update failed after 3 attempts")
}The pattern relies on FOR UPDATE to lock the row during the read. The UPDATE uses a WHERE clause with the expected version. If zero rows are updated, the client must retry.
This pattern is particularly effective when conflicts are rare but expensive to resolve, such as in financial systems or inventory management.
This page was rewritten on 15 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.
We use cookies for analytics (Google Analytics) and advertising (Google AdSense) to improve your experience and support free content. Privacy Policy