Quartz 4

阻塞队列

18 min read

实现原理

实现一个线程安全的阻塞队列（Blocking Queue）是一个经典的并发编程问题。其核心在于正确、高效地协调多个线程对共享资源（队列）的访问，确保数据的一致性、操作的原子性，并在特定条件下（如队列满或空）让线程能够阻塞等待。

以下是实现一个阻塞队列的理论重点：

1. 线程安全 (Thread Safety) - 基石

这是最根本的要求。多个生产者和消费者线程同时访问队列时，必须保证队列内部状态（如元素数组、头尾指针、大小计数器）的一致性和完整性。

  • 问题: 如果没有保护，两个线程可能同时修改 size 变量，导致计数错误；或者一个线程在读取 front 指针时，另一个线程正在修改它，导致读取到不一致的状态。
  • 解决方案: 互斥锁 (Mutex)。
    • 所有对队列内部状态的读写操作（put, take, size, empty 等）都必须被一个互斥锁保护起来。
    • 这确保了在任何时刻，只有一个线程可以执行这些临界区代码，从而保证了操作的原子性和数据的一致性。

2. 阻塞与唤醒 (Blocking And Signaling) - 核心机制

阻塞队列的关键特性是“阻塞”。当操作无法立即完成时（生产者遇到满队列，消费者遇到空队列），线程不应忙等（浪费 CPU），而应进入等待状态，直到条件满足。

  • 问题: 如何让线程在条件不满足时安全地“睡着”，并在条件满足时被“叫醒”？
  • 解决方案: 条件变量 (Condition Variables) 配合 wait/notify 机制。
    • 条件变量: 代表一个或多个线程正在等待的某个特定条件（如“队列非空”、“队列非满”）。它通常与一个互斥锁关联。
    • wait 操作:
      1. 线程首先持有互斥锁。
      2. 检查条件（如 size == 0）。
      3. 如果条件不满足，调用 cond_wait(&cond_var, &mutex)。这个原子操作会释放互斥锁并将线程放入该条件变量的等待队列中挂起。这一步至关重要，因为它释放了锁，允许其他线程修改队列状态。
    • signal/broadcast 操作:
      1. 当一个线程（如生产者）成功放入一个元素后，它知道“队列非空”这个条件现在可能满足了。
      2. 它调用 cond_signal(&not_empty_cond)（唤醒一个等待者）或 cond_broadcast(&not_empty_cond)（唤醒所有等待者）。
      3. 被唤醒的线程会重新尝试获取互斥锁。一旦获取到锁，cond_wait 调用返回，线程从挂起点继续执行。
    • 关键点: wait 必须在循环中检查条件。因为存在虚假唤醒 (Spurious Wakeup) 的可能性（线程被唤醒但条件并未真正满足），或者有多个线程在等待同一个条件（signal 可能唤醒了错误的线程，或者 broadcast 唤醒了所有线程但只有一个能成功操作）。因此，被唤醒的线程必须重新检查条件是否真的满足。

3. 状态管理与边界条件 (State Management and Edge Cases)

  • 队列实现: 通常使用循环缓冲区 (Circular Buffer) 来高效利用固定大小的数组。通过模运算 ((index + 1) % capacity) 来处理头尾指针的回绕。
  • 容量管理: 需要精确维护 size 或通过 front/rear 指针计算当前大小，以判断队列是满还是空。
  • 关闭/终止机制 (Graceful Shutdown):
    • 一个重要的设计是允许队列被“关闭”。关闭后，新的 put 操作应失败（返回错误或抛出异常），而 take 操作在取完剩余元素后应返回一个特定值（如 NULL）以通知消费者可以退出。
    • 这通常通过一个 closed 标志位实现。关闭时，需要 broadcast 所有等待的条件变量，唤醒所有阻塞的线程，让它们检查 closed 标志并做出相应处理（生产者放弃放入，消费者在空时返回 NULL）。这避免了线程永久阻塞。

4. 原子性与可见性 (Atomicity And Visibility)

  • 原子性: 互斥锁保证了临界区操作的原子性。
  • 可见性: 在多核 CPU 和现代编译器优化下，一个线程对共享变量的修改可能不会立即被其他线程看到。互斥锁的 lock 和 unlock 操作不仅提供互斥，还建立了内存屏障 (Memory Barrier) 或 happens-before 关系。这确保了在锁释放前对共享变量的所有修改，对于之后获得同一把锁的线程都是可见的。

