Linux入门多线程(线程概念生产者消费者模型消息队列线程池)万字解说

Posted 世_生

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Linux入门多线程(线程概念生产者消费者模型消息队列线程池)万字解说相关的知识，希望对你有一定的参考价值。

目录

1️⃣线程概念

什么是线程

  • 线程（thread）是进程中的一条执行路线，也可以说成线程是“一个进程内部的控制序列”。

通过下面内容可以理解“线程（thread）是进程中的一条执行路线”：
在我们之前学的进程中，一个进程的创建，操作系统会给该进程创建一个进程控制块（PCB），还要拷贝父进程的进程地址空间。如果子进程对父进程的数据进行读取并写入，就会发生写时拷贝，体现了进程的独立性。如果我们想要让该子进程能够和父进程一起去执行某个任务，则需要让子进程task_struct去指向父进程的进程地址空间，自己不需要自己的进程地址空间，这样当子进程去对父进程的数据进行写入时，就不会发生写时拷贝了，也可以和父进程一起完成任务，想当于该父进程有两个执行流，而这样的子进程可以通过vfork函数来创建。

我们有可以得出

  • 一切进程至少都有一个执行线程
    （我们之前学的进程都是单线程进程）

如果多线程创建好了，进程中的多个线程都看见看到同一块资源，而进程对这块资源分配给线程来完成一个任务。

  • 所以说线程是在进程内部完成的，本质是在进程地址空间内运行的。
  • 在Linux系统中，在CPU眼中，看到的PCB都要比传统的进程更加轻量化
  • 透过进程虚拟地址空间，可以看到进程的大部分资源，将进程资源合理分配给每个执行流，就形成了线程执行流

线程的优点

  • 创建一个新线程的代价要比创建一个新进程小得多
  • 与进程之间的切换相比，线程之间的切换需要操作系统做的工作要少很多
  • 线程占用的资源要比进程少很多
  • 能充分利用多处理器的可并行数量
  • 在等待慢速I/O操作结束的同时，程序可执行其他的计算任务
  • 计算密集型应用，为了能在多处理器系统上运行，将计算分解到多个线程中实现
  • I/O密集型应用，为了提高性能，将I/O操作重叠。线程可以同时等待不同的I/O操作。
    计算密集型：执行流的大部分任务，主要以计算为主：加密解密，排序查找。
    IO密集型：执行流的大部分任务是以IO为主的：刷磁盘，访问数据库，访问网络。

线程的缺点

性能损失

  • 一个很少被外部事件阻塞的计算密集型线程往往无法与共它线程共享同一个处理器。如果计算密集型线程的数量比可用的处理器多，那么可能会有较大的性能损失，这里的性能损失指的是增加了额外的同步和调度开销，而可用的资源不变。

健壮性降低

  • 编写多线程需要更全面更深入的考虑，在一个多线程程序里，因时间分配上的细微偏差或者因共享了不该共享的变量而造成不良影响的可能性是很大的，换句话说线程之间是缺乏保护的。

缺乏访问控制

  • 进程是访问控制的基本粒度，在一个线程中调用某些OS函数会对整个进程造成影响。
    比如：在多线程进程中，其中一个线程进行了一次I/O调用，这导致从用户态切换到内核态，把该进程置于阻塞状态，并切换到另一个进程（对用户级线程）。

编程难度提高

  • 编写与调试一个多线程程序比单线程程序困难得多

线程异常

  • 合理的使用多线程，能提高CPU密集型程序的执行效率
  • 线程是进程的执行分支，线程出异常，就类似进程出异常，进而触发信号机制，终止进程，进程终止，该进程内的所有线程也就随即退出

线程异常

  • 合理的使用多线程，能提高CPU密集型程序的执行效率
  • 合理的使用多线程，能提高IO密集型程序的用户体验（如生活中我们一边写代码一边下载开发工具，就是多线程运行的一种表现）

Linux进程VS线程

进程是资源分配的基本实体。
线程是调度的基本单位。
线程共享一部分数据，但也拥有自己的一部分数据

  • 线程ID
  • 一组寄存器
  • 栈
  • error
  • 信号屏蔽字
  • 调度优先级

