多线程和线程同步(Linux/C)

这是 C 语言版的，C++ 版 https://xingzhu.top/archives/duo-xian-cheng-xian-cheng-chi

线程基础

  • 进程有自己独立的地址空间，多个线程共用同一个地址空间
  • 线程是程序的最小执行单位，进程是操作系统中最小的资源分配单位
  • CPU 划分时间片，多个线程抢占时间片执行
  • 线程的上下文切换比进程要快的多
    • 上下文切换是指继续上次线程没执行完的部分，接着执行后续的操作
  • Linux 看来，线程就是轻量版的进程，但是 Windows 不是这样的
  • 父线程创建的子线程，子线程之间共享堆区、全局区、代码区，但是栈区和寄存器是各自独有

线程基本函数

在一个 main 函数中，是一个进程，此时创建线程后，这个进程变为了父线程和子线程

创建线程

1
pthread_t pthread_self(void);	// 返回当前线程的线程ID
  • 这个 ID 是一个无符号长整型形数
1
2
3
4
#include <pthread.h>
int pthread_create(pthread_t *thread, const pthread_attr_t *attr,
                   void *(*start_routine) (void *), void *arg);
// Compile and link with -pthread, 线程库的名字叫pthread, 全名: libpthread.so libptread.a

参数:

  • thread: 传出参数，是无符号长整形数，线程创建成功, 会将子线程 ID 写入到这个指针指向的内存中
  • attr: 线程的属性, 一般情况下使用默认属性即可, 写 NULL
  • start_routine: 函数指针，创建出的子线程的处理动作，也就是该函数在子线程中执行
  • arg: 作为实参传递到 start_routine 指针指向的函数内部
  • 返回值：线程创建成功返回 0，创建失败返回对应的错误号

线程退出

1
2
#include <pthread.h>
void pthread_exit(void *retval);
  • 参数: 线程退出的时候携带的数据，当前子线程的主线程会得到该数据。如果不需要使用，指定为 NULL
  • 只要调用该函数当前线程就马上退出了，并且不会影响到其他线程的正常运行，其他线程照常进行，父进程退出，子线程同理会继续执行（如果分离线程了，不分离线程，虽然也可执行，但是会造成内存泄漏，子线程资源得不到收回）

线程回收

1
2
3
4
#include <pthread.h>
// 这是一个阻塞函数, 子线程在运行这个函数就阻塞
// 子线程退出, 函数解除阻塞, 回收对应的子线程资源, 类似于回收进程使用的函数 wait()
int pthread_join(pthread_t thread, void **retval);
  • 参数:
    • thread: 要被回收的子线程的线程 ID
    • retval: 二级指针, 指向一级指针的地址, 是一个传出参数, 这个地址中存储了 pthread_exit() 传递出的数据，如果不需要这个参数，可以指定为 NULL
  • 返回值：线程回收成功返回 0，回收失败返回错误号
  • 如果还有子线程在运行，调用该函数就会阻塞，子线程退出函数解除阻塞进行资源的回收，函数被调用一次，只能回收一个子线程，如果有多个子线程则需要循环进行回收

void * 是指任何类型的都可以指向

示例

使用主线程栈方式获取子线程数据

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
#include <stdio.h>  
#include <string.h>  
#include <pthread.h>  
  
// 定义结构  
struct Persion {  
    int id;  
    char name[36];  
    int age;  
};  
  
// 子线程的处理代码  
void* working(void* arg)  
{  
    struct Persion* p = (struct Persion*)arg;  
    printf("我是子线程, 线程ID: %ld\n", pthread_self());  
    for(int i=0; i<9; ++i)  
    {  
        printf("child == i: = %d\n", i);  
        if(i == 6)  
        {  
            // 使用主线程的栈内存  
            p->age  =12;  
            strcpy(p->name, "tom");  
            p->id = 100;  
            // 该函数的参数将这个地址传递给了主线程的pthread_join()  
            pthread_exit(p);  
        }  
    }  
    return NULL;  
}  
  
