devbox
不正经
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Linux线程（二）—— 多线程（同步与互斥、线程安全）_linux之线程同步二

作者：不正经 | 2024-03-29 18:10:39

赞

踩

linux之线程同步二

前置概念

大家都能看到的资源：公共资源

  1. 互斥：任何一个时刻，都只允许一个执行流在进行共享资源的访问 (可通过加锁解决)。本质就是将临界资源独立使用。
  2. 原子性：要么不做，要么做完，只有两种确定状态的属性。

多线程中，如果有一个全局的变量，这个变量是被所有执行流共享的。

线程大部分资源都会直接或间接共享，而只要存在共享，就可能存在并发访问的问题。

任何一个时刻，都只允许一个执行流再进行流在进行访问的共享资源，叫做临界资源。

所以要对临界资源进行一定的保护。

临界资源是要通过代码访问的，凡是访问临界资源的代码，叫做临界区。

如果想让多个线程安全的访问临界资源，就可以采用加锁，也就是互斥访问。

互斥

任何一个时刻，都只允许一个执行流再进行共享资源的访问 (可通过加锁解决)。本质就是将临界资源独立使用。

下面一个小实验没有对临界资源做保护，存在并发访问的问题：

  1. //模拟抢票
  2.  
  3. #include <iostream>
  4. #include <string>
  5. #include <pthread.h>
  6. #include <unistd.h>
  7. using namespace std;
  8.  
  9. int tickets = 10000;    //临界资源
  10.  
  11. void* ThreadRoutine(void* args)
  12. {
  13.     string name = static_cast<const char*>(args);
  14.  
  15.     while(true)
  16.     {
  17.         if(tickets > 0)     //临界区
  18.         {
  19.             usleep(2000);   //模拟抢票所花费时间
  20.             cout << name << " get ticket:  " << tickets-- << endl;    //临界区，因为对数据进行++/--操作并不是原子的，所以可能存在并发访问问题。
  21.  
  22.         }
  23.         else
  24.         {
  25.             break;
  26.         }
  27.         usleep(1000);
  28.     }
  29.     
  30.     
  31.     return nullptr;
  32. }
  33.  
  34. int main()
  35. {
  36.     //创建4个进程
  37.     pthread_t tids[4];
  38.     int size = sizeof(tids) / sizeof(tids[0]);
  39.     for(int i = 0; i < size; i++)
  40.     {
  41.         char *name = new char[64];
  42.         snprintf(name, 64, "thread-%d", i + 1);
  43.         pthread_create(tids+i, nullptr, ThreadRoutine, name);
  44.  
  45.     }
  46.  
  47.     for(int i = 0; i < size; i++)
  48.     {
  49.         pthread_join(tids[i], nullptr);
  50.     }
  51.  
  52.     return 0;
  53. }

可以看见没有对临界区做保护，票被买成了负数。

为什么可能无法获得正确结果？

  • if 语句判断条件为真以后，代码可以并发的切换到其他线程
  • usleep 这个模拟漫长业务的过程，在这个漫长的业务过程中，可能有很多个线程会进入该代码段
  • --ticket 操作本身就不是一个原子操作

-- 操作并不是原子操作，而是对应三条汇编指令：

load ：将共享变量ticket从内存加载到寄存器中 update : 更新寄存器里面的值，执行-1操作 store ：将新值，从寄存器写回共享变量ticket的内存地址

要解决以上问题，需要做到三点：

  • 代码必须要有互斥行为：当代码进入临界区执行时，不允许其他线程进入该临界区。
  • 如果多个线程同时要求执行临界区的代码，并且临界区没有线程在执行，那么只能允许一个线程进入该临界区。
  • 如果线程不在临界区中执行，那么该线程不能阻止其他线程进入临界区。

要做到这三点，本质上就是需要一把锁。Linux上提供的这把锁叫互斥量。

互斥量

初始化