进程的多个线程共享 同一地址空间,因此Text Segment、Data Segment都是共享的,如果定义一个函数,在各线程中都可以调用,如果定义一个全局变量,在各线程中都可以访问到,除此之外,各线程还共享以下进程资源和环境:

  • 文件描述符表
  • 每种信号的处理方式（SIG_IGN、SIG_DFL或者自定义的信号处理函数）
  • 当前工作目录
  • 用户id和组id

进程和线程的关系如下图：

2️⃣线程控制

Linux中没有正真的线程，线程中的结构是模拟了进程的PCB，所以，Linux内核中没有正真意义上关于线程的系统调用，我们使用的使用要引用<pthread.h>的头文件。
在使用编译器编译的时候，要指明使用pthread库，选项-lpthread

创建线程

功能：创建一个线程
原型：int pthread_create(pthread_t *thread, const pthread_attr_t *attr,
                      void *(*start_routine) (void *), void *arg);
参数：theard：返回线程ID（输入型参数）
     attr：设置线程的属性，attr为NULL表示默认属性
     start_routine：函数地址，线程启动后执行的函数
     arg：传给线程启动函数的参数
返回值：成功返回0，失败返回错误码

错误检查：

传统的一些函数是，成功返回0，失败返回-1，并且对全局变量errno赋值以指示错误。
pthreads函数出错时不会设置全局变量errno（而大部分其他POSIX函数会这样做）。而是将错误代码通
过返回值返回
pthreads同样也提供了线程内的errno变量，以支持其它使用errno的代码。对于pthreads函数的错误，
建议通过返回值业判定，因为读取返回值要比读取线程内的errno变量的开销更小

#include<stdio.h>
#include<pthread.h>
#include<unistd.h>

void* Routine(void* buf)

  printf("%s\\n",(char*)buf);
  return NULL;


int main()

  //创建线程t1
  pthread_t t1;
  pthread_create(&t1,NULL,Routine,(void*)"establish succeed");
  //主线程，循环，防止进程退出
  while(1);
  return 0;

运行结果：

establish succeed

获取线程的id

功能：获取线程在用户层的id
原型：pthread_self(void);

在我们创建了一个线程的时候，通过ps ajx |head -1&&ps ajx|grep ./a.out |grep -v grep命令查看进程时，只能看到一个进程。并且这两个执行流的pid是一样的。

#include<stdio.h>
#include<pthread.h>
#include<unistd.h>
void* Routine(void* buf)

  while(1)
   printf("%s:---->pid:%d---\\n",(char*)buf,getpid());
   sleep(1);
  
  return NULL;

int main()

  pthread_t t1;
  pthread_create(&t1,NULL,Routine,(void*)"establish succeed");  
  while(1)
    printf("--->pid:%d<----\\n",getpid());
    sleep(1);
  
  return 0;


这说明这两个执行流是一个进程。
我们是通过ps -aL|head -1 &&ps -aL|grep a.out来查看线程。

但是，当我们通过pthread_self函数获取的线程id和LWP不同。LWP是给内核看到，而pthread_self函数获取的id是用户层的id，给用户看到。

LWP是轻量级进程，在Linux下进程是资源分配的基本单位，线程是cpu调度的基本单位，而线程使用进程PCB描述实现，并且同一个进程中的所有PCB共用同一个虚拟地址空间，因此相较于传统进程更加的轻量化。
那么用户层的id又是什么呢？进程地址空间的一块地址

#include<stdio.h>
#include<pthread.h>
#include<unistd.h>
void* Routine(void * asg)

  pthread_t ret=pthread_self();
  while(1)
    printf("----id:%lu-----\\n",ret);
    printf("----id:%p-----\\n",ret);
    sleep(1);
  
  return NULL;


int main()

  pthread_t t1;
  pthread_create(&t1,NULL,Routine,NULL);  
  while(1);
  return 0;


如图：

我们使用的pthread库是通过动态链接的，在进程地址空间的共享区中，其中创建线程中线程的结构也在其中（线程的一些属性），通过上图我们可以看到，该结构是在动态库中的，所以在我们调度线程的时候或者切换线程的时候不用区内核中，而是在库中来找到相关的函数来调度，这也就是为什么说线程是在进程地址空间中运行的。
而我们可以通过用户级的id找到这块空间，来调度这个线程。这就是使用pthread_self函数获得的id的作用。