int main()  
{  
    // 1. 创建一个子线程  
    pthread_t tid;  
  
    struct Persion p;  
    // 主线程的栈内存传递给子线程  
    pthread_create(&tid, NULL, working, &p);  
  
    printf("子线程创建成功, 线程ID: %ld\n", tid);  
    // 2. 子线程不会执行下边的代码, 主线程执行  
    printf("我是主线程, 线程ID: %ld\n", pthread_self());  
    for(int i = 0; i < 3; ++i) {  
        printf("i = %d\n", i);  
    }  
  
    // 阻塞等待子线程退出，获取子线程数据
    // 方式一
    void* ptr = NULL;
    pthread_join(tid, &ptr);  
    struct Persion* ptr2 = (struct Persion*)ptr; 
    printf("name: %s, age: %d, id: %d\n", ptr2->name, ptr2->age, ptr2->id);  

    // 方式二
    void* ptr = NULL;
    void **ptr1 = &ptr;  
    pthread_join(tid, ptr1);  
    struct Persion** ptr3 = (struct Persion**)ptr1;
    printf("name: %s, age: %d, id: %d\n", (*ptr3)->name, (*ptr3)->age, (*ptr3)->id); 
  
    // 方式一或二执行后都可使用的 
    printf("name: %s, age: %d, id: %d\n", p.name, p.age, p.id);  
    printf("子线程资源被成功回收...\n");  
    return 0;  
}

总结

  • 首先在主线程访问的子线程资源，一定要是有效的，子线程定义在栈区的资源会被释放，这种就可以在主线程传地址进去，还可全局变量
  • 这里以传地址形式示例返回值使用教程
  • 要明白这两个线程函数（退出和回收的机制），当传递 void **retval 时，pthread_join 可以通过解引用这个指针来修改它指向的一级指针，从而将子线程返回的指针传递回主线程，本质是修改了一级指针指向的（存的）地址值
  • 所以要用二级指针，一级指针就不能修改地址值了，不灵活了，然后这个指针就能访问子线程数据了

上述案例可能不太好，因为可以直接调用，是从主线程传地址进去的 但使用返回值可以访问子线程创建在堆区的数据，这是一个使用场景

线程分离

1
2
3
#include <pthread.h>
// 参数就子线程的线程ID, 主线程就可以和这个子线程分离了
int pthread_detach(pthread_t thread);

  • 如果让主线程负责子线程的资源回收，调用 pthread_join() 只要子线程不退出主线程就会一直被阻塞，主要线程的任务也就不能被执行了

  • thread_detach()，调用这个函数之后指定的子线程就可以和主线程分离，当子线程退出的时候，其占用的内核资源就被系统的其他进程接管并回收了

  • 此时主线程可以执行其他的事物或者退出当前线程了

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
// 子线程的处理代码  
void* working(void* arg) 
{  
    printf("我是子线程, 线程ID: %ld\n", pthread_self());  
    for(int i = 0; i < 20; ++i) {  
        printf("child == i: = %d\n", i);  
    }  
    return NULL;  
}  
  
int main()  
{  
    // 1. 创建一个子线程  
    pthread_t tid;  
    pthread_create(&tid, NULL, working, NULL);  
    printf("子线程创建成功, 线程ID: %ld\n", tid);  
  
    // 设置子线程和主线程分离  
    pthread_detach(tid);  
  
    // 让主线程自己退出即可  
    pthread_exit(NULL);  
  
    return 0;  
}
  • 程序会打印完子线程的内容，然后被回收
  • 注意，若不调用 pthread_exit(NULL); ，父线程声明周期结束，那么内存会被释放，而子线程是使用的父线程内存，所以子线程会被终止
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
// 子线程的处理代码  
void* working(void* arg)  
{  
    printf("我是子线程, 线程ID: %ld\n", pthread_self());  
    for(int i = 0; i < 10000; ++i)  {  
        printf("child == i: = %d\n", i);  
    }  
    return NULL;  
}  
  