动态分配

  1. #include <pthread.h>
  2. int pthread_mutex_init(pthread_mutex_t *restrict mutex, 
  3. 											const pthread_mutexattr_t *restrict attr);
  4. //参数：
  5. //mutex：要初始化的条件变量
  6. //attr：NULL

静态分配

  1. pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
  2. //**静态**或**全局**的锁可以用这个方式初始化，之后也不需要手动销毁
  3. //局部锁不可以使用！

销毁

int pthread_mutex_destroy(pthread_mutex_t *mutex);

销毁互斥量需要注意：

  • 使用 PTHREAD_ MUTEX_ INITIALIZER 初始化的互斥量不需要销毁
  • 不要销毁一个已经加锁的互斥量
  • 已经销毁的互斥量，要确保后面不会有线程再尝试加锁

加锁

  1. int pthread_mutex_lock(pthread_mutex_t *mutex);
  2. //加锁失败会将当前执行流阻塞

调用 pthread_mutex_lock 时，可能会遇到以下情况:

  • 互斥量处于未锁状态，该函数会将互斥量锁定，同时返回成功
  • 发起函数调用时，其他线程已经锁定互斥量，或者存在其他线程同时申请互斥量，但没有竞争到互斥量，那么pthread_mutex_lock调用会陷入阻塞(执行流被挂起)，等待互斥量解锁。
  1. int pthread_mutex_trylock(pthread_mutex_t *mutex);
  2. //非阻塞版本

解锁

int pthread_mutex_unlock(pthread_mutex_t *mutex);

当上面的抢票小实验引入互斥锁之后，发现票数正常了，并且速度降低了很多，这是因为并发访问的问题不存在了。

  1. //模拟抢票
  2.  
  3. #include <iostream>
  4. #include <string>
  5. #include <pthread.h>
  6. #include <unistd.h>
  7. using namespace std;
  8.  
  9. int tickets = 10000;    //临界资源，需要枷锁保证安全
  10. pthread_mutex_t mutex;  //定义一个互斥锁
  11.  
  12. void* ThreadRoutine(void* args)
  13. {
  14.     string name = static_cast<const char*>(args);
  15.  
  16.     while(true)
  17.     {
  18.         pthread_mutex_lock(&mutex);     //进入临界区前先加锁，所有线程都要遵守这个规则
  19.         if(tickets > 0)     //临界区
  20.         {
  21.             usleep(2000);   //模拟抢票所花费时间
  22.             cout << name << " get ticket:  " << tickets-- << endl;    //临界区
  23.  
  24.             pthread_mutex_unlock(&mutex);   //退出临界区解锁
  25.         }
  26.         else
  27.         {
  28.             pthread_mutex_unlock(&mutex);   //退出临界区解锁
  29.             break;
  30.         }
  31.  
  32.         usleep(1000);   //充当抢票后做的后续动作
  33.     }
  34.     
  35.     
  36.     return nullptr;
  37. }
  38.  
  39. int main()
  40. {
  41.     pthread_mutex_init(&mutex, nullptr);    //初始化互斥锁
  42.  
  43.     //创建4个进程
  44.     pthread_t tids[4];
  45.     int size = sizeof(tids) / sizeof(tids[0]);
  46.     for(int i = 0; i < size; i++)
  47.     {
  48.         char *name = new char[64];
  49.         snprintf(name, 64, "thread-%d", i + 1);
  50.         pthread_create(tids+i, nullptr, ThreadRoutine, name);
  51.  
  52.     }
  53.  
  54.     for(int i = 0; i < size; i++)
  55.     {
  56.         pthread_join(tids[i], nullptr);
  57.     }
  58.  
  59.     pthread_mutex_destroy(&mutex);     //销毁互斥锁
  60.     return 0;
  61. }

Untitled