补充一下内容：线程是有自己的寄存器的，当线程还没有执行完自己的任务然而时间片到了，那里该寄存器是来存放上下文数据的。线程是有自己的栈的，当一个线程在执行任务时产生了临时数据是放在这个栈中，不会干扰其他的进程。（自己的理解哈）

线程终止

在主线程中直接用return结束，是整个进程的结束。

如果需要终止某个线程而不是终止整个进程，可以有三种方法：

  1. 从线程函数return。

  2. 线程可以调用pthread_exit终止自己

  3. 一个线程可以调用pthread_cancel终止同一进程中的同一线程

     功能：线程终止
 原型：void pthread_exit(void *retval);
 参数：retval：retval不要指向一个局部变量。

需要注意,pthread_exit或者return返回的指针所指向的内存单元必须是全局的或者是用malloc分配的,不能在线程函
数的栈上分配,因为当其它线程得到这个返回指针时线程函数已经退出了。

功能：取消一个执行中的线程
原型：int pthread_cancel(pthread_t thread);
参数：thread：线程id
返回值：成功返回0；失败返回错误码

等待线程

进程中父进程需要等待子进程，防止子进程形成僵尸进程，造成内存泄漏。
那么，在线程中，主线程一样也要等待其他是线程。当线程退出后，如果主进程没有等待其他线程，那么主线程不知道其他线程是否完成了自己的任务，这导致线程的空间没有被释放，仍然在进程地址空间中，当创建新线程后，不会复用这块空间，这就会导致内存泄漏。

功能：等待线程结束
原型：int pthread_join(pthread_t thread, void **retval);
参数：thread：线程的id
     retval：:它指向一个指针，后者指向线程的返回值
返回值：成功返回0；失败返回错误码

调用该函数的线程将挂起等待,直到id为thread的线程终止。
thread线程以不同的方法终止,通过pthread_join得到的终止状态是不同的，总结如下：

  1. 如果thread线程通过return返回,value_ ptr所指向的单元里存放的是thread线程函数的返回值。
  2. 如果thread线程被别的线程调用pthread_ cancel异常终掉,value_ ptr所指向的单元里存放的是常数PTHREAD_ CANCELED。
  3. 如果thread线程是自己调用pthread_exit终止的,value_ptr所指向的单元存放的是传给pthread_exit的参数。
  4. 如果对thread线程的终止状态不感兴趣,可以传NULL给value_ ptr参数。
#include<stdio.h>
#include<stdlib.h>
#include<pthread.h>
#include<unistd.h>

pthread_t t1,t2,t3,t4;
//return退出，不管退出码
void* Routine1(void * asg)

  printf("%s....quit\\n",(char*)asg);
  return NULL;

//return退出，管退出码
void* Routine2(void *asg)


  printf("%s....quit\\n",(char*)asg);
  return (void*)1;

//调用pthread_exit来退出
void *Routine3(void* asg)

  printf("%s....quit\\n",(char*)asg);
  pthread_exit((void*)2);

//调用 pthread_cancel来取消自己，这个函数的用法一般不是来取消自己的，而是取消别的线程的。
void *Routine4(void* asg)

  printf("%s....quit\\n",(char*)asg);
  pthread_cancel(t4);
  return NULL;

int main()

  //创建线程
  pthread_create(&t1,NULL,Routine1,(void*)"thread 1");
  pthread_create(&t2,NULL,Routine2,(void*)"thread 2");
  pthread_create(&t3,NULL,Routine3,(void*)"thread 3");
  pthread_create(&t4,NULL,Routine4,(void*)"thread 4");
  
  void* ret1=NULL;
  void* ret2=NULL;
  void* ret3=NULL;
  void* ret4=NULL;
	
  //线程等待
  pthread_join(t1,&ret1); 
  pthread_join(t2,&ret2);
  pthread_join(t3,&ret3);
  pthread_join(t4,&ret4);
  
  //打印线程退出时的退出码
  printf("thread return, thread id %lu, return code:%d\\n", t1 , *(int*)&ret1);
  printf("thread return, thread id %lu, return code:%d\\n", t2 , *(int*)&ret2);
  printf("thread return, thread id %lu, return code:%d\\n", t3, *(int*)&ret3);
  printf("thread return, thread id %lu, return code:%d\\n", t4, *(int*)&ret4);
 
  return 0;