int main()  
{  
    // 1. 创建一个子线程  
    pthread_t tid;  
    pthread_create(&tid, NULL, working, NULL);  
    printf("子线程创建成功, 线程ID: %ld\n", tid);  
  
    // 设置子线程和主线程分离  
    pthread_detach(tid);  
    usleep(1000); // 休眠 1000us--->1ms
    
    return 0;  
}
  • 上述休眠只是为了演示，看出此时子线程还为执行完毕就终止了
  • 这个案例看出父线程终止，子线程也会随之终止

线程同步

互斥锁

1
2
// 创建互斥锁
pthread_mutex_t  mutex;
1
2
3
4
5
6
7
8
// 初始化互斥锁
// restrict: 是一个关键字, 用来修饰指针, 只有这个关键字修饰的指针可以访问指向的内存地址, 其他指针是不行的
// 利于定义 p = mutex p不能操控这个内存
int pthread_mutex_init(pthread_mutex_t *restrict mutex,
           const pthread_mutexattr_t *restrict attr);

// 释放互斥锁资源            
int pthread_mutex_destroy(pthread_mutex_t *mutex);
  • mutex: 互斥锁变量的地址
  • attr: 互斥锁的属性， 一般使用默认属性即可，这个参数指定为 NULL
1
2
3
4
5
6
7
8
// 加锁
// 不成功会一直阻塞
int pthread_mutex_lock(pthread_mutex_t *mutex);

// 尝试加锁
// 如果这把锁没有被锁定是打开的，线程加锁成功
// 如果锁变量被锁住了，调用这个函数加锁的线程，不会被阻塞，加锁失败直接返回错误号
int pthread_mutex_trylock(pthread_mutex_t *mutex);

不是所有的线程都可以对互斥锁解锁，哪个线程加的锁, 哪个线程才能解锁成功

示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
#include <pthread.h>  

// 全局变量  
int cur;  
  
// 创建一把互斥锁  
// 全局变量, 多个线程共享  
pthread_mutex_t mutex;  
  
// 线程处理函数  
void* funcA_num(void* arg)  
{  
    pthread_mutex_lock(&mutex);  
    cur++;  
    pthread_mutex_unlock(&mutex);  
    return NULL;  
}  
  
void* funcB_num(void* arg)  
{  
    pthread_mutex_lock(&mutex);  
    cur++;  
    pthread_mutex_unlock(&mutex);  
    return NULL;  
}  
  
int main(int argc, const char* argv[])  
{  
    pthread_t p1, p2;  
  
    // 初始化互斥锁  
    pthread_mutex_init(&mutex, NULL);  
  
    // 创建两个子线程  
    pthread_create(&p1, NULL, funcA_num, NULL);  
    pthread_create(&p2, NULL, funcB_num, NULL);  
  
    // 阻塞，资源回收  
    pthread_join(p1, NULL);  
    pthread_join(p2, NULL);  
  
    // 销毁互斥锁  
    // 线程销毁之后, 再去释放互斥锁  
    pthread_mutex_destroy(&mutex);  

    return 0;  
}

读写锁

概述

1
注意是一把锁，只是可以锁定读操作和写操作
1
pthread_rwlock_t rwlock;

锁的记录

  • 锁的状态: 锁定/打开
  • 锁定的是什么操作: 读操作/写操作，使用读写锁锁定了读操作，需要先解锁才能去锁定写操作，反之亦然
  • 哪个线程将这把锁锁上了

读写锁特点：

  • 使用读写锁的读锁锁定了临界区，线程对临界区的访问是并行的，读锁是共享的
  • 使用读写锁的写锁锁定了临界区，线程对临界区的访问是串行的，写锁是独占的
  • 使用读写锁分别对两个临界区加了读锁和写锁，两个线程要同时访问者两个临界区，访问写锁临界区的线程继续运行，访问读锁临界区的线程阻塞，因为 写锁比读锁的优先级高

线程对共享资源写操作，读写锁和互斥锁一样，读写锁没有优势 线程对共享资源有读操作和写操作，且读操作较多，读写锁优势明显

函数