细节知识

  • 访问同一个临界资源的线程，都必须使用同一把锁进行加锁保护，所有线程必须遵守，没有例外。

  • 每一个线程访问临界区之前都必须加锁，所以加锁的本质是给临界区加锁。加锁粒度尽量细一些，临界区越小越好，不然代码效率会降低。

  • 所有线程访问临界区时，需要先加锁，也就是说所有线程都必须要先看到同一把锁。所以锁本身就是公共资源，如何保证锁自己的安全？加锁和解锁本身就是原子的！所以是安全的。

  • 临界区可以是一条代码，也可以是一批代码。

    • 线程在加锁后运行临界区代码时会被切换吗？

      线程加锁后运行临界区代码时也有可能被切换，不要特殊化加锁解锁和临界区代码。

    • 切换后会有影响吗，会有访问安全的问题吗？

      不会，因为在申请了锁的线程不在的期间，任何其他线程都无法进入临界区，因为无法成功申请到锁

  • 这也正是体现互斥带来的串行化的表现，站在其他线程的角度，对其他线程有意义的状态就是：锁被当前线程申请(持有锁)、锁被当前线程释放(不持有锁)，原子性就体现在这里。

互斥锁实现原理

为了实现互斥锁 操作,大多数体系结构都提供了swap或exchange指令,该指令的作用是把寄存器和内存单元的数据相交换,由于只有一条指令,保证了原子性。

以下是pthread_mutex_lock的汇编伪代码

  1. lock:
  2. 	movb $0, %a1     # 把0值放进寄存器a1里
  3. 	xchgb %a1, mutex # 交换a1寄存器的内容和锁的值（无线程使用锁时，metux的值为1）—— 加锁，只有一条汇编语句，原子性
  4. 	if (%a1 > 0)
  5. 		return 0; # 得到锁，加锁成功
  6. 	else
  7. 		挂起等待;
  8. 	goto lock;

pthread_mutex_unlock的汇编伪代码

  1. unlock:
  2. 	movb $1 mutex  #把1赋给锁	
  3. 	唤醒等待的线程;
  4. 	return 0;

封装一个互斥锁

使用自己封装的thread类运行售票程序

  1. //Thread.h
  2. #pragma once
  3.  
  4. #include <iostream>
  5. #include <string>
  6. #include <pthread.h>
  7.  
  8. class Thread
  9. {
  10.     typedef void (*FUNC_t)(void *);
  11.     enum Thread_STATUS
  12.     {
  13.         NEW,
  14.         RUNNING,
  15.         EXITED
  16.     };
  17.  
  18. public:
  19.     Thread(int num, FUNC_t func, void* args)
  20.         :_args(args)
  21.         ,_status(NEW)
  22.         ,_func(func)
  23.     {
  24.         char name[64];
  25.         snprintf(name, 64, "Thread-%d", num);
  26.         _name = name;
  27.     }
  28.     ~Thread()
  29.     {}
  30.  
  31.     pthread_t threadid()
  32.     {
  33.         if(_status == RUNNING)
  34.             return _tid;
  35.         else
  36.             return 0;
  37.     }
  38.     std::string threadname()
  39.     {
  40.         return _name;
  41.     }
  42.     Thread_STATUS status()
  43.     {
  44.         return _status;
  45.     }
  46.  
  47.     static void* FuncHelper(void* args)
  48.     {
  49.         Thread* td = static_cast<Thread*>(args);
  50.         
  51.         //td->_func(td->_args);
  52.         (*td) ();
  53.     }
  54.     void operator() ()
  55.     {
  56.         _func(_args);
  57.     }
  58.  
  59.     void run()
  60.     {
  61.         int n = pthread_create(&_tid, nullptr, FuncHelper, this);
  62.         if(n != 0)
  63.         {
  64.             exit(1);
  65.         }
  66.         _status = RUNNING;
  67.     }
  68.  
  69.     void join()
  70.     {
  71.         int n = pthread_join(_tid, nullptr);
  72.         if(n != 0)
  73.         {
  74.             std::cerr << "thread join error" << std::endl;
  75.             return;
  76.         }
  77.         _status = EXITED;
  78.     }
  79.  
  80. private:
  81.     pthread_t _tid;
  82.     std::string _name;
  83.     void* _args;
  84.     Thread_STATUS _status;
  85.     FUNC_t _func;
  86. };
  1.  
  2. #include "Thread.h"
  3. #include "LockGuard.hpp"
  4. #include <iostream>
  5. #include <string>
  6. #include <unistd.h>
  7. using namespace std;
  8.  
  9. int tickets = 1000;
  10.  
  11. void ThreadRoutine(void* args)
  12. {
  13.     string name = static_cast<const char *>(args);
  14.  
  15.     while(true)
  16.   
  17.         if(tickets > 0)
  18.         {
  19.             usleep(2000);
  20.             cout << name << " get a ticket : " << tickets-- << endl;
  21.         }
  22.         else
  23.         {
  24.             break;
  25.         }
  26.  
  27.     }
  28. }
  29.  
  30. int main()
  31. {
  32.     Thread t1(1, ThreadRoutine, (void*)"thread-1");
  33.     Thread t2(2, ThreadRoutine, (void*)"thread-2");
  34.     Thread t3(3, ThreadRoutine, (void*)"thread-3");
  35.     Thread t4(4, ThreadRoutine, (void*)"thread-4");
  36.    
  37.     t1.run();
  38.     t2.run();
  39.     t3.run();
  40.     t4.run();
  41.  
  42.     t1.join();
  43.     t2.join();
  44.     t3.join();
  45.     t4.join();
  46.  
  47.     return 0;
  48. }