5. 死锁预防 (Deadlock Prevention)

  • 问题: 如果设计不当，可能导致死锁。例如，如果 put 和 take 使用了不同的锁，并且以不同的顺序获取，就可能发生死锁。
  • 解决方案: 通常使用单一互斥锁保护整个队列状态和所有条件变量。这从根本上避免了因锁顺序导致的死锁。虽然这可能成为性能瓶颈（串行化所有操作），但对于正确性至关重要。在极高性能场景下，可能会设计更复杂的无锁或分段锁结构，但复杂度剧增。

总结

实现一个阻塞队列的理论重点可以归结为：

  1. 用互斥锁 (Mutex) 保护共享状态，确保线程安全和操作原子性。
  2. 用条件变量 (Condition Variables) 实现阻塞与唤醒，让线程在条件不满足时高效等待，在条件满足时被通知。
  3. 在循环中检查条件，以处理虚假唤醒和竞争条件。
  4. 实现优雅关闭，避免线程永久阻塞。
  5. 正确处理边界条件（空、满、关闭）。
  6. 利用锁的内存语义保证修改的可见性。
  7. 避免死锁，通常通过单一锁策略。

简而言之，互斥锁解决“同时访问”的问题，条件变量解决“何时访问”的问题。两者结合，是实现阻塞队列这类并发数据结构的基石。

好的，这是一个使用 C 语言实现的线程安全阻塞队列（Blocking Queue）。

这个实现使用了 POSIX 线程（pthreads）库，包含互斥锁（mutex）和条件变量（condition variables）来保证线程安全和阻塞行为。

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <assert.h>
#include <string.h>
 
// 阻塞队列结构体
typedef struct {
    void **data;          // 存储数据的数组（指针数组）
    int capacity;         // 队列容量
    int size;             // 当前元素数量
    int front;            // 队头索引
    int rear;             // 队尾索引
    pthread_mutex_t mutex;     // 互斥锁
    pthread_cond_t not_empty;  // 条件变量：队列非空
    pthread_cond_t not_full;   // 条件变量：队列非满
    int closed;           // 标记队列是否已关闭
} blocking_queue_t;
 
/**
 * @brief 创建一个阻塞队列
 * @param capacity 队列的最大容量
 * @return 指向新创建的阻塞队列的指针，失败返回 NULL
 */
blocking_queue_t* bq_create(int capacity) {
    if (capacity <= 0) {
        fprintf(stderr, "Error: Capacity must be positive.\n");
        return NULL;
    }
 
    blocking_queue_t *bq = (blocking_queue_t*)malloc(sizeof(blocking_queue_t));
    if (!bq) {
        fprintf(stderr, "Error: Failed to allocate memory for queue.\n");
        return NULL;
    }
 
    // 分配存储数据的数组
    bq->data = (void**)malloc(capacity * sizeof(void*));
    if (!bq->data) {
        fprintf(stderr, "Error: Failed to allocate memory for data array.\n");
        free(bq);
        return NULL;
    }
 
    bq->capacity = capacity;
    bq->size = 0;
    bq->front = 0;
    bq->rear = 0;
    bq->closed = 0;
 
    // 初始化互斥锁
    if (pthread_mutex_init(&bq->mutex, NULL) != 0) {
        fprintf(stderr, "Error: Failed to initialize mutex.\n");
        free(bq->data);
        free(bq);
        return NULL;
    }
 
    // 初始化条件变量
    if (pthread_cond_init(&bq->not_empty, NULL) != 0) {
        fprintf(stderr, "Error: Failed to initialize not_empty condition variable.\n");
        pthread_mutex_destroy(&bq->mutex);
        free(bq->data);
        free(bq);
        return NULL;
    }
 
    if (pthread_cond_init(&bq->not_full, NULL) != 0) {
        fprintf(stderr, "Error: Failed to initialize not_full condition variable.\n");
        pthread_cond_destroy(&bq->not_empty);
        pthread_mutex_destroy(&bq->mutex);
        free(bq->data);
        free(bq);
        return NULL;
    }
 
    return bq;
}
 
/**
 * @brief 销毁一个阻塞队列
 * @param bq 指向要销毁的阻塞队列的指针
 * @note 调用此函数前，应确保没有线程正在使用或等待该队列。
 *       该函数不负责释放队列中存储的数据（void* 指向的对象）。
 */
void bq_destroy(blocking_queue_t *bq) {
    if (!bq) return;
 
    // 销毁同步原语
    pthread_cond_destroy(&bq->not_empty);
    pthread_cond_destroy(&bq->not_full);
    pthread_mutex_destroy(&bq->mutex);
 
    // 释放数据数组和队列结构体本身
    free(bq->data);
    free(bq);
}
 
