【socket】线程通信

【socket】线程通信

线程通信

• 介绍线程
• • 常用线程函数
  • 代码实现

介绍线程

线程是一条执行路径，是程序执行时的最小单位，它是进程的一个执行流，是CPU调度和分派的基本单位，一个进程可以由很多个线程组成，线程间共享进程的所有资源，每个线程有自己的堆栈和局部变量。线程由CPU独立调度执行，在多CPU环境下就允许多个线程同时运行。同样多线程也可以实现并发操作，每个请求分配一个线程来处理。

一个正在运行的软件(如迅雷)就是一个进程，一个进程可以同时运行多个任务( 迅雷软件可以同时下载多个文件，每个下载任务就是一个线程), 可以简单的认为进程是线程的集合。

线程是一条可以执行的路径。

对于单核CPU而言：多线程就是一个CPU在来回的切换，在交替执行。
对于多核CPU而言：多线程就是同时有多条执行路径在同时(并行)执行，每个核执行一个线程，多个核就有可能是一块同时执行的。

为什么要使用多线程
多线程可以提高程序的效率。

实际生活案例：村长要求喜洋洋在一个小时内打100桶水，可以喜洋洋一个小时只能打25桶水，如果这样就需要4个小时才能完成任务，为了在一个小时能够完成，喜洋洋就请美洋洋、懒洋洋、沸洋洋，来帮忙，这样4只羊同时干活，在一小时内完成了任务。原本用4个小时完成的任务现在只需要1个小时就完成了，如果把每只羊看做一个线程，多只羊即多线程可以提高程序的效率。

常用线程函数

int pthread_create(pthread_t *thread, const pthread_attr_t *attr,void *(*start_routine) (void *), void *arg);
功能:
用来创建一个线程
参数:
thread是一个pthread_t类型的指针，他用来返回该线程的线程ID。    
pthread_attr_t *attr是结构体,定义如下   
start_routine是一个函数指针,它指向的函数原型void *func(void*),这是创建的子线程要执行的任务(函数)   
arg是传给了所调用的函数的参数，如果有多个参数需要传递给子线程则需要封装到一个结构体里传进去  typedef struct
{int 	detachstate; 线程的分离状态int 	schedpolicy; 线程调度策略struct sched_param schedparam; 线程的调度参数int 	inheritsched; 线程的继承性int 	scope; 线程的作用域size_t 	guardsize; 线程栈末尾的警戒缓冲区大小int 	stackaddr_set;void * 	stackaddr; 线程栈的位置size_t 	stacksize; 线程栈的大小
}pthread_attr_t;

int pthread_join (pthread_t tid, void ** status);
功能:
主线程等待子线程的终止。也就是在子线程调用了pthread_join()方法后面的代码，只有等到子线程结束了才能执行
参数:
thread: 线程标识符，即线程ID，标识唯一线程。retval: 用户定义的指针，用来存储被等待线程的返回值。

当调用 pthread_join() 时，当前线程会处于阻塞状态，直到被调用的线程结束后，当前线程才会重新开始执行。当 pthread_join() 函数返回后，被调用线程才算真正意义上的结束，它的内存空间也会被释放（如果被调用线程是非分离的）

• 被释放的内存空间仅仅是系统空间，你必须手动清除程序分配的空间，比如 malloc() 分配的空间。
• 一个线程只能被一个线程所连接。
• 被连接的线程必须是非分离的，否则连接会出错。

作用:

• 用于等待其他线程结束：当调用 pthread_join() 时，当前线程会处于阻塞状态，直到被调用的线程结束后，当前线程才会重新开始执行。
• 对线程的资源进行回收：如果一个线程是非分离的（默认情况下创建的线程都是非分离）并且没有对该线程使用 pthread_join() 的话，该线程结束后并不会释放其内存空间，这会导致该线程变成了“僵尸线程”。

pthread_t pthread_self (void);
功能
获得线程自身的ID。
例如:
void* thread_func(void *arg)
{printf("thread id=%lun", pthread_self());return arg;
}
int main(void)
{pthread_t tid;pthread_create(&tid, NULL, thread_func, NULL);pthread_join(tid,NULL);return 0;
}

int pthread_detach (pthread_t tid);
功能:
pthread_detach用于是指定线程变为**分离**状态，就像进程脱离终端而变为后台进程类似

void pthread_exit (void *status);
功能:
终止线程
//status是 void* 类型的指针，可以指向任何类型的数据，它指向的数据将作为线程退出时的返回值。如果线程不需要返回任何数据，将status参数置为 NULL 即可。

return 语句和 pthread_exit() 函数的含义不同，return 的含义是返回，它不仅可以用于线程执行的函数，普通函数也可以使用；pthread_exit() 函数的含义是线程退出，它专门用于结束某个线程的执行。

