linux多线程编程(C)：互斥量实现的线程安全队列

用互斥量实现的线程安全循环队列。
只是一个示例实现，不大能用于实际编程。
因为当真正多线程时，会因为资源竞争而等待，
导致队列的效率极低。
更有效的方式是使用信号量。

更多的不说了，直接看代码吧。
cir_queue.h

/*

* \File

* cir_queue.h

* \Brief

* circular queue

*/

#ifndef __CIR_QUEUE_H__

#define __CIR_QUEUE_H__

#define QUE_SIZE 8

typedef int DataType;

typedef struct cir_queue_t

{

DataType data[QUE_SIZE];

int front;

int rear;

int count;

}cir_queue_t;

extern pthread_mutex_t queue_mutex;

void init_cir_queue(cir_queue_t* q);

int is_empty_cir_queue(cir_queue_t* q);

int is_full_cir_queue(cir_queue_t* q);

void push_cir_queue(cir_queue_t* q, DataType x);

DataType pop_cir_queue(cir_queue_t* q);

DataType top_cir_queue(cir_queue_t* q);

void destroy_cir_queue(cir_queue_t* q);

void print_queue(cir_queue_t* q);

#endif

main.c

/*

* \File

* main.c

* \Breif

* Thread-safe circular-queue implemented by mutex

* \Author

* Hank.yan

*/

#include <stdio.h>

#include <stdlib.h>

#include <unistd.h>

#include <string.h>

#include <pthread.h>

#include <semaphore.h>

#include "cir_queue.h"

void* thread_queue(void *arg);

/*

* \Func

* main

*/

int main(int argc, char* argv[])

{

int res;

cir_queue_t cq;

DataType e;

pthread_t a_thread;

void* thread_result;

init_cir_queue(&cq);

push_cir_queue(&cq, 1);

push_cir_queue(&cq, 2);

push_cir_queue(&cq, 3);

print_queue(&cq);

res = pthread_create(&a_thread, NULL, thread_queue, (void*)&cq);

if (res != 0)

{

perror("Thread creation failed.");

exit(EXIT_FAILURE);

}

e = pop_cir_queue(&cq);

e = pop_cir_queue(&cq);

print_queue(&cq);

push_cir_queue(&cq, 9);

push_cir_queue(&cq, 100);

print_queue(&cq);

e = pop_cir_queue(&cq);

push_cir_queue(&cq, 20);

print_queue(&cq);

printf("Waiting for thread to finish...\n");

res = pthread_join(a_thread, &thread_result);

if (res != 0)

{

perror("Thread join failed.");

exit(EXIT_FAILURE);

}

print_queue(&cq);

destroy_cir_queue(&cq);

printf("Thread joined, it returned %s\n", (char*)thread_result);

exit(EXIT_SUCCESS);

}

void *thread_queue(void *cirqueue)

{

int flag;

DataType element;

print_queue((cir_queue_t*)cirqueue);

flag = is_empty_cir_queue((cir_queue_t*)cirqueue);

print_queue((cir_queue_t*)cirqueue);

element = pop_cir_queue((cir_queue_t*)cirqueue);

element = pop_cir_queue((cir_queue_t*)cirqueue);

print_queue((cir_queue_t*)cirqueue);

push_cir_queue((cir_queue_t*)cirqueue, 5);

print_queue((cir_queue_t*)cirqueue);

push_cir_queue((cir_queue_t*)cirqueue, 99);

push_cir_queue((cir_queue_t*)cirqueue, 1000);

push_cir_queue((cir_queue_t*)cirqueue, 88);

print_queue((cir_queue_t*)cirqueue);

pthread_exit("Thank you for the cpu time.");

}

cir_queue.c

/*

* \File

* cir_queue.c

*/

#include <stdio.h>

#include <stdlib.h>

#include <unistd.h>

#include <string.h>

#include <pthread.h>

#include <semaphore.h>

#include "cir_queue.h"

pthread_mutex_t queue_mutex;

/*

* \Func

*

*/

void init_cir_queue(cir_queue_t *q)

{

int res;

res= pthread_mutex_init(&queue_mutex, NULL);

if (res != 0)

{

perror("Mutex init failed.\n");

exit(EXIT_FAILURE);

}

memset(q->data, 0, QUE_SIZE*sizeof(DataType));

q->front = q->rear = 0;

q->count = 0;

}

/*

* \Func

*

*/

int is_empty_cir_queue(cir_queue_t *q)

{

int empty_flag;

pthread_mutex_lock(&queue_mutex);

empty_flag = q->front == q->rear;

pthread_mutex_unlock(&queue_mutex);

return empty_flag;

}

/*

* \Func

*

*/

int is_full_cir_queue(cir_queue_t *q)

{

int full_flag;

pthread_mutex_lock(&queue_mutex);

full_flag = q->rear == QUE_SIZE - 1 + q->front;

pthread_mutex_unlock(&queue_mutex);

return full_flag;

}

/*

* \Func

*

*/

void push_cir_queue(cir_queue_t *q, DataType x)

{

if (is_full_cir_queue(q))

{

printf("queue overflow.\n");

return ;

}

pthread_mutex_lock(&queue_mutex);

q->count++;

q->data[q->rear] = x;

q->rear = (q->rear+1) % QUE_SIZE;

pthread_mutex_unlock(&queue_mutex);

}

/*

* \Func

*

*/

DataType pop_cir_queue(cir_queue_t *q)

{

DataType temp;

if (is_empty_cir_queue(q))

{

printf("queue empty.\n");

return 0;

}

pthread_mutex_lock(&queue_mutex);

temp = q->data[q->front];

q->data[q->front] = 0;

q->count--;

q->front = (q->front+1) % QUE_SIZE;

pthread_mutex_unlock(&queue_mutex);

return temp;

}

/*

* \Func

*

*/

DataType top_cir_queue(cir_queue_t *q)

{

DataType x;

if (is_empty_cir_queue(q))

{

printf("queue is empty.\n");

return 0;

}

pthread_mutex_lock(&queue_mutex);

x = q->data[q->front];

pthread_mutex_unlock(&queue_mutex);

return x;

}

void destroy_cir_queue(cir_queue_t *q)

{

pthread_mutex_destroy(&queue_mutex);

return;

}

void print_queue(cir_queue_t* q)

{

int index;

if (is_empty_cir_queue(q))

{

printf("queue is empty.\n");

return;

}

pthread_mutex_lock(&queue_mutex);

printf("QUEUE: ");

for (index = 0; index < QUE_SIZE; index++)

{

printf(" %d ", q->data[index]);

}

printf("\n");

pthread_mutex_unlock(&queue_mutex);

return;

}

makefile

OBJECTS = main.o cir_queue.o

CC = gcc

CFLAGS = -D_REENTRANT -lpthread -g -Wall

thrd_safe_queue: $(OBJECTS)

$(CC) $(CFLAGS) -o
thrd_safe_queue $(OBJECTS)

main.o: cir_queue.h

cir_queue.o: cir_queue.h

.PHONY:clean

clean:

rm thrd_safe_queue $(OBJECTS)