1
2
3
4
5
6
7
8
#include <pthread.h>
pthread_rwlock_t rwlock;
// 初始化读写锁
int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock,
           const pthread_rwlockattr_t *restrict attr);

// 释放读写锁占用的系统资源
int pthread_rwlock_destroy(pthread_rwlock_t *rwlock);
  • rwlock: 读写锁的地址，传出参数
  • attr: 读写锁属性，一般使用默认属性，指定为 NULL
1
2
// 在程序中对读写锁加读锁, 锁定的是读操作
int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);

调用这个函数，如果读写锁是打开的，那么加锁成功；如果读写锁已经锁定了读操作，调用这个函数依然可以加锁成功，因为读锁是共享的；如果读写锁已经锁定了写操作，调用这个函数的线程会被阻塞

1
2
// 在程序中对读写锁加写锁, 锁定的是写操作
int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);

调用这个函数，如果读写锁是打开的，那么加锁成功；如果读写锁已经锁定了读操作或者锁定了写操作，调用这个函数的线程会被阻塞

不阻塞的加锁

1
2
3
4
5
6
// 这个函数可以有效的避免死锁
// 如果加读锁失败, 不会阻塞当前线程, 直接返回错误号
int pthread_rwlock_tryrdlock(pthread_rwlock_t *rwlock);

// 如果加写锁失败, 不会阻塞当前线程, 直接返回错误号
int pthread_rwlock_trywrlock(pthread_rwlock_t *rwlock);
1
2
// 解锁, 不管锁定了读还是写都可用解锁
int pthread_rwlock_unlock(pthread_rwlock_t *rwlock);

示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
#include <stdio.h>  
#include <stdlib.h>  
#include <unistd.h>  
#include <pthread.h>  
  
// 全局变量  
int number = 0;  
  
// 定义读写锁  
pthread_rwlock_t rwlock;  
  
// 写的线程的处理函数  
void* writeNum(void* arg)  
{  
    for(int i = 0; i < 100; i++) {  
        pthread_rwlock_wrlock(&rwlock);  
        number++;  
        printf("写操作: tid = %ld, number : %d\n", pthread_self(), number);  
        pthread_rwlock_unlock(&rwlock);  
        // 添加sleep目的是要看到多个线程交替工作  
        usleep(rand() % 5000);  
    }
  
    return NULL;  
}  
  
void* readNum(void* arg)  
{  
    for(int i = 0; i < 100; i++) {  
        pthread_rwlock_rdlock(&rwlock);  
        printf("读操作: tid = %ld, number = %d\n", pthread_self(), number);  
        pthread_rwlock_unlock(&rwlock);  
        usleep(rand() % 5);  
    }  
    return NULL;  
}  
  
int main()  
{  
    // 初始化读写锁  
    pthread_rwlock_init(&rwlock, NULL);  
  
    // 3个写线程, 5个读的线程  
    pthread_t wtid[3];  
    pthread_t rtid[5];  
    for(int i = 0; i < 3; ++i) {  
        pthread_create(&wtid[i], NULL, writeNum, NULL);  
    }
    for(int i=0; i<5; ++i) {  
        pthread_create(&rtid[i], NULL, readNum, NULL);  
    } 
  
    // 释放资源  
    for(int i = 0; i < 3; ++i) {  
        pthread_join(wtid[i], NULL);  
        pthread_join(rtid[i], NULL);  
    }
  
    // 销毁读写锁  
    pthread_rwlock_destroy(&rwlock);  
    return 0;  
}
1
2
3
4
5
6
7
8
9
10
// 一部分可能的结果
写操作: tid = 1, number : 1
写操作: tid = 1, number : 2
写操作: tid = 2, number : 3
写操作: tid = 2, number : 4
写操作: tid = 3, number : 5
写操作: tid = 3, number : 6
读操作: tid = 4, 全局变量number = 6
读操作: tid = 4, 全局变量number = 6
读操作: tid = 4, 全局变量number = 6

条件变量

函数

用于阻塞线程和唤醒线程

