多线程同步

多线程

程序中默认只有一个线程，pthread_create()函数调用后就有2个线程。

pthread_create()

#include <pthread.h>
#include <string.h>
#include <unistd.h>
#include <iostream>
using namespace std;
//线程函数
void * callback(void * arg) {cout <<"pthread arg:"<< *(int *)arg << endl;return NULL;
}
int main() {pthread_t tid;// 创建一个子线程int num = 10;//传入参数int ret = pthread_create(&tid, NULL, callback, (void *)&num);if(ret != 0) {//不等于0说明失败了char * errstr = strerror(ret);cout << errstr << endl;}//主进程运行cout << "child pthread:" << tid << endl;cout << "main pthread:" << pthread_self() << endl;for (int i = 0; i < 50; i++){cout << i << endl;}//主线程退出不会影响其他进程pthread_exit(NULL);return 0;
}

pthread_exit()是退出线程

pthread_join()

子线程的资源可以由任意线程通过：pthread_join()回收资源,一般由主线程负责回收。

#include <pthread.h>
#include <string.h>
#include <unistd.h>
#include <iostream>
using namespace std;
//线程函数
int val = 10;//全局变量
void *callback(void *arg)
{cout << "child thread:" << pthread_self() << endl;// 返回值是void*,作用相当returnpthread_exit((void *)&val);
}
int main() {pthread_t tid;// 创建一个子线程pthread_create(&tid, NULL, callback, NULL);//主进程运行for (int i = 0; i < 10; i++){cout << i << endl;}//主线程回收子线程资源//返回值是二级指针int *pth_ret;pthread_join(tid, (void **)&pth_ret);cout << "exit data:" << *pth_ret << endl;pthread_exit(NULL);return 0;
}

pthread_join()是一个阻塞函数

用全局变量才能保持值稳定

pthread_death()

分离一个线程，被分离的线程的资源会被系统自动释放

• 不可以多次分离线程
• 不能分离一个已经分离的线程

#include <pthread.h>
#include <string.h>
#include <unistd.h>
#include <iostream>
using namespace std;
//线程函数
void* callback(void *arg){pthread_exit(NULL);
}
int main() {pthread_t tid;// 创建一个子线程pthread_create(&tid, NULL, callback, NULL);// 输出线程号cout << "main:" << pthread_self() << ",child:" << tid << endl;//设置子线程分离pthread_detach(tid);//释放主线程pthread_exit(NULL);return 0;
}

pthread_cancel()

向线程发送一个取消的线程，说白了就是终止线程

不是立即终止，而是执行到某个取消点才会终止

• 取消点是系统规定好的一些系统调用
• 可以把取消点理解为从用户区到内核区的切换点

//线程函数
void* callback(void *arg){for (int i = 0; i < 6;i++){cout << "child pthread:" << i << endl;}return NULL;
}
int main() {pthread_t tid;// 创建一个子线程pthread_create(&tid, NULL, callback, NULL);// 输出线程号cout << "main:" << pthread_self() << ",child:" << tid << endl;/*取消线程*/pthread_cancel(tid);for (int i = 0; i < 6; i++){cout << "main pthread:" << i << endl;}// 释放主线程pthread_exit(NULL);return 0;
}

说明cancel的线程不是立即取消的

线程属性attr

//线程属性类型 pthread_attr_t
int pthread_attr_init(pthread_attr_t *attr);
int pthread_attr_destroy(pthread_attr_t *attr);
int pthread_attr_getdetachstate(const pthread_attr_t *attr, int *detachstate);
int pthread_attr_setdetachstate(pthread_attr_t *attr, int detachstate);

还有很多的线程属性相关函数

//线程函数
void* callback(void *arg){for (int i = 0; i < 6;i++){cout << "child pthread:" << i << endl;}return NULL;
}
int main() {//创建线程属性变量pthread_attr_t attr;//初始化线程属性pthread_attr_init(&attr);//设置线程属性pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);pthread_t tid;//第二个参数是自己设置好的线程属性pthread_create(&tid, &attr, callback, NULL);cout << "main:" << pthread_self() << ",child:" << tid << endl;//释放线程属性资源pthread_attr_destroy(&attr);pthread_exit(NULL);return 0;
}