加上自己封装的互斥锁

  1. //LockGuard.h
  2. #pragma once
  3.  
  4. #include <iostream>
  5. #include <pthread.h>
  6.  
  7. class Mutex
  8. {
  9. public:
  10.     Mutex(pthread_mutex_t* mutex)
  11.         :_mutex(mutex)
  12.     {}
  13.     ~Mutex()
  14.     {}
  15.  
  16.     void lock()
  17.     {
  18.         pthread_mutex_lock(_mutex);
  19.     }
  20.     void unlock()
  21.     {
  22.         pthread_mutex_unlock(_mutex);
  23.     }
  24.  
  25. private:
  26.     pthread_mutex_t* _mutex;
  27. };
  28.  
  29. class lockGuard
  30. {
  31. public:
  32.     lockGuard(pthread_mutex_t mutex)
  33.         :mx(&mutex)
  34.     {
  35.         mx.lock();
  36.     }
  37.     ~lockGuard()
  38.     {
  39.         mx.unlock();    //离开区域调用析构自动解锁
  40.     }
  41. private:
  42.     Mutex mx;
  43. };
  1. //BuyTickets.cpp
  2. #include "Thread.h"
  3. #include "LockGuard.hpp"
  4. #include <iostream>
  5. #include <string>
  6. #include <unistd.h>
  7. using namespace std;
  8.  
  9. int tickets = 1000;
  10. pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;  //定义并初始化互斥锁
  11.  
  12. void ThreadRoutine(void* args)
  13. {
  14.     string name = static_cast<const char *>(args);
  15.  
  16.     while(true)
  17.     {
  18.         lockGuard lg(&mutex);     //加锁，离开循环区域调用析构自动解锁
  19.         if(tickets > 0)
  20.         {
  21.             usleep(2000);
  22.             cout << name << " get a ticket : " << tickets-- << endl;
  23.         }
  24.         else
  25.         {
  26.             break;
  27.         }
  28.  
  29.         usleep(1000);
  30.     }
  31. }
  32.  
  33. int main()
  34. {
  35.     Thread t1(1, ThreadRoutine, (void*)"thread-1");
  36.     Thread t2(2, ThreadRoutine, (void*)"thread-2");
  37.     Thread t3(3, ThreadRoutine, (void*)"thread-3");
  38.     Thread t4(4, ThreadRoutine, (void*)"thread-4");
  39.     cout << "name: " << t1.threadname() << ", tid: " << t1.threadid() << ", status: " << t1.status() << endl;
  40.     cout << "name: " << t2.threadname() << ", tid: " << t2.threadid() << ", status: " << t2.status() << endl;
  41.     cout << "name: " << t3.threadname() << ", tid: " << t3.threadid() << ", status: " << t3.status() << endl;
  42.     cout << "name: " << t4.threadname() << ", tid: " << t4.threadid() << ", status: " << t4.status() << endl;
  43.  
  44.     t1.run();
  45.     t2.run();
  46.     t3.run();
  47.     t4.run();
  48.  
  49.     t1.join();
  50.     t2.join();
  51.     t3.join();
  52.     t4.join();
  53.  
  54.     return 0;
  55. }
  56.

