linux 线程间队列通讯

Linux线程间队列通讯基础概念

在Linux系统中，线程间通信（Inter-thread Communication, ITC）是一种机制，允许不同的线程共享数据和信息。队列通讯是其中一种常见的方法，主要用于在生产者和消费者线程之间传递数据。

基础概念

队列（Queue）：一种先进先出（FIFO）的数据结构，用于存储和管理数据元素。
生产者线程（Producer Thread）：负责生成数据并将其放入队列中。
消费者线程（Consumer Thread）：负责从队列中取出数据并进行处理。

类型

有界队列：队列有一个固定的大小限制，当队列满时，生产者线程会被阻塞。
无界队列：队列大小不受限制，但需要注意内存消耗。

应用场景

任务分发：多个生产者线程生成任务，一个或多个消费者线程处理任务。
日志处理：多个线程生成日志，一个专门的线程负责写入日志文件。
数据流水线：多个阶段的数据处理，每个阶段由不同的线程完成。

示例代码

以下是一个简单的C语言示例，展示了如何使用POSIX线程（pthread）和队列进行线程间通信。

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>

#define QUEUE_SIZE 10
#define MAX_ITEM 100

typedef struct {
    int buffer[QUEUE_SIZE];
    int front;
    int rear;
    int count;
    pthread_mutex_t lock;
    pthread_cond_t not_empty;
    pthread_cond_t not_full;
} Queue;

Queue queue;

void* producer(void* arg) {
    for (int i = 0; i < MAX_ITEM; ++i) {
        pthread_mutex_lock(&queue.lock);
        while (queue.count == QUEUE_SIZE) {
            pthread_cond_wait(&queue.not_full, &queue.lock);
        }
        queue.buffer[queue.rear] = i;
        queue.rear = (queue.rear + 1) % QUEUE_SIZE;
        queue.count++;
        pthread_cond_signal(&queue.not_empty);
        pthread_mutex_unlock(&queue.lock);
        usleep(100);
    }
    return NULL;
}

void* consumer(void* arg) {
    for (int i = 0; i < MAX_ITEM; ++i) {
        pthread_mutex_lock(&queue.lock);
        while (queue.count == 0) {
            pthread_cond_wait(&queue.not_empty, &queue.lock);
        }
        int item = queue.buffer[queue.front];
        queue.front = (queue.front + 1) % QUEUE_SIZE;
        queue.count--;
        pthread_cond_signal(&queue.not_full);
        pthread_mutex_unlock(&queue.lock);
        printf("Consumed: %d\n", item);
        usleep(200);
    }
    return NULL;
}

int main() {
    pthread_t producer_thread, consumer_thread;

    pthread_mutex_init(&queue.lock, NULL);
    pthread_cond_init(&queue.not_empty, NULL);
    pthread_cond_init(&queue.not_full, NULL);

    queue.front = queue.rear = queue.count = 0;

    pthread_create(&producer_thread, NULL, producer, NULL);
    pthread_create(&consumer_thread, NULL, consumer, NULL);

    pthread_join(producer_thread, NULL);
    pthread_join(consumer_thread, NULL);

    pthread_mutex_destroy(&queue.lock);
    pthread_cond_destroy(&queue.not_empty);
    pthread_cond_destroy(&queue.not_full);

    return 0;
}