Linux基础——多线程实现任务

这里，我们首先要实现一种数据结构，将相应的任务，线程的fd，还有队列实现。

声明代码如下:

#ifndef _HEAD_H
#define _HEAD_H
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/select.h>
#include <sys/types.h>
#include <fcntl.h>
#include <pthread.h>
#include <sys/time.h>
#include <signal.h>
#define MSG_LEN 1024
#define TASK_CNT 1024
extern pthread_mutex_t lock;
extern pthread_cond_t cond1;
extern pthread_cond_t cond2;
typedef struct tag_fd
{
    int s_rfd;
    int s_wfd;
    struct tag_fd *next;
}FD,*pFD;
typedef struct tag_task
{
    char s_msg[MSG_LEN];
    int s_fd;
}TASK,*pTASK;
typedef struct tag_que
{
    TASK arr[TASK_CNT+1];
    int front;
    int tail;
}QUEUE,*pQUEUE;
void fd_insert(pFD *phead,int rfd,int wfd);
void fd_init(pFD *phead);
int  fd_find(pFD phead,int rfd);
void fd_del(pFD *phead,int rfd);
void add_task(pQUEUE pq,pTASK pt);
void get_task(pQUEUE pq,pTASK pt);
void excute_task(pTASK pt);
#endif

我们需要根据线程的占用情况，控制好，所以我们应把线程插入到一个链表中。
实现代码如下:

#include "head.h"
void fd_init(pFD *phead)
{
    *phead= NULL;
}
void fd_insert(pFD *phead,int rfd,int wfd)
{
    pFD pnew = (pFD )calloc(1,sizeof(FD));
    pnew->s_rfd=rfd;
    pnew->s_wfd=wfd;
    pnew->next = *phead;
    *phead = pnew;
}
int fd_find(pFD phead,int rfd)
{
    while(phead)
    {
        if(phead->s_rfd==rfd)
            break;
        else
            phead = phead->next;
    }
    if(phead == NULL)
        return -1;
    else
        return phead->s_wfd;
}

void fd_del(pFD *phead,int rfd)
{
    pFD pcur,ppre;
    pcur=*phead;
    ppre=NULL;
    while(pcur)
    {
        if(pcur->s_rfd == rfd)
            break;
        else
        {
            ppre=pcur;
            pcur = pcur ->next;
        }
    }
    if(ppre==NULL)
    {
        *phead=pcur->next;
        free(pcur);
        pcur=NULL;
    }
    else
    {
        ppre->next=pcur->next;
        free(pcur);
        pcur=NULL;
    }
}

然后，我们还需要实现对任务的控制，例如任务的添加、获得、执行等。

实现代码如下:

