1. Johngo学长首页
  2. 技术点详细复盘专题
  3. 大数据

【socket】线程通信-服务器上报温度

• 大数据 • 阅读 25

介绍线程

线程是一条执行路径，是程序执行时的最小单位，它是进程的一个执行流，是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=%lu\n", 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);
功能:
终止线程

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
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include

#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 now\n", argv[0]);

        log_fd = open("temper.log", O_CREAT | O_RDWR, 0666);
        if (log_fd < 0)
        {
            printf("Open the logfile failure : %s\n", strerror(errno));
            return 0;
        }

        dup2(log_fd, STDOUT_FILENO);
        dup2(log_fd, STDERR_FILENO);

        if ((daemon(1, 1)) < 0)
        {
            printf("Deamon failure : %s\n", strerror(errno));
            return 0;
        }
    }

    signal(SIGUSR1, sig_stop);

    pthread_mutex_init(&worker_ctx.lock,NULL);

    if( (ser_fd = socket_listen(NULL, port)) < 0 )
    {
        printf("ERROR: %s server listen on serv_port %d failure\n", argv[0], port);
        return -2;
    }
    printf("server start to listen on serv_port %d\n",  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:%s\n", 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 %s\n");
        pthread_exit(NULL);
    }

    cli_fd = (int)ctx;

    printf("Child thread start to commuicate with socket client...\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 exit\n", 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("%s\n",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 :%s\n", strerror(errno));
        goto cleanup;
    }

    if(pthread_attr_setstacksize(&thread_attr,120*1024))
    {
        printf("pthread_attr_setstacksize() failure:%s\n",strerror(errno));
        goto cleanup;
    }

    if(pthread_attr_setdetachstate(&thread_attr,PTHREAD_CREATE_DETACHED))
    {
        printf("pthread_attr_setdetachstate() failure :%s\n", strerror(errno));
        goto cleanup;
    }

    if(pthread_create(thread_id,&thread_attr, thread_workbody, thread_arg))
    {
        printf("create thread failure:%s\n",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;
}

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: %s\n", 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 failure\n");
            rv = -2;
            goto cleanup;
        }
    }

    if( bind(ser_fd, (struct sockaddr *) &servaddr, sizeof(servaddr)) < 0 )
    {
        printf("Use bind() to bind the TCP socket failure: %s\n", strerror(errno));
        rv = -3;
        goto cleanup;
    }

    if( listen(ser_fd, 64) < 0 )
    {
        printf("Use bind() to bind the TCP socket failure: %s\n", strerror(errno));
        rv = -4;
        goto cleanup;
    }

cleanup:
    if( rv < 0 )
        close(ser_fd);
    else
        rv = 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:%s\n", 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: %s\n",zErrMsg);
    }

    if(snprintf(sql1,sizeof(sql1), "insert into temper values('%s','%s','%s')", ipaddr, datetime, temper) < 0)
    {
        printf("Failed to write data\n");
    }

    sqlite3_exec(db, sql1, 0, 0, &zErrMsg);
    sqlite3_free(zErrMsg);
    sqlite3_close(db);
    return;
}

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

Original: https://blog.csdn.net/weixin_45942238/article/details/122625168
Author: 夜里生花
Title: 【socket】线程通信-服务器上报温度

原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/816570/

转载文章受原作者版权保护。转载请注明原作者出处！

赞 (0)
0

【自取】最近整理的，有需要可以领取学习：



Linux核心资料大放送~

全栈面试题汇总（持续更新&可下载）

一个提高学习100%效率的工具！

【超详细】深度学习面试题目！

LeetCode Python刷题答案下载！

LeetCode Java版刷题答案下载！

LeetCode C++ 版本，抓紧保存！

LeetCode GO语言 刷题答案下载！

大家都在看

亲爱的 Coder【最近整理，可免费获取】👉 最新必读书单  | 👏 面试题下载  | 🌎 免费的AI知识星球