1
2
3
4
5
6
7
8
#include <pthread.h>
pthread_cond_t cond;
// 初始化
int pthread_cond_init(pthread_cond_t *restrict cond,
      const pthread_condattr_t *restrict attr);

// 销毁释放资源        
int pthread_cond_destroy(pthread_cond_t *cond);
1
2
// 线程阻塞函数, 哪个线程调用这个函数, 哪个线程就会被阻塞
int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex);
  • 这个函数执行后，如果阻塞了，如果当前上了锁，会把当前锁打开，防止后续的死锁，比如生产者要生产等
  • 当线程解除阻塞的时候，函数内部会帮助这个线程再次将这个 mutex 互斥锁锁上，继续向下访问临界区
  • 被唤醒后，如果抢到了互斥锁控制权，则继续执行这个 pthread_cond_wait 之后的执行体
  • 因此这个条件变量的顺序只能是在加锁的后面，也就是临界区内，因为会自动解锁和加锁，如果实现在加锁前，就相当于二次加锁，会导致死锁现象
1
2
3
4
5
6
7
8
9
// 表示的时间是从1971.1.1到某个时间点的时间, 总长度使用秒/纳秒表示
struct timespec {
    time_t tv_sec;      /* Seconds */
    long   tv_nsec;     /* Nanoseconds [0 .. 999999999] */
};

// 将线程阻塞一定的时间长度, 时间到达之后, 线程就解除阻塞了
int pthread_cond_timedwait(pthread_cond_t *restrict cond,
           pthread_mutex_t *restrict mutex, const struct timespec *restrict abstime);
  • 第三个参数表示线程阻塞的时长，如果条件变量达到要求，线程不阻塞
  • 如果不达要求，会阻塞，但是如果超过阻塞时长，还为达到要求，返回 -1
  • 需要额外注意一点：struct timespec 这个结构体中记录的时间是从 1971.1.1 到某个时间点的时间，总长度使用秒/纳秒表示
1
2
3
4
time_t mytim = time(NULL);	// 1970.1.1 0:0:0 到当前的总秒数
struct timespec tmsp;
tmsp.tv_nsec = 0;
tmsp.tv_sec = time(NULL) + 100;	// 线程阻塞100s
1
2
3
4
// 唤醒阻塞在条件变量上的线程, 至少有一个被解除阻塞
int pthread_cond_signal(pthread_cond_t *cond);
// 唤醒阻塞在条件变量上的线程, 被阻塞的线程全部解除阻塞
int pthread_cond_broadcast(pthread_cond_t *cond);
  • pthread_cond_signal 是唤醒至少一个被阻塞的线程（总个数不定）
  • pthread_cond_broadcast 是唤醒所有被阻塞的线程

生产者消费者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <pthread.h>

// 链表的节点
struct Node {
    int number;
    struct Node* next;
};

// 定义条件变量, 控制消费者线程
pthread_cond_t cond;
// 互斥锁变量
pthread_mutex_t mutex;
// 指向头结点的指针
struct Node * head = NULL;

// 生产者的回调函数
void* producer(void* arg)
{
    // 一直生产
    while(1)
    {
        pthread_mutex_lock(&mutex);
        
        // 创建一个链表的新节点
        struct Node* newnode = (struct Node*)malloc(sizeof(struct Node));
        // 节点初始化
        newnode->number = rand() % 1000;
        // 节点的连接, 添加到链表的头部, 新节点就新的头结点
        newnode->next = head;
        // head指针前移
        head = newnode;
        printf("+++producer, number = %d, tid = %ld\n", newnode->number, pthread_self());
        pthread_mutex_unlock(&mutex);

        // 生产了任务, 通知消费者消费
        pthread_cond_broadcast(&cond);
        // 生产慢一点
        usleep(rand() % 5000);
    }
    return NULL;
}