线程安全

线程安全和重入

线程安全：多个线程并发执行同一段代码时，不会出现不同的结果。常见对全局变量或者静态变量进行操作，并且没有锁保护的情况下，会出现该问题。

重入：同一个函数被不同的执行流调用，当前一个流程还没有执行完，就有其他的执行流再次进入，我们称之为重入。一个函数在重入的情况下，运行结果不会出现任何不同或者任何问题，则该函数被称为可重入函数，否则，是不可重入函数。

常见的线程不安全的情况

  • 不保护共享变量的函数
  • 函数状态随着被调用，状态发生变化的函数
  • 返回指向静态变量指针的函数
  • 调用线程不安全函数的函数

常见的线程安全的情况

  • 每个线程对全局变量或者静态变量只有读取的权限，而没有写入的权限，一般来说这些线程是安全的
  • 类或者接口对于线程来说都是原子操作
  • 多个线程之间的切换不会导致该接口的执行结果存在二义性

常见不可重入的情况

  • 调用了malloc/free函数，因为malloc函数是用全局链表来管理堆的
  • 调用了标准I/O库函数，标准I/O库的很多实现都以不可重入的方式使用全局数据结构
  • 可重入函数体内使用了静态的数据结构

常见可重入的情况

  • 不使用全局变量或静态变量
  • 不使用用malloc或者new开辟出的空间
  • 不调用不可重入函数
  • 不返回静态或全局数据，所有数据都由函数的调用者提供
  • 使用本地数据，或者通过制作全局数据的本地拷贝来保护全局数据

死锁

死锁是指在一组进程中的各个进程均占有不会释放的资源，但因互相申请被其他进程所占用且不会释放的资源而处于的一种永久等待状态。

死锁四个必要条件

  • 互斥条件：一个资源每次只能被一个执行流使用
  • 请求与保持条件：一个执行流因请求资源而阻塞时，对已获得的资源保持不放
  • 不剥夺条件:一个执行流已获得的资源，在末使用完之前，不能强行剥夺
  • 循环等待条件:若干执行流之间形成一种头尾相接的循环等待资源的关系

如何避免死锁

核心思想：破坏死锁的4个必要条件任意一个。

  1. 不加锁
  2. 主动释放锁
  3. 按照顺序申请锁
  4. 控制线程统一释放锁

测试锁能否被其他线程解锁

  1. #include <iostream>
  2. #include <string>
  3. #include <cstdio>
  4. #include <pthread.h>
  5. #include <unistd.h>
  6. using namespace std;
  7.  
  8. pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
  9.  
  10. void *threadRoutine(void *args)
  11. {
  12.     cout << "I am a new thread " << endl;
  13.  
  14.     pthread_mutex_lock(&mutex);
  15.     cout << "I got a mutex!" << endl;
  16.  
  17.     pthread_mutex_lock(&mutex); // 申请锁的问题，它会停下来
  18.     cout << "I alive again" << endl;
  19.  
  20.     return nullptr;
  21. }
  22.  
  23. int main()
  24. {
  25.     pthread_t tid;
  26.     pthread_create(&tid, nullptr, threadRoutine, nullptr);
  27.  
  28.     sleep(3);
  29.     cout << "main thread run begin" << endl;
  30.     pthread_mutex_unlock(&mutex);
  31.     cout << "main thread unlock..." << endl;
  32.  
  33.     sleep(3);
  34.     return 0;
  35. }