线程分离

创建线程，要对线程进行等待，否则无法释放资源，从而导致内存泄漏。如果不关心线程的符号值，那么等待就是一种负担，这个时候，我们可以告诉系统，当这个线程退出时，自动释放线程的资源。

功能：线程分离
原型：int pthread_detach(pthread_t thread);
参数：线程id
返回值：成功时返回0；出错时，它返回一个错误号。
可以是线程组内其他线程对目标线程进行分离，也可以线程自己分离

3️⃣线程互斥

进程线程间的互斥概念

在学习管道的时候，管道是自带同步与互斥的。而在线程中，当多个线程没有加锁的情况下同时访问临界资源时会发生混乱。在举例之前，先了解几个概念。

  • 临界资源：多个线程执行流共享的资源叫做临界资源
  • 临界区：每个线程内部访问临界资源的代码叫做临界区
  • 互斥：任何时刻，互斥保证有且只有一个执行流进入临界区，访问临界资源，通常对临界资源起保护作用
  • 原子性：不会被任何调度机制打断的操作，该操作只有两态，要么完成，要么未完
    成

互斥量

大部分情况，线程使用的数据都是局部变量，变量的地址空间在线程栈空间内，这种情况，变量归属单个线程，其他线程无法获得这种变量。但有时候，很多变量都需要在线程间共享，这样的变量称为共享变量，可以通过数据的共享，完成线程之间的交互。多个线程并发的操作共享变量，会带来一些问题。

我们可以通过一个买票的例子，来看这块问题。

#include<stdio.h>
#include<pthread.h>
#include<stdlib.h>
#include<string.h>
#include<unistd.h>


int ticket=2000;
void *STicket(void* asg)
 
  while(1)
     if(ticket>0)
       usleep(100);
       printf("%s sang ticket:%d \\n",(char*)asg,ticket--);
     
     else
       break;
     
  
  return NULL;


int main()

  pthread_t t[4];
  int i;
  for(i=0;i<4;i++)
   char* p=(char*)malloc(sizeof(char)*64);
   sprintf(p,"pthread t%d",i);
   pthread_create(&t[i],NULL,STicket,(void *)p);
  
   pthread_join(t[0],NULL);
   pthread_join(t[1],NULL);
   pthread_join(t[2],NULL);
   pthread_join(t[3],NULL);

  return 0;

我们在运行结果中可以看到，票的数量本不可能出现负数的，但是在结果中出现了，那么这就是一个问题。
多个线程并发的访问同一块临界资源，我们用t1，t2，t3，t4，来表示四个线程。一开始票的数量有1000张。

《出现问题1》当t1首先访问到票时，判断票还有剩余，于是拿走一张票，票还剩999张。但是这些线程是并发执行的，有可能多个线程同时拿到票，且通过对票进行减减操作，那么这个票是重复了。
《出现问题2》当t3拿到票的时候，刚准备对票进行减减，时间片就到了，线程退出，那么在t3这个线程内把读取到的票的数量保存起来，当t3这个线程有运行时，先恢复上下文数据，然后对山下文数据中保存票的数量进行减减，当t3这个线程完成了操作后，把剩余票的数量进行更新，那么在t3没有运行前，票已经抢完了，但是t3它不知道，然后又把票的数量进行更新了，票又回来了，这个时候又出错了。出现负数的情况就是这样。
在我们判断票是否有剩余的时候，和对票减减的时候，并不是具有原子性的，因为这个时候，其他线程也在进行抢票，可能拿到重复的票。我们可以通过汇编来验证是否具有原子性。

#include<stdio.h>

int main()

	int a=5;
	a--;
	return 0;

–操作并不是原子性，而是对应了三条汇编：

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

想要解决上面的问题，需要做到三点：

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

而以上的三点本质就是加一把锁，在Linux上提供的这把锁叫做互斥量。

先要理解这个锁。当多个线程同时要执行临界区的代码，那么谁先申请到这把锁，谁就执行，其他的线程就开始进行等待，等待这把锁被释放，然后申请这把锁。

互斥量的接口

初始化互斥量有两中方法：

  • 方法1，静态分配
    pthread_mutex_t mutex=PTHREAD_MUTEX_INITIALIZER
  • 方法2，动态分配