获取栈的大小

int main() {//创建线程属性变量pthread_attr_t attr;//初始化线程属性pthread_attr_init(&attr);//设置线程属性pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);pthread_t tid;//第二个参数是自己设置好的线程属性pthread_create(&tid, &attr, callback, NULL);//获取栈的大小size_t size;pthread_attr_getstacksize(&attr,&size);cout << "pthread size:" << size << endl;// 释放线程属性资源pthread_attr_destroy(&attr);pthread_exit(NULL);return 0;
}

线程同步

多线程编程的重点

#include <pthread.h>
#include <string.h>
#include <unistd.h>
#include <iostream>
using namespace std;
/* 三个窗口一起卖100张票 */
//线程函数是卖票
int ticket = 100;//全局
void* sellTicket(void* arg){usleep(100);while (ticket > 0){cout << pthread_self() << "在卖第" << ticket << "张票" << endl;ticket--;}return NULL;
}
int main() {//创建3个子线程,主线程只进行线程资源回收pthread_t tid1, tid2, tid3;pthread_create(&tid1, NULL, sellTicket, NULL);pthread_create(&tid2, NULL, sellTicket, NULL);pthread_create(&tid3, NULL, sellTicket, NULL);//回收子线程资源pthread_join(tid1, NULL);pthread_join(tid2, NULL);pthread_join(tid3, NULL);//退出主线程pthread_exit(NULL);return 0;
}

线程的优势就是可以通过全局变量共享信息,不过共享就存在了竞争

多个线程不会同时修改同一变量；某一线程不会读取其他线程正在修改的变量

这个代码的问题在于:

同一张票可以被多个线程卖；线程执行时间问题导致买了负票

临界资源

临界区是指访问某一共享资源的代码片段，并且这段代码的执行应为原子操作，也就是同时访问同一共享资源的其他线程不应终端该片段的执行。

线程同步：

即当有一个线程在对内存进行操作时，其他线程都不可以对这个内存地址进行操作.直到该线程完成操作，其他线程才能对该内存地址进行操作，而其他线程则处于等待状态。

线程同步影响并发效率,但是为了数据安全是必须的

原子操作就是执行一段代码只能完整执行,不能被中断,防止并发执行带来的问题

互斥锁

mutex--------- mutual exclusion的缩写

• 确保某时刻仅有一个线程可以访问某项共享资源。
• 可以使用互斥量来保证对任意共享资源的原子访问。

互斥量有两种状态：

• 已锁定（locked）
• 未锁定（unlocked）

每一线程在访问同一资源时将采用如下协议：

• 针对共享资源锁定互斥量
• 访问共享资源
• 对互斥量解锁

只有一个线程能对共享资源加锁,其他线程加锁会被阻塞

---

//互斥量的类型 
pthread_mutex_t  mt;

//相关函数
//restrict--c语言修饰符
int pthread_mutex_init(pthread_mutex_t *restrict mutex, constpthread_mutexattr_t *restrict attr);
int pthread_mutex_destroy(pthread_mutex_t *mutex);
int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_trylock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);

代码使用:

#include <pthread.h>
#include <string.h>
#include <unistd.h>
#include <iostream>
using namespace std;
/* 三个窗口一起卖1000张票 */
int ticket = 1000;//全局
/*互斥量:全局变量*/
pthread_mutex_t mutex;
void *sellTicket(void *arg){while(1){/*加锁*/pthread_mutex_lock(&mutex);if (ticket > 0){usleep(200);cout << pthread_self() << "在卖第" << ticket << "张票" << endl;ticket--;}else{/*解锁*/pthread_mutex_unlock(&mutex);break;}//解锁pthread_mutex_unlock(&mutex);}return NULL;
}
int main() {//初始化互斥量pthread_mutex_init(&mutex, NULL);// 创建3个子线程,主线程只进行线程资源回收pthread_t tid1, tid2, tid3;pthread_create(&tid1, NULL, sellTicket, NULL);pthread_create(&tid2, NULL, sellTicket, NULL);pthread_create(&tid3, NULL, sellTicket, NULL);//回收子线程资源pthread_join(tid1, NULL);pthread_join(tid2, NULL);pthread_join(tid3, NULL);//退出主线程pthread_exit(NULL);//释放互斥量pthread_mutex_destroy(&mutex);return 0;
}

死锁

什么是死锁

两个线程都在等待对方释放锁，在没有外力的作用下，这些线程会一直相互等待，就没办法继续运行，这种情况就是发生了死锁。

线程B锁住了资源2但是等待线程A释放资源1

线程A锁住资源1同时又在等待线程B释放资源2

死锁发生的4个条件

• 互斥条件

  多个线程不能同时使用同一个资源

• 持有并等待条件

  线程 在等待资源 1的同时并不会释放自己已经持有的资源 2。

• 不可剥夺条件

  当线程已经持有了资源 ，在自己使用完之前不能被其他线程获取

• 环路等待条件

  在死锁发生的时候，两个线程获取资源的顺序构成了环形链

死锁的例子

#include <pthread.h>
#include <string.h>
#include <unistd.h>
#include <iostream>
using namespace std;
//初始化两个锁
pthread_mutex_t mutex1, mutex2;
//线程函数
void* workA(void* arg){pthread_mutex_lock(&mutex1);sleep(1);pthread_mutex_lock(&mutex2);cout << "workA,,,," << endl;pthread_mutex_unlock(&mutex2);pthread_mutex_unlock(&mutex1);return NULL;
}
void* workB(void* arg){pthread_mutex_lock(&mutex2);sleep(1);pthread_mutex_lock(&mutex1);cout << "workB,,,," << endl;pthread_mutex_unlock(&mutex1);pthread_mutex_unlock(&mutex2);return NULL;
}
int main(){//初始化锁pthread_mutex_init(&mutex1,NULL);pthread_mutex_init(&mutex2, NULL);//创建两个线程pthread_t tid1, tid2;pthread_create(&tid1, NULL,workA, NULL);pthread_create(&tid1, NULL,workB, NULL);//退出主线程pthread_exit(NULL);//释放两把锁pthread_mutex_destroy(&mutex1);pthread_mutex_destroy(&mutex2);return 0;
}

程序卡死

读写锁

实际上多个线程同时读访问共享资源并不会导致问题。

在对数据的读写操作中，更多的是读操作，写操作较少

读写锁的特点：

• 如果有其它线程读数据，则允许其它线程执行读操作，但不允许写操作

• 如果有其它线程写数据，则其它线程都不允许读、写操作

• 写是独占的，写的优先级高

  同样有读锁和写锁请求资源,优先满足写

//读写锁的类型 
pthread_rwlock_t
//函数
int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock, const pthread_rwlockattr_t *restrict attr);
int pthread_rwlock_destroy(pthread_rwlock_t *rwlock);
int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_tryrdlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_trywrlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_unlock(pthread_rwlock_t *rwlock);

读写锁也是一把锁,只是可以指定锁类型

#include <pthread.h>
#include <string.h>
#include <unistd.h>
#include <iostream>
using namespace std;
/*8个线程,3个写,5个读*/
//共享数据
int nums = 100;
// 初始化锁
pthread_rwlock_t rw;
//线程函数
void* writeNum(void* arg){while(1){pthread_rwlock_wrlock(&rw);nums++;cout << pthread_self() << "is write,nums=" << nums << endl;pthread_rwlock_unlock(&rw);usleep(100);}return NULL;
}
void* readNum(void* arg){while(1){pthread_rwlock_rdlock(&rw);cout << pthread_self() << "is read,nums=" << nums << endl;pthread_rwlock_unlock(&rw);usleep(100);}return NULL;
}
int main(){//初始化锁pthread_rwlock_init(&rw,NULL);// 创建线程pthread_t wtid[3],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_detach(wtid[i]);for (int i = 0; i < 5;i++)pthread_detach(rtid[i]);pthread_exit(NULL);//释放锁pthread_rwlock_destroy(&rw);return 0;
}

