UNIX多线程数据共享与线程同步.doc

UNIX多线程数据共享与线程同步 2001/12/1917:33赛迪网-中国计算机报文/杨海平、姚洪利在UNIX中，一个进程让另外实体进行某项事务而采取的操作为fork的一个子进程，子进程只是将父进程的数据区拷贝一份到自己的数据区。在符合POSIX标准的UNIX操作系统下，同一个进程的线程之间共享进程指令、大多数数据(线程私有数据除外)、信号处理方式、进程运行环境等。由于线程共享进程的全局变量，因此可以采用用户自己编写的消息队列来实现数据的共享。建立多任务模型，并用线程来实现符合POSIX标准的UNIX操作系统提供了线程的控制函数，如：线程的创建和终止、线程之间的互斥、线程之间的同步等。利用这些系统函数可以成功地模拟消息队列，来实现线程间数据共享和同步，以完成多任务的实时性。为成功地描述线程间数据共享和同步，以下列任务模型为例。首先建立消息队列属性数据结构#define MAXQUEUE 30typedef struct mq_attrib char name20;pthread_mutex_t mutex_buff;pthread_mutex_t mutex_cond;pthread cond_t cond;int maxElements;int elementLength;int curElementNum;caddr_t buff;mq_attrib,mq_attribstruct,mq_attrib_t;mq_attrib_t msqueueMAXQUEUE;数据结构定义了消息队列的名字name，最大消息个数maxElements，单个消息长度elementLength，当前消息个数curElementNum，存放消息的缓冲区buff，保护缓冲区锁mutex_buff，线程同步条件变量cond，保护线程同步条件变量锁mutex_cond。消息队列的创建依据此数据结构进行消息队列的创建，函数为msqueue_create(参数解释：name消息队列名，maxnum消息的最大个数，length单个消息的长度)。int msqueue_create( name, maxnum, length )charname;int maxnum,length;int i;for ( i=0; iif ( msqueuei=NULL )break;/如果消息队列全部被分配，返回错if ( i=MAXQUEUE ) return MQERROR;msqueuei=malloc(sizeof(mq_attribstruct);sprintf( msqueuei-name, %s, name);msqueuei-maxElements = maxnum;msqueuei-elementLength = length;msqueuei-curElementNum = 0;msqueuei-buff=malloc(maxnumlength);/对保护锁进行初始化pthread_mutex_init(&msqueuei-mutex_buff, NULL);pthread_mutex_init(&msqueuei-mutex_cond, NULL);/对线程同步条件变量初始化pthread_cond_init(&msqueuei-cond, NULL);return i;应用消息队列进行消息的发送和接收发送消息到消息队列：消息队列的发送和接收是在不同的线程中进行的。首先介绍发送消息到消息队列的函数：int msqueue_send ( id, buff, length )int id, length;caddr_t buff;int pos;/消息队列id错，返回错if ( id= MAXQUEU ) return MQERROR;/消息长度与创建时的长度不符，返回错if ( length != msqueueid-elementLength ) return MQERROR;/消息队列满，不能发送if ( msqueueid-curElementNum = msqueueid-maxElements )return MQERROR;/在对消息队列缓冲区操作前，锁住缓冲区，以免其他线程操作pthread_mutex_lock ( &msqueueid-mutex_buff );pos = msqueueid-curElementNum * msqueueid-elementLength;bcopy ( buff, &msqueueid-buffpos, msqueueid-elementLength );msqueueid-curElementNum +;pthread_mutex_unlock ( &msqueueid-mutex_buff );/如果插入消息前，消息队列是空的，插入消息后，消息队列为非空，则通知等待从消息队列取消息的线程，条件满足，可以取出消息进行处理if ( msqueueid-curElementNum = 1 ) pthread_mutex_lock ( &msqueueid-mutex_cond );pthread_cond_broadcast ( &msqueueid-cond );pthread_mutex_unlock ( &msqueueid-mutex_cond );return length;从消息队列中接收消息：消息队列的接收函数msqueue_receive，其参数：id为消息队列数组的索引号，buff为消息内容，length为消息长度。int msqueue_receive ( id, buff, length )int id, length;caddr_t buff;caddr_t temp;int pos;if(id=MAXQUEUE)return MQERROR;if(length != msqueueid-elementLength)return MQERROR;/如果消息队列为空，则等待，直到消息队列为非空条件满足if ( msqueueid-curElementNum = 0)pthread_mutex_lock ( &msqueueid-mutex_cond );pthread_cond_wait ( &msqueueid-cond, &msqueueid-mutex_cond );pthread_mutex_unlock ( &msqueueid-mutex_cond );/取消息前，锁住消息队列缓冲区，以免其他线程存放或取消息pthread_mutex_lock ( &msqueueid-mutex_buff );/为符合消息队列FIFO特性，取出消息后，进行消息队列的调整temp =malloc(msqueueid-curElementNum-1)msqueueid-elementLength );bcopy ( &msqueueid-buff0, buff, msqueueid-elementLength );msqueueid-curElementNum -;bcopy ( &msqueueid-buffmsqueueid-elementLength, temp,msqueueid-elementLengthmsqueueid-curElementNum);bcopy ( temp, &msqueueid-buff0,msqueueid-elementLengthmsqueueid-curElementNum);free ( temp );/解除缓冲区锁pthread_mutex_unlock ( &msqueueid-mutex_buff );return length;多任务模型的实现在讨论完消息队列的创建、删除、发送和接收后，下面讲述消息队列在线程中的应用以实现多任务线程间的数据共享。首先在main主函数中创建消息队列和线程：/定义全局变量Int msqueue_record, msqueue_process;Void main()pthread_t pthreadID1;/创建消息队列，用于线程间通信msqueue_record = msqueue_create (“record”, 200, 200);msqueue_process = msqueue_create (“process”, 200, 200);/创建数据采集线程pthread_create ( &pthreadID1, NULL, receiveData, NULL);/创建数据处理线程pthread_create ( &pthreadID2, NULL, process, NULL);/创建数据记录线程pthread_create ( &pthreadID1, NULL, record, NULL);/等待进程结束wait_thread_end( );数据采集线程：voidreceiveData( )int count;unsigned char buff200;for(;) /从数据口采集数据，并将数据放置于buff中/wait_data_from_data_port( buff )/将数据写入消息队列msqueue_record中msqueue_send ( msqueue_record, buff, 200 );/将数据写入消息队列msqueue_process中msqueue_send ( msqueue_process, buff, 200 );记录线程函数：voidrecord ( )int num, count;unsigned char buffer200;for ( ; ) count = msqueue_receive ( msg_record, &buffer, 200 );if ( count 0) perror ( msgrcv in record);continue;/将取到的消息进行记录处理/record_message_to_lib();数据处理线程函数：int process( )int count;unsigned char buffer200;for ( ; ) count = msqueue_receive ( msg_