c++ 多线程 队列 顺序取出数据

菜鸟实现耗时测试，勿喷

#include <iostream>

#include <queue>

#include <thread>

#include <mutex>

#include <unistd.h>

#include <semaphore.h>

using namespace std

#define msleep(x) usleep(x*1000)

queue<int>queueIn

queue<int>queueOut

std::mutex threadMutex

int max_len = 4

int index = 0

static sem_t g_semaphore

static sem_t g_semaphore_queuOut

bool isEnd = false

void decoder() {

    int i = 0

    while (i <10000) {

        if(queueIn.size() >max_len) {

            continue

        }

        queueIn.push(i++)

    }

    while (index != 9999) {

    }

    isEnd = true

}

void encoder() {

    int i = -1

    while (!isEnd || !queueIn.empty()) {

        i = -1

        sem_wait(&g_semaphore)

        if(!queueIn.empty()){

            i = queueIn.front()

            queueIn.pop()

        }

        sem_post(&g_semaphore)

        if(i == -1){

            continue

        }

        while (true){

            if(i == 0){

                break

            }

            //printf("index %d i %d\n",index,i)

            if(i - index == 1){

                break

            }

        }

        sem_wait(&g_semaphore_queuOut)

        queueOut.push(i)

        if(i >index) {

            index = i

        }

        sem_post(&g_semaphore_queuOut)

    }

}

void enCoderH264() {

    while (!isEnd || !queueOut.empty() || !queueIn.empty()) {

        sem_wait(&g_semaphore_queuOut)

        if(!queueOut.empty()){

            int i = queueOut.front()

            queueOut.pop()

            printf("queueout %d \n",i)

        }

        sem_post(&g_semaphore_queuOut)

    }

    //printf("end %d \n",queueOut.size())

}

int main() {

    sem_init(&g_semaphore, 0, 1)

    sem_init(&g_semaphore_queuOut,0,1)

    thread t(decoder)

    thread t2(encoder)

    thread t3(encoder)

    thread t4(encoder)

    thread t5(enCoderH264)

    t.detach()

    t2.detach()

    t3.detach()

    t4.detach()

    t5.detach()

    std::cout <<"Hello, World!" <<std::endl

    pthread_exit(NULL)

    return 0

}

信号量初始化。 int sem_init (sem_t *sem , int pshared, unsigned int value) 这是对由sem指定的信号量进行初始化，设置好它的共享选项(linux 只支持为0，即表示它是当前进程的局部信号量)，然后给它一个初始值VALUE。等待信号量。给信号量减1，然后等待直到信号量的值大于0。int sem_wait(sem_t *sem) 释放信号量。信号量值加1。并通知其他等待线程。int sem_post(sem_t *sem) 销毁信号量。我们用完信号量后都它进行清理。归还占有的一切资源。int sem_destroy(sem_t *sem)#include #include #include #include #include #include #define return_if_fail(p) if((p) == 0){printf ("[%s]:func error!/n", __func__)return} typedef struct _PrivInfo { sem_t s1 sem_t s2 time_t end_time }PrivInfo static void info_init (PrivInfo* thiz) static void info_destroy (PrivInfo* thiz) static void* pthread_func_1 (PrivInfo* thiz) static void* pthread_func_2 (PrivInfo* thiz) int main (int argc, char** argv) { pthread_t pt_1 = 0 pthread_t pt_2 = 0 int ret = 0 PrivInfo* thiz = NULL thiz = (PrivInfo* )malloc (sizeof (PrivInfo)) if (thiz == NULL) { printf ("[%s]: Failed to malloc priv./n") return -1 } info_init (thiz) ret = pthread_create (&pt_1, NULL, (void*)pthread_func_1, thiz) if (ret != 0) { perror ("pthread_1_create:") } ret = pthread_create (&pt_2, NULL, (void*)pthread_func_2, thiz) if (ret != 0) { perror ("pthread_2_create:") } pthread_join (pt_1, NULL) pthread_join (pt_2, NULL) info_destroy (thiz) return 0 } static void info_init (PrivInfo* thiz) { return_if_fail (thiz != NULL) thiz->end_time = time(NULL) + 10 sem_init (&thiz->s1, 0, 1) sem_init (&thiz->s2, 0, 0) return } static void info_destroy (PrivInfo* thiz) { return_if_fail (thiz != NULL) sem_destroy (&thiz->s1) sem_destroy (&thiz->s2) free (thiz) thiz = NULL return } static void* pthread_func_1 (PrivInfo* thiz) { return_if_fail(thiz != NULL) while (time(NULL) thiz->end_time) { sem_wait (&thiz->s2) printf ("pthread1: pthread1 get the lock./n") sem_post (&thiz->s1) printf ("pthread1: pthread1 unlock/n") sleep (1) } return } static void* pthread_func_2 (PrivInfo* thiz) { return_if_fail (thiz != NULL) while (time (NULL) thiz->end_time) { sem_wait (&thiz->s1) printf ("pthread2: pthread2 get the unlock./n") sem_post (&thiz->s2) printf ("pthread2: pthread2 unlock./n") sleep (1) } return }

---