C语言消息队列生产者和消费者代码

在C语言中，我们可以使用`pthread`库来创建线程并实现消息队列的生产者（Producer）和消费者（Consumer）。下面是一个简单的例子，展示了如何使用互斥量（mutex）、条件变量（condition variable）以及数组作为消息队列。

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

#define MAX_MSG 10
#define BUFFER_SIZE sizeof(char*) * MAX_MSG

typedef struct {
    char* messages[MAX_MSG];
    int count;
    int producer_id;
    int consumer_id;
} MessageQueue;

void* producer_func(void* arg) {
    MessageQueue* mq = (MessageQueue*)arg;
    int i;
    for (i = 0; i < MAX_MSG; ++i) {
        // 生产消息
        char* msg = malloc(sizeof(char) * 50);
        strcpy(msg, "Producer produced message: ");
        strcat(msg, "msg");
        printf("Producer: Added %s\n", msg);

        pthread_mutex_lock(&mq->mutex);
        if (mq->count < MAX_MSG) {  // 如果队列不满
            mq->messages[mq->count++] = msg;
            pthread_cond_signal(&mq->cv);
        }
        pthread_mutex_unlock(&mq->mutex);

        free(msg);  // 释放内存
    }
}

void* consumer_func(void* arg) {
    MessageQueue* mq = (MessageQueue*)arg;
    while (1) {
        pthread_mutex_lock(&mq->mutex);
        while (mq->count == 0) {  // 队列为空，等待
            pthread_cond_wait(&mq->cv, &mq->mutex);
        }
        if (mq->count > 0) {
            char* msg = mq->messages[--mq->count];  // 消费消息
            printf("Consumer consumed: %s\n", msg);
            free(msg);
        }
        pthread_mutex_unlock(&mq->mutex);
    }
}

int main() {
    MessageQueue mq;
    pthread_t producer_thread, consumer_thread;
    
    // 初始化
    mq.count = 0;
    mq.producer_id = 1;
    mq.consumer_id = 2;
    pthread_mutex_init(&mq.mutex, NULL);
    pthread_cond_init(&mq.cv, NULL);

    // 创建线程
    if (pthread_create(&producer_thread, NULL, producer_func, (void*)&mq)) {
        perror("Producer thread creation failed");
        return -1;
    }

    if (pthread_create(&consumer_thread, NULL, consumer_func, (void*)&mq)) {
        perror("Consumer thread creation failed");
        return -1;
    }

    // 等待线程结束
    pthread_join(producer_thread, NULL);
    pthread_join(consumer_thread, NULL);

    pthread_mutex_destroy(&mq.mutex);
    pthread_cond_destroy(&mq.cv);
    return 0;
}

这个例子中，生产者不断添加消息到队列，而消费者从队列中消费消息，直到队列为空。通过互斥量保护数据，避免并发访问导致的数据混乱，并使用条件变量控制生产者和消费者的同步。