#include "head.h"
static int que_empty(pQUEUE pq)
{
    return pq->front == pq->tail;
}
static int que_full(pQUEUE pq)
{
    return (pq->tail+1)%(TASK_CNT+1)==pq->front;
}
static int que_cnt(pQUEUE pq)
{
    return (pq->tail - pq->front +TASK_CNT+1)%(TASK_CNT + 1);
}
void add_task(pQUEUE pq ,pTASK pt)
{
    pthread_mutex_lock(&lock);
    while(que_full(pq))
        pthread_cond_wait(&cond1,&lock);
    pq->arr[pq->tail]=*pt;
    pq->tail = (pq->tail+1)%(TASK_CNT+1);
    if(que_cnt(pq)==1)
        pthread_cond_broadcast(&cond2);
    printf("添加了一个任务!!
");
    pthread_mutex_unlock(&lock);
}
void get_task(pQUEUE pq ,pTASK pt)
{
    pthread_mutex_lock(&lock);
    while(que_empty(pq))
        pthread_cond_wait(&cond2,&lock);
    *pt=pq->arr[pq->front];
    pq->front = (pq->front+1)%(TASK_CNT+1);
    if(que_cnt(pq)== TASK_CNT -1)
        pthread_cond_broadcast(&cond1);
    printf("获得了一个任务!!
");
    pthread_mutex_unlock(&lock);
}


void excute_task(pTASK pt)
{
    char buf[1024];
    memset(buf,0,1024);
    strcpy(buf,pt->s_msg);
    int index;
    for(index=0;index < strlen(buf);index++)
        buf[index]=toupper(buf[index]);
    buf[index]='';
    write(pt -> s_fd,buf,strlen(buf));
}

最后，我们只需在服务器端应用select循环查询是否有任务，再执行相应的操作。

服务器实现代码如下:

#include "head.h"
pthread_mutex_t lock;
pthread_cond_t cond1,cond2;
void* hand(void* arg)
{
    pthread_detach(pthread_self());
    TASK task;
    pQUEUE pq = (pQUEUE)arg;
    while(1)
    {
        get_task(pq,&task);
        excute_task(&task);
        sleep(1);
    }
}
int main(int argc,char *argv[])
{
    if(argc != 3)
    {
        perror("参数错误!!
");
        exit(1);
    }
    signal(SIGINT,SIG_IGN);
    signal(SIGPIPE,SIG_IGN);
    signal(SIGQUIT,SIG_IGN);
    QUEUE que;
    int fd;
    fd_set read_set,revc;
    pFD list;
    memset(&que,0,sizeof(QUEUE));
    fd_init(&list);
    int cnt = atoi(argv[2]);
    pthread_t *arr=(pthread_t *)calloc(cnt,sizeof(pthread_t));
    pthread_mutex_init(&lock,NULL);
    pthread_cond_init(&cond1,NULL);
    pthread_cond_init(&cond2,NULL);
    int index=0;
    while(cnt > 0)
    {
        pthread_create(arr+index,NULL,hand,(void*)&que);
        cnt--;
        index++;
    }
    fd = open(argv[1],O_RDONLY);
    if(fd == -1)
    {
        perror("管道打开失败!!
");
        exit(1);
    }
    struct timeval tm;
    int ret;
    FD_ZERO(&read_set);
    FD_ZERO(&revc);
    FD_SET(fd,&read_set);
    while(1)
    {
        tm.tv_sec=0;
        tm.tv_usec=1000;
        revc = read_set;
        ret=select(1024,&revc,NULL,NULL,&tm);
        if(ret == 0)
            continue;
        else if(ret > 0)
        {
            if(FD_ISSET(fd,&revc))
            {
                char buf[32];
                memset(buf,0,32);
                if(read(fd,buf,32)==0)
                    continue;
                else
                {
                    char name[32];
                    int r_fd,w_fd;
                    buf[strlen(buf)-1]='';
                    memset(name,0,32);
                    sprintf(name,"r.%s",buf);
                    w_fd=open(name,O_WRONLY);
                    memset(name,0,32);
                    sprintf(name,"w.%s",buf);
                    r_fd=open(name,O_RDONLY);
                    fd_insert(&list,r_fd,w_fd);
                    FD_SET(r_fd,&read_set);
                }
            }
        }
        pFD pcur=list;
        while(pcur)
        {
            if(FD_ISSET(pcur->s_rfd,&revc))
            {
                char buf[1024];
                memset(buf,0,1024);
                if(read(pcur->s_rfd,buf,1024)==0)
                {
                    FD_CLR(pcur->s_rfd,&read_set);
                    int i=pcur->s_rfd;
                    pcur=pcur->next;
                    fd_del(&list,i);
                }
                else
                {
                    TASK tk;
                    memset(&tk,0,sizeof(TASK));
                    tk.s_fd=pcur->s_wfd;
                    strcpy(tk.s_msg,buf);
                    add_task(&que,&tk);
                    pcur=pcur->next;
                }
            }
            else
                pcur=pcur->next;
        }
    }
    pthread_mutex_destory(&lock);
    pthread_cond_destory(&cond1);
    pthread_cond_destory(&cond2);
    return 0;    
}

客户端实现代码如下:

#include <stdio.h>
#include <stdlib.h>
#include <fcntl.h>
#include <unistd.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
int main(int argc,char *argv[])
{
    int fd_server,send,revc;
    char rname[32],wname[32];
    memset(rname,0,32);
    memset(wname,0,32);
    sprintf(rname,"r.%d",getpid());
    sprintf(wname,"w.%d",getpid());
    mkfifo(rname,0666);
    mkfifo(wname,0666);
    fd_server=open(argv[1],O_WRONLY);
    char msg[1024];
    memset(msg,0,1024);
    sprintf(msg,"%d
",getpid());
    write(fd_server,msg,strlen(msg));
    revc=open(rname,O_RDONLY);
    send=open(wname,O_WRONLY);
    while(memset(msg,0,1024),fgets(msg,1024,stdin))
    {
        write(send,msg,strlen(msg));
        memset(msg,0,1024);
        read(revc,msg,1024);
        write(1,msg,strlen(msg));
    }
    close(fd_server);
    close(send);
    close(revc);
    unlink(rname);
    unlink(wname);
    return 0;
}