// 消费者的回调函数
void* consumer(void* arg)
{
    while(1)
    {
        pthread_mutex_lock(&mutex);
        while(head == NULL) {
            pthread_cond_wait(&cond, &mutex);
        }
        
        // 取出链表的头结点, 将其删除
        struct Node* pnode = head;
        printf("--consumer: number: %d, tid = %ld\n", pnode->number, pthread_self());
        head  = pnode->next;
        free(pnode);
        pthread_mutex_unlock(&mutex);        
        usleep(rand() % 5000);
    }
    return NULL;
}

int main()
{
    // 初始化条件变量
    pthread_cond_init(&cond, NULL);
    pthread_mutex_init(&mutex, NULL);

    // 创建5个生产者, 5个消费者
    pthread_t ptid[5];
    pthread_t ctid[5];
    for(int i = 0; i < 5; ++i) {
        pthread_create(&ptid[i], NULL, producer, NULL);
    }
    for(int i = 0; i < 5; ++i) {
        pthread_create(&ctid[i], NULL, consumer, NULL);
    }

    // 释放资源
    for(int i = 0; i < 5; ++i) {
        // 阻塞等待子线程退出
        pthread_join(ptid[i], NULL);
        pthread_join(ctid[i], NULL);
    }

    // 销毁条件变量
    pthread_cond_destroy(&cond);
    pthread_mutex_destroy(&mutex);
    return 0;
}

注意消费者的条件变量那里不能使用 if(head == NULL)

  • 由于生产者调用的 pthread_cond_broadcast(&cond); 唤醒了所有线程，此时所有线程都竞争抢锁的控制权
  • 就会存在一个锁加锁成功，删除节点成功，另一个锁在这之后加锁成功，但是删除节点的时候删除了空节点，报段错误
  • 因为换成 if 后，当前线程被唤醒，重新获得锁后，就接着 pthread_cond_wait 之后执行语句，也就是 if 之后的语句了，就有 bug
  • 但是使用 while 就不同了，执行后面的语句，下一步就是判断循环条件成立与否了，就能避免段错误，而 if 执行体的下一步是跳出 if 语句

信号量

函数

1
2
3
4
5
6
7
8
#include <semaphore.h>
sem_t sem;
// 初始化信号量/信号灯
int sem_init(sem_t *sem, int pshared, unsigned int value);

// 资源释放, 线程销毁之后调用这个函数即可
// 参数 sem 就是 sem_init() 的第一个参数            
int sem_destroy(sem_t *sem);
  • sem：信号量变量地址
  • pshared：
    • 0：线程同步
    • 非 0：进程同步
  • value：初始化当前信号量拥有的资源数（>=0），如果资源数为 0，线程就会被阻塞了
1
2
3
// 参数 sem 就是 sem_init() 的第一个参数  
// 函数被调用sem中的资源就会被消耗1个, 资源数-1
int sem_wait(sem_t *sem);

当线程调用这个函数，并且 sem 中的资源数 >0，线程不会阻塞，线程会占用 sem 中的一个资源，因此 资源数-1，直到 sem 中的资源数减为 0 时，资源被耗尽，因此线程也就被阻塞了

1
2
3
4
// 参数 sem 就是 sem_init() 的第一个参数  
// 函数被调用sem中的资源就会被消耗1个, 资源数-1
// 资源数为 0 也不阻塞，也不拥有资源
int sem_trywait(sem_t *sem);
1
2
3
4
5
6
7
8
9
// 表示的时间是从1971.1.1到某个时间点的时间, 总长度使用秒/纳秒表示
struct timespec {
    time_t tv_sec;      /* Seconds */
    long   tv_nsec;     /* Nanoseconds [0 .. 999999999] */
};

// 调用该函数线程获取sem中的一个资源，当资源数为0时，线程阻塞，在阻塞abs_timeout对应的时长之后，解除阻塞。
// abs_timeout: 阻塞的时间长度, 单位是s, 是从1970.1.1开始计算的
int sem_timedwait(sem_t *sem, const struct timespec *abs_timeout);

  • 信号量计数大于 0：
    如果信号量的当前值大于 0，那么 sem_timedwait 会立即减小信号量的计数值，并成功返回 0，不会发生阻塞

  • 信号量计数等于 0：

    • 如果信号量的当前值为 0，线程会阻塞，等待其他线程通过 sem_post 函数增加信号量的计数值
    • 如果在 abs_timeout 指定的绝对时间内，信号量的计数值增加到大于 0，则函数解除阻塞，减小信号量的计数值，并返回 0
    • 如果在 abs_timeout 时间内信号量计数仍然为 0，函数将返回 -1 并设置 errno 为 ETIMEDOUT，表示超时