/**
 * @brief 将一个元素放入队列末尾（阻塞操作）
 * 如果队列已满，调用线程将阻塞，直到队列有空间或队列被关闭。
 * @param bq 指向阻塞队列的指针
 * @param item 要放入队列的元素（指针）
 * @return 0 成功放入；-1 队列已关闭或出错
 */
int bq_put(blocking_queue_t *bq, void *item) {
    if (!bq || !item) {
        return -1;
    }
 
    pthread_mutex_lock(&bq->mutex);
 
    // 等待队列不满且未关闭
    while (bq->size >= bq->capacity && !bq->closed) {
        pthread_cond_wait(&bq->not_full, &bq->mutex);
    }
 
    // 检查队列是否已关闭
    if (bq->closed) {
        pthread_mutex_unlock(&bq->mutex);
        return -1; // 表示放入失败，因为队列关闭
    }
 
    // 将元素放入队尾
    bq->data[bq->rear] = item;
    bq->rear = (bq->rear + 1) % bq->capacity;
    bq->size++;
 
    // 唤醒一个可能在等待队列非空的消费者线程
    pthread_cond_signal(&bq->not_empty);
 
    pthread_mutex_unlock(&bq->mutex);
    return 0;
}
 
/**
 * @brief 从队列头部获取一个元素（阻塞操作）
 * 如果队列为空，调用线程将阻塞，直到队列有元素或队列被关闭。
 * @param bq 指向阻塞队列的指针
 * @return 成功时返回队列头部的元素指针；如果队列为空且已关闭，返回 NULL
 */
void* bq_take(blocking_queue_t *bq) {
    if (!bq) {
        return NULL;
    }
 
    pthread_mutex_lock(&bq->mutex);
 
    // 等待队列非空且未关闭
    while (bq->size == 0 && !bq->closed) {
        pthread_cond_wait(&bq->not_empty, &bq->mutex);
    }
 
    // 如果队列已关闭且为空，则返回 NULL
    if (bq->closed && bq->size == 0) {
        pthread_mutex_unlock(&bq->mutex);
        return NULL;
    }
 
    // 取出队头元素
    void *item = bq->data[bq->front];
    bq->front = (bq->front + 1) % bq->capacity;
    bq->size--;
 
    // 唤醒一个可能在等待队列非满的生产者线程
    pthread_cond_signal(&bq->not_full);
 
    pthread_mutex_unlock(&bq->mutex);
    return item;
}
 
/**
 * @brief 关闭阻塞队列
 * 关闭后，任何后续的 bq_put 操作将失败（返回 -1）。
 * 对于 bq_take 操作：
 *   - 如果队列中还有元素，可以继续取出。
 *   - 当队列变为空时，bq_take 将立即返回 NULL。
 * @param bq 指向要关闭的阻塞队列的指针
 */
void bq_close(blocking_queue_t *bq) {
    if (!bq) return;
 
    pthread_mutex_lock(&bq->mutex);
    bq->closed = 1;
    // 唤醒所有等待的线程，让它们检查 bq->closed 状态
    pthread_cond_broadcast(&bq->not_empty);
    pthread_cond_broadcast(&bq->not_full);
    pthread_mutex_unlock(&bq->mutex);
}
 
/**
 * @brief 获取队列当前大小
 * @param bq 指向阻塞队列的指针
 * @return 队列中当前元素的数量
 */
int bq_size(blocking_queue_t *bq) {
    if (!bq) return -1;
 
    pthread_mutex_lock(&bq->mutex);
    int size = bq->size;
    pthread_mutex_unlock(&bq->mutex);
    return size;
}
 
/**
 * @brief 检查队列是否为空
 * @param bq 指向阻塞队列的指针
 * @return 1 表示为空，0 表示非空
 */
int bq_empty(blocking_queue_t *bq) {
    if (!bq) return 1;
 
    pthread_mutex_lock(&bq->mutex);
    int empty = (bq->size == 0);
    pthread_mutex_unlock(&bq->mutex);
    return empty;
}
 
/**
 * @brief 检查队列是否已满
 * @param bq 指向阻塞队列的指针
 * @return 1 表示已满，0 表示未满
 */
int bq_full(blocking_queue_t *bq) {
    if (!bq) return 0;
 
    pthread_mutex_lock(&bq->mutex);
    int full = (bq->size >= bq->capacity);
    pthread_mutex_unlock(&bq->mutex);
    return full;
}
 
/**
 * @brief 检查队列是否已关闭
 * @param bq 指向阻塞队列的指针
 * @return 1 表示已关闭，0 表示未关闭
 */
int bq_is_closed(blocking_queue_t *bq) {
    if (!bq) return 1;
 
    pthread_mutex_lock(&bq->mutex);
    int closed = bq->closed;
    pthread_mutex_unlock(&bq->mutex);
    return closed;
}
 
