linux 内核 多生产者 多消费者 无锁 队列 实现示例代码。

以下是一个简单的无锁队列实现，支持多生产者和多消费者：

#include <stdatomic.h>

#define QUEUE_SIZE 1024

typedef struct {
    int data[QUEUE_SIZE];
    atomic_int head;
    atomic_int tail;
} queue_t;

void queue_init(queue_t *queue) {
    queue->head = 0;
    queue->tail = 0;
    atomic_thread_fence(memory_order_seq_cst);
}

int queue_push(queue_t *queue, int value) {
    int tail = atomic_load_explicit(&queue->tail, memory_order_relaxed);
    int next_tail = (tail + 1) % QUEUE_SIZE;
    if (next_tail != atomic_load_explicit(&queue->head, memory_order_acquire)) {
        queue->data[tail] = value;
        atomic_store_explicit(&queue->tail, next_tail, memory_order_release);
        return 1;
    }
    return 0;
}

int queue_pop(queue_t *queue, int *value) {
    int head = atomic_load_explicit(&queue->head, memory_order_relaxed);
    if (head == atomic_load_explicit(&queue->tail, memory_order_acquire)) {
        return 0;
    }
    *value = queue->data[head];
    atomic_store_explicit(&queue->head, (head + 1) % QUEUE_SIZE, memory_order_release);
    return 1;
}

在这个实现中，队列使用循环数组来存储数据。队列头和尾都是原子整型变量，用来保证多线程下的一致性。在 push 操作中，先获取尾指针，然后计算下一个尾指针，如果下一个尾指针不等于头指针，则将数据存储在当前尾指针处，然后更新尾指针。在 pop 操作中，先获取头指针，如果头指针等于尾指针，则队列为空，返回 0，否则将头指针处的数据存储在 value 变量中，然后更新头指针。

使用示例：

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

#define NUM_PRODUCERS 4
#define NUM_CONSUMERS 4
#define NUM_ITEMS 100000

queue_t queue;

void *producer_thread(void *arg) {
    int id = *(int *)arg;
    srand(time(NULL) + id);
    for (int i = 0; i < NUM_ITEMS; ++i) {
        int value = rand() % 100;
        while (!queue_push(&queue, value)) {
            sched_yield();
        }
    }
    return NULL;
}

void *consumer_thread(void *arg) {
    int id = *(int *)arg;
    int sum = 0;
    while (1) {
        int value;
        if (!queue_pop(&queue, &value)) {
            if (__atomic_load_n(&queue.tail, __ATOMIC_RELAXED) == NUM_PRODUCERS * NUM_ITEMS) {
                break;
            }
            sched_yield();
            continue;
        }
        sum += value;
    }
    printf("consumer %d sum=%d\n", id, sum);
    return NULL;
}

int main() {
    pthread_t producers[NUM_PRODUCERS], consumers[NUM_CONSUMERS];
    int producer_ids[NUM_PRODUCERS], consumer_ids[NUM_CONSUMERS];

    queue_init(&queue);

    for (int i = 0; i < NUM_PRODUCERS; ++i) {
        producer_ids[i] = i;
        pthread_create(&producers[i], NULL, producer_thread, &producer_ids[i]);
    }

    for (int i = 0; i < NUM_CONSUMERS; ++i) {
        consumer_ids[i] = i;
        pthread_create(&consumers[i], NULL, consumer_thread, &consumer_ids[i]);
    }

    for (int i = 0; i < NUM_PRODUCERS; ++i) {
        pthread_join(producers[i], NULL);
    }

    for (int i = 0; i < NUM_CONSUMERS; ++i) {
        pthread_join(consumers[i], NULL);
    }

    return 0;
}

在这个示例中，我们创建了 4 个生产者线程和 4 个消费者线程，每个生产者线程向队列中随机插入 100000 个数据，每个消费者线程从队列中取出数据并计算总和。最后，我们等待所有线程结束，程序退出