[pthread多线程]描述生产者消费者模型 #C #多线程

dynamic_cond_and_mutex.cpp

pthread_cond_t *condv = nullptr;
pthread_mutex_t *mlock = nullptr;

static data *phead = nullptr;

static void initLock() {
    condv = new pthread_cond_t;
    mlock = new pthread_mutex_t;
    pthread_mutex_init(mlock, nullptr);
    pthread_cond_init(condv, nullptr);
}

static void destroyLock() {
    pthread_mutex_destroy(mlock);
    pthread_cond_destroy(condv);
    delete condv;
    condv = nullptr;
    delete mlock;
    mlock = nullptr;
}

static void *producer(void *arg) {
    printf("producer thread running.\n");
    int count = 0;
    while (true) {
        int n = randomNum(100);
        data *nd = static_cast<data *>(malloc(sizeof(data)));
        memset(nd, 1, sizeof(data));
        nd->n = n;

        pthread_mutex_lock(mlock);
        // 头插法
        data *tmp = phead;
        phead = nd;
        nd->next = tmp;
        pthread_mutex_unlock(mlock);
        pthread_cond_signal(condv);

        count += n;

        if (count > LIMIT) {
            break;
        }
        sleep(rand() % 5);
    }
    printf("producer count=%d\n", count);
    return nullptr;
}

static void *consumer(void *arg) {
    printf("consumer thread running.\n");
    int count = 0;
    while (true) {
        pthread_mutex_lock(mlock);
        if (phead == nullptr) {
            pthread_cond_wait(condv, mlock);
        } else {
            while (phead != nullptr) {
                data *tmp = phead;
                count += tmp->n;
                phead = phead->next;
                free(tmp);
            }
        }
        pthread_mutex_unlock(mlock);
        if (count > LIMIT) {
            break;
        }
    }
    printf("consumer count=%d", count);
    return nullptr;
}

int main() {
    initLock();
    pthread_t t1;
    pthread_t t2;
    pthread_create(&t1, nullptr, producer, nullptr);
    pthread_create(&t2, nullptr, consumer, nullptr);

    pthread_join(t1, nullptr);
    pthread_join(t2, nullptr);
    destroyLock();

    return 0;
}

static_cond_and_mutex.cpp

static pthread_cond_t condv = PTHREAD_COND_INITIALIZER;
static pthread_mutex_t mlock = PTHREAD_MUTEX_INITIALIZER;
static data *phead = nullptr;

static void *producer(void *arg) {
    printf("producer thread running.\n");
    int count = 0;
    while (true) {
        int n = rand() % 100;
        data *nd = static_cast<data *>(malloc(sizeof(data)));
        memset(nd, 1, sizeof(data));
        nd->n = n;

        pthread_mutex_lock(&mlock);
        data *tmp = phead;
        phead = nd;
        nd->next = tmp;
        pthread_mutex_unlock(&mlock);
        pthread_cond_signal(&condv);

        count += n;

        if (count > LIMIT) {
            break;
        }
        sleep(rand() % 5);
    }
    printf("producer count=%d\n", count);
    return nullptr;
}

static void *consumer(void *arg) {
    printf("consumer thread running.\n");
    int count = 0;
    while (true) {
        pthread_mutex_lock(&mlock);
        if (phead == nullptr) {
            pthread_cond_wait(&condv, &mlock);
        } else {
            while (phead != nullptr) {
                data *tmp = phead;
                count += tmp->n;
                phead = phead->next;
                free(tmp);
            }
        }
        pthread_mutex_unlock(&mlock);
        if (count > LIMIT) {
            break;
        }
    }
    printf("consumer count=%d", count);
    return nullptr;
}

int main() {
    pthread_t t1;
    pthread_t t2;
    pthread_create(&t1, nullptr, producer, nullptr);
    pthread_create(&t2, nullptr, consumer, nullptr);

    pthread_join(t1, nullptr);
    pthread_join(t2, nullptr);
    return 0;
}