// ==================== 示例用法 ====================
 
#include <unistd.h> // for sleep
 
// 生产者线程函数
void* producer(void *arg) {
    blocking_queue_t *queue = (blocking_queue_t*)arg;
    int item_id = 1;
 
    while (item_id <= 15) {
        char *item = (char*)malloc(32);
        if (!item) {
            perror("malloc");
            break;
        }
        snprintf(item, 32, "Item-%d", item_id);
 
        printf("Producer: Putting %s\n", item);
        if (bq_put(queue, item) == 0) {
            printf("Producer: Successfully put %s\n", item);
            item_id++;
        } else {
            printf("Producer: Failed to put %s (Queue closed?)\n", item);
            free(item); // 放入失败，需要释放内存
            break;
        }
        sleep(1); // 模拟生产耗时
    }
 
    printf("Producer: Done.\n");
    return NULL;
}
 
// 消费者线程函数
void* consumer(void *arg) {
    blocking_queue_t *queue = (blocking_queue_t*)arg;
 
    while (1) {
        void *item = bq_take(queue);
        if (item == NULL) {
            // 队列已关闭且为空
            printf("Consumer: Queue is closed and empty. Exiting.\n");
            break;
        }
 
        printf("Consumer: Took %s\n", (char*)item);
        // 模拟消费耗时
        sleep(2);
        free(item); // 消费完毕，释放内存
    }
 
    printf("Consumer: Done.\n");
    return NULL;
}
 
// 主函数示例
int main() {
    // 创建容量为 5 的阻塞队列
    blocking_queue_t *queue = bq_create(5);
    if (!queue) {
        fprintf(stderr, "Failed to create blocking queue.\n");
        return 1;
    }
 
    pthread_t prod_thread, cons_thread;
 
    // 创建生产者和消费者线程
    if (pthread_create(&prod_thread, NULL, producer, queue) != 0) {
        fprintf(stderr, "Error creating producer thread.\n");
        bq_destroy(queue);
        return 1;
    }
 
    if (pthread_create(&cons_thread, NULL, consumer, queue) != 0) {
        fprintf(stderr, "Error creating consumer thread.\n");
        bq_destroy(queue);
        return 1;
    }
 
    // 让生产者线程运行一段时间后关闭队列
    sleep(10);
    printf("Main: Closing the queue...\n");
    bq_close(queue);
 
    // 等待线程结束
    pthread_join(prod_thread, NULL);
    pthread_join(cons_thread, NULL);
 
    // 销毁队列
    bq_destroy(queue);
    printf("Main: Queue destroyed. Program finished.\n");
 
    return 0;
}

主要特性

  1. 线程安全: 使用 pthread_mutex_t 保护对队列状态（size, front, rear, data 数组）的访问。
  2. 阻塞操作:
    • bq_put: 当队列满时，生产者线程会阻塞在 pthread_cond_wait(&bq->not_full, &bq->mutex) 上，直到有空间。
    • bq_take: 当队列空时，消费者线程会阻塞在 pthread_cond_wait(&bq->not_empty, &bq->mutex) 上，直到有元素。
  3. 条件变量: not_empty 通知消费者有新元素，not_full 通知生产者有新空间。
  4. 关闭机制: bq_close 允许优雅地关闭队列。关闭后，bq_put 会失败，bq_take 在取完剩余元素后会返回 NULL。
  5. 循环缓冲区: 使用模运算 (% capacity) 实现数组的循环使用，提高空间利用率。
  6. 通用性: 使用 void* 指针存储数据，可以存放任意类型的数据（需要用户自己管理数据的内存）。

编译和运行

你需要链接 pthread 库。

gcc -o blocking_queue blocking_queue.c -lpthread
./blocking_queue

注意事项

  • 内存管理: 该实现不管理 void* 指向的数据的内存。生产者放入的指针，需要由消费者在取出后负责释放（如示例所示）。如果放入的数据是栈上分配的或不需要动态释放的，需注意不要重复释放。
  • 错误处理: 示例中的错误处理相对基础，实际应用中可能需要更精细的处理。
  • 平台依赖: 使用了 POSIX 线程（pthreads），在 Linux 和 macOS 上可用。Windows 需要使用 MinGW 或 WSL，或者改用 Windows API。
  • 性能: 对于极高性能要求的场景，可能需要更复杂的无锁队列（lock-free queue），但实现难度和复杂度大大增加。

这个实现提供了一个功能完整、线程安全的阻塞队列，适用于生产者 - 消费者模型。

Graph View

Backlinks