生产者消费者模型

什么是生产者-消费者模式

比如有两个进程A和B，它们共享一个固定大小的缓冲区，A进程产生数据放入缓冲区，B进程从缓冲区中取出数据进行计算，那么这里其实就是一个生产者和消费者的模式，A相当于生产者，B相当于消费者

该模式的对象:

• 生产者
• 消费者
• 缓冲区

为了达到生产者和消费者生产数据和消费数据之间的平衡，那么就需要一个缓冲区用来存储生产者生产的数据

生产者-消费者模式的特点

• 保证生产者不会在缓冲区满的时候继续向缓冲区放入数据，而消费者也不会在缓冲区空的时候，消耗数据
• 当缓冲区满的时候，生产者会进入休眠状态，当下次消费者开始消耗缓冲区的数据时，生产者才会被唤醒，开始往缓冲区中添加数据；
• 当缓冲区空的时候，消费者也会进入休眠状态，直到生产者往缓冲区中添加数据时才会被唤醒

#include <pthread.h>
#include <string.h>
#include <unistd.h>
#include <iostream>
using namespace std;
//创建一个互斥量
pthread_mutex_t mutex;
/*生产者消费者模型*/
struct node{int num;node *next;node(int val,node* next=nullptr):num(val),next(next){}
};
node *head = NULL;
void *product(void *)
{//不断创建新节点到链表中while(1){pthread_mutex_lock(&mutex);node *newNode = new node(rand() % 100, head);head = newNode;cout << pthread_self << " create node" << newNode->num << endl;pthread_mutex_unlock(&mutex);usleep(100);}return NULL;
}
void* custom(void*){//不断取出头节点while(1){pthread_mutex_lock(&mutex);node *tmp = head;if(head){head = head->next;cout << pthread_self() << " consume node" << tmp->num << endl;delete (tmp);pthread_mutex_unlock(&mutex);usleep(100);}else{pthread_mutex_unlock(&mutex);}}return NULL;
}
int main(){//创建互斥量pthread_mutex_init(&mutex,NULL);// 创建5个生产者和5个消费者pthread_t pTids[5], cTids[5];for (int i = 0; i < 5;i++){pthread_create(&pTids[i], NULL, product, NULL);pthread_create(&cTids[i], NULL, custom, NULL);}//设置分离for (int i = 0; i < 5;i++){pthread_detach(pTids[i]);pthread_detach(cTids[i]);}/*保证程序执行*/while(1){sleep(10);}//释放互斥量pthread_mutex_destroy(&mutex);// 退出主线程pthread_exit(NULL);return 0;
}

有可能发生死锁,没有解决消费者没数据要通知生产者生产数据

条件变量

condition variables

不是锁,只是阻塞和解除阻塞

• 分为一直等待和限时等待

  wait,timewait

• 可以唤醒1个或者多个线程

  signal一个,broadcast多个

//条件变量的类型 
//pthread_cond_t
int pthread_cond_init(pthread_cond_t *restrict cond, const pthread_condattr_t*restrict attr);
int pthread_cond_destroy(pthread_cond_t *cond);
int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex);
int pthread_cond_timedwait(pthread_cond_t *restrict cond, pthread_mutex_t*restrict mutex, const struct timespec *restrict abstime);
int pthread_cond_signal(pthread_cond_t *cond);
int pthread_cond_broadcast(pthread_cond_t *cond);

加了条件变量