int pthread_attr_init(pthread_attr_t *attr);功能：初始化一个线程属性对象 
参数：*attr 线程属性结构体指针变量 
返回值：0 - 成功，非0 - 失败 

int pthread_attr_destroy(pthread_attr_t *attr);功能：销毁一个线程属性对象 
参数：*attr 线程属性结构体指针变量 
返回值：0 - 成功，非0 - 失败 

int pthread_attr_setguardsize(pthread_attr_t *attr, size_t guardsize);功能：设置一个线程属性栈大小 
参数：*attr 线程属性结构体指针变量 guardsize设置一个保护区大小
返回值：0 - 成功，非0 - 失败 

int pthread_attr_setdetachstate(pthread_attr_t *attr,intdetachstate);功能：修改线程的分离状态属性
参数：*attr 线程属性结构体指针变量 , intdetachstate 线程的分离状态属性(PTHREAD_CREATE_DETACHED)
返回值：0 - 成功，非0 - 失败 

代码实现

#include <stdio.h>
#include <fcntl.h>
#include <unistd.h>
#include <signal.h>
#include <string.h>
#include <errno.h>
#include <getopt.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <pthread.h>#include "sqlite3.h"#define BACKLOG 13
typedef void *(THREAD_BODY) (void *thread_arg);void print_usage(char *progname);
void sig_stop(int signum);
int socket_listen(char *listen_ip, int port);
void sqlite_tem(char *buf);
int thread_start(pthread_t *thread_id ,THREAD_BODY* thread_workbody, void *thread_arg);
void* thread_worker(void *args);static int g_stop = 0;typedef struct worker_ctx_s
{pthread_mutex_t     lock;}worker_ctx_t;int main(int argc, char **argv)
{int                     rv;int                     ret;int                     opt;int                     idx;int                     port;int                     log_fd;int                     ch = 1;int                     daemon_run = 0;int                     ser_fd = -1;int                     cli_fd = -1;struct sockaddr_in      ser_addr;struct sockaddr_in      cli_addr;socklen_t               cliaddr_len = 20;worker_ctx_t            worker_ctx;pthread_t               tid;pthread_attr_t          thread_attr;struct option            opts[] = {{"daemon", no_argument, NULL, 'd'},{"port", required_argument, NULL, 'p'},{"help", no_argument, NULL, 'h'},{NULL, 0, NULL, 0}};while ((opt = getopt_long(argc, argv, "dp:h", opts, &idx)) != -1){switch (opt){case 'd':daemon_run = 1;break;case 'p':port = atoi(optarg);break;case 'h':print_usage(argv[0]);	return 0;}}if (!port){print_usage(argv[0]);return 0;}/*创建日志*/if (daemon_run){printf("Program %s is running at the background nown", argv[0]);log_fd = open("temper.log", O_CREAT | O_RDWR, 0666);if (log_fd < 0){printf("Open the logfile failure : %sn", strerror(errno));return 0;}dup2(log_fd, STDOUT_FILENO);dup2(log_fd, STDERR_FILENO);if ((daemon(1, 1)) < 0){printf("Deamon failure : %sn", strerror(errno));return 0;}}/*安装信号*/signal(SIGUSR1, sig_stop);/*初始化锁*/pthread_mutex_init(&worker_ctx.lock,NULL);/*socket创建*/if( (ser_fd = socket_listen(NULL, port)) < 0 ){printf("ERROR: %s server listen on serv_port %d failuren", argv[0], port);return -2;}printf("server start to listen on serv_port %dn",  port);while (!g_stop){cli_fd = accept(ser_fd, (struct sockaddr*) & cli_addr, &cliaddr_len);if (cli_fd < 0){printf("Accept the request from client failure:%sn", strerror(errno));continue;}thread_start(&tid ,thread_worker,(void*)cli_fd); /*调用子线程*/}/*销毁锁*/pthread_mutex_destroy(&worker_ctx.lock); close(ser_fd);return 0;
}/*创建子线程*/
void *thread_worker(void *ctx)
{int                 rv;char                buf[1024];int                 cli_fd;worker_ctx_t        s;if( !ctx){printf("Invalid input arguments in %sn");pthread_exit(NULL);}cli_fd = (int)ctx;printf("Child thread start to commuicate with n");while(1){pthread_mutex_lock(&s.lock); /*上锁*/memset(buf, 0, sizeof(buf));rv = read(cli_fd, buf, sizeof(buf));if( rv < 0){printf("Read data from client sockfd[%d] failure: %s and thread will exitn", cli_fd,strerror(errno));close(cli_fd);pthread_exit(NULL);}else if( rv == 0){printf("Socket[%d] get disconnected and thread will exit.n", cli_fd);close(cli_fd);pthread_exit(NULL);}else		{printf("n");printf("%sn",buf);sqlite_tem(buf);printf("Database inserted successfully!n");pthread_mutex_unlock(&s.lock); /*释放锁*/}}close(cli_fd);return 0;
}/*创建线程*/
int thread_start(pthread_t *thread_id ,THREAD_BODY* thread_workbody, void *thread_arg)
{int             rv = -1;pthread_attr_t  thread_attr;if(pthread_attr_init(&thread_attr)){printf("pthread_attr_init() failure :%sn", strerror(errno));goto cleanup;}if(pthread_attr_setstacksize(&thread_attr,120*1024)){printf("pthread_attr_setstacksize() failure:%sn",strerror(errno));goto cleanup;}if(pthread_attr_setdetachstate(&thread_attr,PTHREAD_CREATE_DETACHED)){printf("pthread_attr_setdetachstate() failure :%sn", strerror(errno));goto cleanup;}if(pthread_create(thread_id,&thread_attr, thread_workbody, thread_arg)){printf("create thread failure:%sn",strerror(errno));goto cleanup;}rv = 0;cleanup:pthread_attr_destroy(&thread_attr);return rv ;
}/*帮助信息*/
void print_usage(char* progname)
{printf("-d(--daemon):let program run in the background.n");printf("-p(--port):enter server port.n");printf("-h(--help):print this help information.n");return;
}/*信号函数*/
void sig_stop(int signum)
{if (SIGUSR1 == signum) {g_stop = 1;		}return;
}/*socket函数*/
int socket_listen(char *listen_ip, int port)
{int                     rv = 0;int                     on = 1;int                     ser_fd;struct sockaddr_in      servaddr;if ( (ser_fd = socket(AF_INET, SOCK_STREAM, 0)) < 0 ){   printf("Use socket() to create a TCP socket failure: %sn", strerror(errno));return -1; }   setsockopt(ser_fd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));memset(&servaddr, 0, sizeof(servaddr));servaddr.sin_family = AF_INET;servaddr.sin_port = htons(port);if( !listen_ip ){   servaddr.sin_addr.s_addr = htonl(INADDR_ANY);}   else{   if( inet_pton(AF_INET, listen_ip, &servaddr.sin_addr) <= 0 ){   printf("Inet_pton() set listen IP address failuren");rv = -2; goto cleanup;}   }if( bind(ser_fd, (struct sockaddr *) &servaddr, sizeof(servaddr)) < 0 ){   printf("Use bind() to bind the TCP socket failure: %sn", strerror(errno));rv = -3; goto cleanup;}if( listen(ser_fd, 64) < 0 ){   printf("Use bind() to bind the TCP socket failure: %sn", strerror(errno));rv = -4; goto cleanup;}   cleanup:if( rv < 0 )close(ser_fd);elserv = ser_fd;return rv;
}/*创建数据库*/
void sqlite_tem(char *buf)
{  int             nrow=0;int             ncolumn = 0;char          **azResult=NULL;int             rv; sqlite3        *db=NULL;char           *zErrMsg = 0;char            sql1[100];char           *ipaddr=NULL;char           *datetime=NULL;char           *temper=NULL;char           *sql = "create table if not exists temperature(ipaddr char(30), datetime char(50), temper  char(30))"; ipaddr = strtok(buf,"/");datetime = strtok(NULL, "/");temper = strtok(NULL, "/");rv = sqlite3_open("tempreture.db", &db);if(rv){printf("Can't open database:%sn", sqlite3_errmsg(db));sqlite3_close(db);return;}printf("opened a sqlite3 database named tempreture.db successfully!n");int ret = sqlite3_exec(db,sql, NULL, NULL, &zErrMsg);if(ret != SQLITE_OK){printf("create table fail: %sn",zErrMsg);}if(snprintf(sql1,sizeof(sql1), "insert into temper values('%s','%s','%s')", ipaddr, datetime, temper) < 0){printf("Failed to write datan");}//printf("Write data successfully!n");sqlite3_exec(db, sql1, 0, 0, &zErrMsg);sqlite3_free(zErrMsg);sqlite3_close(db);return;
}

使用多线程可以实现服务器和多个客户端得通信,使用多线程的方式，因为可以使用共享的全局变量，所以线程间的通信（数据交换）变得非常高效。
但是在Unix系统中，一个进程包含很多东西，包括可执行程序以及一大堆的诸如文件描述符地址空间等资源。在很多情况下，完成相关任务的不同代码间需要交换数据。
如果采用多进程的方式，进程的创建所花的时间片要比线程大些，另外进程间的通信比较麻烦，需要在用户空间和内核空间进行频繁的切换，开销很大。所以多线程服务器对于数据交换变得高效起来。