结论：其他的进程也可以解锁

线程同步

同步：在保证数据安全的前提下，让线程能够按照某种特定的顺序访问临界资源，从而有效避免饥饿问题，叫做同步

条件变量

当一个线程互斥地访问某个变量时，它可能发现在其它线程改变状态之前，它什么也做不了。

例如一个线程访问队列时，发现队列为空，它只能等待，直到其它线程将一个节点添加到队列中。这种情况就需要用到条件变量。

初始化条件变量

  1. #include <pthread.h>
  2. int pthread_cond_init(pthread_cond_t *restrict cond,const pthread_condattr_t *restrict
  3. attr);
  4. //参数：
  5. //cond：要初始化的条件变量
  6. //attr：NULL
  7. pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

销毁

  1. #include <pthread.h>
  2.  
  3. int pthread_cond_destroy(pthread_cond_t *cond);

等待条件满足

  1. #include <pthread.h>
  2.  
  3. int pthread_cond_wait(pthread_cond_t *restrict cond,
  4.               pthread_mutex_t *restrict mutex);
  5. //参数：
  6. //cond：要在这个条件变量上等待
  7. //mutex：互斥量,要被释放的互斥量

为什么等待条件满足需要传一个互斥锁呢？

因为条件变量是在加锁之后使用的，pthread_cond_wait在调用时，会自动释放锁。

线程在等待结束后，继续从pthread_cond_wait继续向后运行，并重新申请锁。

唤醒等待

  1. int pthread_cond_broadcast(pthread_cond_t *cond);   //唤醒等待队列中的全部进程
  2. int pthread_cond_signal(pthread_cond_t *cond);      //唤醒等待队列中的一个进程

简单案例

  1. #include <iostream>
  2. #include <pthread.h>
  3. #include <unistd.h>
  4.  
  5. using namespace std;
  6.  
  7. const int num = 4;
  8. pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
  9. pthread_cond_t cond = PTHREAD_COND_INITIALIZER;       //条件变量
  10.  
  11. void* active(void* args)
  12. {
  13.     const char* name = static_cast<const char *> (args);
  14.  
  15.     while(true)
  16.     {
  17.         pthread_mutex_lock(&mutex);
  18.         pthread_cond_wait(&cond, &mutex);     //等待条件满足
  19.         cout << name << " 活动中" << endl;
  20.         pthread_mutex_unlock(&mutex);
  21.     }
  22.  
  23.     return nullptr;
  24. }
  25.  
  26. int main()
  27. {
  28.     pthread_t tids[num];
  29.     for(int i = 0; i < num; i++)
  30.     {
  31.         char* name = new char[64];
  32.         snprintf(name, 64, "thread-%d", i + 1);
  33.         pthread_create(tids + 1, nullptr, active, name);
  34.     }
  35.  
  36.     sleep(3);
  37.  
  38.     while(true)
  39.     {
  40.         cout << "主线程唤醒线程...." << endl;
  41.         pthread_cond_signal(&cond);    //唤醒线程
  42.         sleep(1);
  43.     }
  44.     
  45.  
  46.     for(int i = 0; i < num; i++)
  47.     {
  48.         pthread_join(tids[i], nullptr);
  49.     }
  50.  
  51.     return 0;
  52. }

可以发现线程一个一个按顺序被唤醒运行

如果将

  1. while(true)
  2.     {
  3.         cout << "主线程唤醒线程...." << endl;
  4.         pthread_cond_signal(&cond);    //唤醒线程
  5.         sleep(1);
  6.     }

改成

  1. while(true)
  2.     {
  3.         cout << "主线程唤醒线程...." << endl;
  4.         pthread_cond_signal(&cond);    //唤醒线程
  5.         sleep(1);
  6.     }

主线程一次唤醒多个线程，并按顺序运行。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/不正经/article/detail/336603
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号