1
2
// 调用该函数给sem中的资源数+1
int sem_post(sem_t *sem);
1
2
3
// 查看信号量 sem 中的整形数的当前值, 这个值会被写入到sval指针对应的内存中
// sval是一个传出参数
int sem_getvalue(sem_t *sem, int *sval);

生产者消费者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <semaphore.h>
#include <pthread.h>

// 链表的节点
struct Node {
    int number;
    struct Node* next;
};

// 生产者线程信号量
sem_t psem;
// 消费者线程信号量
sem_t csem;

// 互斥锁变量
pthread_mutex_t mutex;
// 指向头结点的指针
struct Node * head = NULL;

// 生产者的回调函数
void* producer(void* arg)
{
    // 一直生产
    while(1)
    {
        // 生产者检测生产的资源的空位是否不为 0，不为 0 就占用一个空位
        sem_wait(&psem);
        pthread_mutex_lock(&mutex);
        
        // 创建一个链表的新节点
        struct Node* newnode = (struct Node*)malloc(sizeof(struct Node));
        newnode->number = rand() % 1000;
        newnode->next = head;
        head = newnode;
        
        printf("+++producer, number = %d, tid = %ld\n", newnode->number, pthread_self());
        pthread_mutex_unlock(&mutex);

        // 通知消费者消费
        sem_post(&csem);
        
        // 生产慢一点
        usleep(rand() % 5000);
    }
    return NULL;
}

// 消费者的回调函数
void* consumer(void* arg)
{
    while(1)
    {
        sem_wait(&csem);
        pthread_mutex_lock(&mutex);
        struct Node* pnode = head;
        printf("--consumer: number: %d, tid = %ld\n", pnode->number, pthread_self());
        head  = pnode->next;
        free(pnode);
        pthread_mutex_unlock(&mutex);
        
        // 通知生产者生成, 给生产者生产的资源加空位
        sem_post(&psem);
        usleep(rand() % 5000);
    }
    return NULL;
}

int main()
{
    // 初始化信号量
    sem_init(&psem, 0, 5);  // 生成者线程一共有 5 个可以生产资源的空位
    sem_init(&csem, 0, 0);  // 消费者线程一共有 0 个可以使用的资源
    // 初始化互斥锁
    pthread_mutex_init(&mutex, NULL);

    // 创建5个生产者, 5个消费者
    pthread_t ptid[5];
    pthread_t ctid[5];
    for(int i = 0; i < 5; ++i) {
        pthread_create(&ptid[i], NULL, producer, NULL);
    }
    for(int i = 0; i < 5; ++i) {
        pthread_create(&ctid[i], NULL, consumer, NULL);
    }

    // 释放资源
    for(int i = 0; i < 5; ++i) {
        pthread_join(ptid[i], NULL);
        pthread_join(ctid[i], NULL);
    }

    sem_destroy(&psem);
    sem_destroy(&csem);
    pthread_mutex_destroy(&mutex);
    return 0;
}
1
2
3
4
5
6
7
// 消费者
sem_wait(&csem);
pthread_mutex_lock(&mutex);

// 生产者
sem_wait(&csem);
pthread_mutex_lock(&mutex);

  • 这两行不能颠倒，会造成死锁局面

  • 颠倒后就是先加互斥锁，假设消费者抢到了锁的控制权加锁，此时判断是否有资源，假设此时没有，那么消费者就阻塞在这里，等待生产者生产后唤醒

  • 由于加了互斥锁，生产者无法进入临界区生产资源，阻塞等待锁的控制权

  • 这样就死锁的局面了

说明：参考 https://subingwen.cn/