int pthread_mutex_init(pthread_mutex_t *restrict mutex, const pthread_mutexattr_t *restrict attr);
参数：mutex：要初始化的互斥量
attr：设置属性，一般设置NULL，用默认设置
返回值：成功返回0，错误返回错误号

功能：销毁互斥量
原型：int pthread_mutex_destroy(pthread_mutex_t *mutex);
参数：mutex：要销毁的互斥量
返回值：成功返回0，错误返回错误号

注意：

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

互斥量加锁和解锁

功能：加锁
原型：int pthread_mutex_lock(pthread_mutex_t *mutex);
参数：mutex：要加锁的互斥量
返回值：成功返回0，错误返回错误号

功能：解锁
原型：int pthread_mutex_unlock(pthread_mutex_t *mutex);
参数：mutex：要解锁的互斥量
返回值：成功返回0，错误返回错误号

调用pthread_mutex_lock会遇到的情况

  • 互斥量处于没锁的状态，该函数将互斥量锁定，同时返回成功。
  • 发起函数调用时，其他线程已经锁定互斥量，或者存在其他线程同时申请互斥量，但没有竞争到互斥量，那么pthread_ lock调用会陷入阻塞(执行流被挂起)，等待互斥量解锁。

现在我们对之前的买票系统进行改进

#include<stdio.h>
#include<pthread.h>
#include<stdlib.h>
#include<string.h>
#include<unistd.h>

int ticket=2000;
pthread_mutex_t lock;

void *STicket(void* asg)
 
  while(1)
  	 //在执行临界区的代码前，先申请锁（加锁）  
     pthread_mutex_lock(&lock);
     if(ticket>0)
       usleep(100);
       printf("%s sang ticket:%d \\n",(char*)asg,ticket--);
     
     else
       //当没有票的时候，也释放锁（解锁）
       pthread_mutex_unlock(&lock);
       break;
     
     //访问完了临界资源时，释放锁（解锁）
     pthread_mutex_unlock(&lock);
  
  return NULL;


int main()

  //动态的初始化锁
  pthread_mutex_init(&lock,NULL);
  pthread_t t[4];
  int i;
  for(i=0;i<4;i++)
   char* p=(char*)malloc(sizeof(char)*64);
   sprintf(p,"pthread t%d",i);
   pthread_create(&t[i],NULL,STicket,(void *)p);
   
   pthread_join(t[0],NULL);
   pthread_join(t[1],NULL);
   pthread_join(t[2],NULL);
   pthread_join(t[3],NULL);
   //最后销毁锁
   pthread_mutex_destroy(&lock);
  return 0;


《问题1》：一个线程拿到了锁，会不会被其他线程切换？
答：会被切换，当这个拿到锁的线程切换到了其他线程，其他线程依然没有锁，依然要等待，然而当拿到锁的线程又开始运行时，首先要先恢复上下文数据，这个线程依然是拿到锁的状态（这个线程是拿着锁被切走的），可以继续执行临界区的代码。

《问题2》：申请锁的过程是不是原子性的？
答：申请锁的原子性的，要么没有申请到锁，要么锁已经释放了，可以申请锁。

《问题3》：锁本身就是临界资源，那么谁来保护锁？
答：锁是来保护临界资源的，但是锁也是临界资源的呀。但是锁本身就具有原子性，申请锁的过程必须是原子性的。

互斥量的实现原理研究

通过上面的例子，大家已经意识到单纯的i++和++i都不是原子的，有可能会有数据一致性的问题。

为了实现互斥锁的操作，大多数体系结构都提供了swap或exchange指令，该指令的作用是把寄存器和内存单元的数据相互交换，由于只有一条指令，保证了原子性。即使是多处理器平台，访问内存的总线周期也是有

以上是关于Linux入门多线程(线程概念生产者消费者模型消息队列线程池)万字解说的主要内容，如果未能解决你的问题，请参考以下文章

Linux入门多线程(线程概念生产者消费者模型消息队列线程池)万字解说

linux：线程&&多线程 初见

Linux多线程_(线程同步,基于阻塞队列的生产者消费者模型)

LINUX多线程（生产者消费者模型，POXIS信号量）

Linux多线程——生产者消费者模型

Linux：详解多线程(线程同步信号量和生产者与消费者模型的实现)