#include <pthread.h>
#include <string.h>
#include <unistd.h>
#include <iostream>
using namespace std;
pthread_mutex_t mutex;
/*创建条件变量*/
pthread_cond_t cond;
struct node
{int num;node *next;node(int val,node* next=nullptr):num(val),next(next){}
};
node *head = NULL;
void *product(void *){while(1){pthread_mutex_lock(&mutex);node *newNode = new node(rand() % 100, head);head = newNode;cout << pthread_self << " create node" << newNode->num << endl;/*只要生产了,就通知消费者消费*/pthread_cond_signal(&cond);pthread_mutex_unlock(&mutex);usleep(100);}return NULL;
}
void* custom(void*){while(1){pthread_mutex_lock(&mutex);node *tmp = head;if(head){head = head->next;cout << pthread_self() << " consume node" << tmp->num << endl;delete (tmp);pthread_mutex_unlock(&mutex);usleep(100);}else{/*没有数据,等待*///阻塞会释放互斥锁pthread_cond_wait(&cond, &mutex);pthread_mutex_unlock(&mutex);//没数据也要释放,不然死锁}}return NULL;
}
int main(){pthread_mutex_init(&mutex,NULL);/*创建条件变量*/pthread_cond_init(&cond, NULL);pthread_t pTids[5], cTids[5];for (int i = 0; i < 5;i++){pthread_create(&pTids[i], NULL, product, NULL);pthread_create(&cTids[i], NULL, custom, NULL);}for (int i = 0; i < 5;i++){pthread_detach(pTids[i]);pthread_detach(cTids[i]);}while(1){sleep(10);}pthread_mutex_destroy(&mutex);/*释放条件变量*/pthread_cond_destroy(&cond);pthread_exit(NULL);return 0;
}

信号量

信号量是一个计数器，用于控制多个进程对共享资源的访问。

信号量是一种特殊的变量，它可以被增加或减少，但对其的关键访问被保证是原子操作。

//信号量的类型 
//sem_t
int sem_init(sem_t *sem, int pshared, unsigned int value);
int sem_destroy(sem_t *sem);
int sem_wait(sem_t *sem);
int sem_trywait(sem_t *sem);
int sem_timedwait(sem_t *sem, const struct timespec *abs_timeout);
int sem_post(sem_t *sem);
int sem_getvalue(sem_t *sem, int *sval);

• 初始化:第二个变量区分进程和线程(0是线程)
• wait阻塞,把信号量值-1
• post,把信号量值+1,唤醒wait

#include <pthread.h>
#include <string.h>
#include <unistd.h>
#include <iostream>
#include <semaphore.h>
using namespace std;
pthread_mutex_t mutex;
//创建2个信号量
sem_t psem, csem;
struct node
{int num;node *next;node(int val,node* next=nullptr):num(val),next(next){}
};
node *head = NULL;
void *product(void *){while(1){//生产者-1sem_wait(&psem);pthread_mutex_lock(&mutex);node *newNode = new node(rand() % 100, head);head = newNode;cout << pthread_self << " create node" << newNode->num << endl;pthread_mutex_unlock(&mutex);//消费者+1sem_post(&csem);usleep(100);}return NULL;
}
void* custom(void*){while(1){//消费者-1sem_wait(&csem);pthread_mutex_lock(&mutex);node *tmp = head;head = head->next;cout << pthread_self() << " consume node" << tmp->num << endl;delete (tmp);pthread_mutex_unlock(&mutex);//生产者+1sem_post(&psem);}return NULL;
}
int main(){pthread_mutex_init(&mutex,NULL);/*创建信号量*/sem_init(&psem, 0, 8);sem_init(&csem, 0, 0);pthread_t pTids[5], cTids[5];for (int i = 0; i < 5;i++){pthread_create(&pTids[i], NULL, product, NULL);pthread_create(&cTids[i], NULL, custom, NULL);}for (int i = 0; i < 5;i++){pthread_detach(pTids[i]);pthread_detach(cTids[i]);}while(1){sleep(10);}pthread_mutex_destroy(&mutex);/*释放信号量*/sem_destroy(&psem);sem_destroy(&csem);pthread_exit(NULL);return 0;
}