Golang语言社区--【游戏服务器知识】多线程并发

void *thread_entry( void * args );

void *thread_entry( void *args )
{
        int fd = *(int *)args ;
        do_handler_fd( fd );
}

for(;;){
    fd = accept();
    pthread_create(...,thread_entry,&fd);
}

void mc_thread_pool_add_task(void *task , size_t tasksize )

void *mc_thread_pool_get_task()

typedef struct _thread_pool_t
{
    pthread_mutex_t  queue_lock ;
    pthread_cond_t   task_cond  ;
    list_t         * tasks       // treat it as queue thread_task_t type
    pthread_t      * pthreads   ;
    int              isdestoried;
    int              workersnum ;
    char             ready      ;
    thread_task_handler  thread_pool_task_handler;
}thread_pool_t;
    /*
     *  this structure is a global control block of threads poll
     *  as you can see , queue_lock and task_cond is define to protecte access of this whole poll
     *  and task_cond is used to signal to threads that the task queue is ready
     *  tasks is a queue of tasks , each task should posted to this queue and threads
     *  in this pool can get it , we defined this task as void * to use wildly
     *  isdestoried is a boolean flag as his/her name
     *  workersnum is the total number of threads
     *  ready is a flag also and used to judge if the tasks queue is ready
     *  thread_pool_task_handler is a function point which points to the task handler you defined
     */

void mc_thread_pool_ini( mc_thread_pool_t * par_tp , int workersnum ,thread_task_handler par_handler )
{
    int err ;
    //par_tp = ( thread_pool_t *)malloc( sizeof(thread_pool_t) );
     
    if( par_tp == NULL )
    {
        fprintf( stderr , "thread_pool_t mallocn");
        return  ;
    }
    par_tp->workersnum = workersnum ;
     
    pthread_mutex_init( &par_tp->queue_lock ,NULL );
    pthread_cond_init(&par_tp->task_cond , NULL );
     
    /*
    par_tp->queue_lock = PTHREAD_MUTEX_INITIALIZER ;
    par_tp->task_cond  = PTHREAD_COND_INITIALIZER  ;
    */
    par_tp->tasks = mc_listcreate() ;
    if( par_tp->tasks == NULL )
    {
        fprintf( stderr , "listcreate() errorn");
        //free( par_tp ) ;
        return  ;
    }
     
    par_tp->pthreads = ( pthread_t *)malloc( sizeof( pthread_t )*workersnum );
     
    if( par_tp->pthreads == NULL )
    {
        fprintf( stderr , "pthreads mallocn");
        //free( par_tp );
        mc_freelist( par_tp->tasks ) ;
        return NULL ;
    }
     
    int i = 0 ;
    for( ; i < workersnum ; i++ )
    {
        fprintf(stderr,"start to create threadsn");
        err = pthread_create(&(par_tp->pthreads[i]),NULL,mc_thread_entry,NULL) ;
        if( err == -1 )
        {
            fprintf( stderr , "pthread_create errorn");
            //free( par_tp );
            mc_freelist( par_tp->tasks ) ;
            free(par_tp->pthreads) ;
        }
    }
     
    par_tp->thread_pool_task_handler = par_handler ;
    par_tp->ready = 0 ;
    fprintf(stderr,"successed to create threadsn");
}

static void *mc_thread_entry( void *args )
{
    void * task ;
    for(;;)
    {
        pthread_mutex_lock( &mc_global_threads_pool.queue_lock ) ;
        fprintf(stderr, " locked to wait taskn");
        while( mc_global_threads_pool.ready == 0 )
        {
            pthread_cond_wait( &mc_global_threads_pool.task_cond , &mc_global_threads_pool.queue_lock ) ;
        }
        task = mc_thread_pool_get_task() ;
        fprintf(stderr, "get a task and ready to unlock n");
        pthread_mutex_unlock( &mc_global_threads_pool.queue_lock ) ;
        mc_global_threads_pool.thread_pool_task_handler( task ) ;
    }
}

void mc_thread_pool_add_task(void *task , size_t tasksize )
{
    pthread_mutex_lock( &mc_global_threads_pool.queue_lock );
     
    fprintf( stderr ,"thread locked and append to listn");
     
    mc_list_append( mc_global_threads_pool.tasks , task , tasksize ) ;
     
    pthread_mutex_unlock( &mc_global_threads_pool.queue_lock );
     
    fprintf( stderr ,"thread unlocked and successed append to listn");
     
    mc_global_threads_pool.ready = 1 ;
     
    if( mc_global_threads_pool.ready == 1 )
    {
        fprintf( stderr ,"signal to threadsn");
        pthread_cond_signal( &mc_global_threads_pool.task_cond ) ;
    }
}

void *mc_thread_pool_get_task()
{
    void * ret_task ;
    ret_task = mc_getnode_del( mc_global_threads_pool.tasks , 0 );
    if( ret_task == NULL )
    {
        fprintf(stderr,"get node_del errorn");
    }
    fprintf( stderr ," got a taskn");
    mc_global_threads_pool.ready = 0 ;
    if( ret_task == NULL )
    {
        fprintf(stderr, "getnode_del errorn");
        return NULL ;
    }
    else
        return ret_task ;
}<br><br>

typedef struct _thread_task_t
{
    int     task_num ;
}mc_thread_task_t ;

void my_thread_task_handler( void * task )
{
 
    fprintf(stderr,"task->tasknum %dn",((mc_thread_task_t *)task)->task_num );
     
    /*
     *  if the task is a event we can like this demo:
     *  (event_t *)task->handler( (event_t *)task );
     *  so in event_t structure there should be a callback called handler
     */
}

int main()
{
    mc_thread_task_t ltask;
    ltask.task_num = 1 ;
    fprintf(stderr,"begin to ini pooln");
    mc_thread_pool_ini( &mc_global_threads_pool , 20 , my_thread_task_handler );
    mc_thread_pool_add_task( &ltask , sizeof(mc_thread_task_t) );
    int i = 0 ;
    for(;i < 10000; i++)
    {
        ltask.task_num = i ;
        mc_thread_pool_add_task( &ltask , sizeof(mc_thread_task_t) );
        sleep(1);
    }
    return 0;
}

mc_thread_pool_ini( &mc_global_threads_pool , N , task_handler );  // 第二个参数为线程池工作线程数
 
for(;;)
{
    fd = accept();
    struct task * newtask = ( struct task *)malloc( sizeof(struct task) );
    newtask->fd = fd ;
    mc_thread_pool_add_task( &newtask,sizeof(struct task*) ); //将newtask 指针加入队列，而不是实例，可以减少队列的存储空间
}