线程
1. 为什么使用线程?
使用fork创建进程以执行新的任务，该方式的代价很高。多个进程间不会直接共享内存线程是进程的基本执行单元，一个进程的所有任务都在线程中执行，进程要想执行任务，必须得有线程，进程至少要有一条线程，程序启动会默认开启一条线程，这条线程被称为主线程或 UI 线程
2. 什么是线程
线程，是进程内部的一个控制序列。即使不使用线程，进程内部也有一个执行线程。
类比：创建一个进程，类似于“克隆”一个家庭,该“家庭”与原来的家庭完全相同但是新“家庭”和原来的家庭完全独立。 进程包含一个或多个线 程,类似与一个家庭，包含一个或多个家庭成员。 家庭内的各成员同时做各自的事情（父亲工作、母亲持家、小孩上学）而对于家庭外部的人来说，这个家庭同时在做多件事情,家庭内的每个成员，就是一个线程。各个家庭成员有自己的个人资源（线程有自己的局部变量）但是所有家庭成员都能共享这个家庭的资源:房子、汽车、家庭的公共资金。
（同一个进程内的各个线程，能够共享整个进程的全局变量，除了线程的局部变量外，其他资源都共享）
注意：单核处理器上，同一个时刻只能运行一个线程。但是对于用户而言，感觉如同同时执行了多个线程一样
（各线程在单核CPU上切换，在一段时间内，同时执行了多个线程）
3. 线程的优点、缺点
优点： 创建线程比创建进程，开销要小。
缺点： 1）多线程编程，需特别小心，很容易发生错误。
2）多线程调试很困难。
3）把一个任务划分为两部分,用两个线程在单处理器上运行时，不一定更快。
除非能确定这两个部分能同时执行、且运行在多处理器上。
4. 线程的应用场合
- 需要让用户感觉在同时做多件事情时，比如，处理文档的进程，一个线程处理用户编辑，一个线程同时统计用户的字数。
- 当一个应用程序，需要同时处理输入、计算、输出时,可开3个线程，分别处理输入、计算、输出。让用户感觉不到等待。
- 高并发编程。
5. 线程的使用
1）线程的创建: pthread_create
原型：int pthread_create (pthread_t *thread, pthread_attr_t *attr, void *(start_routine)(void), void arg);
参数：thread, 指向新线程的标识符。 通过该指针返回所创建线程的标识符。
attr, 用来设置新线程的属性。一般取默认属性，即该参数取NULL
start_routine, 该线程的处理函数该函数的返回类型和参数类型都是void
arg, 线程处理函数start_routine的参数
功能：创建一个新线程，同时指定该线程的属性、执行函数、执行函数的参数通过参数1返回该线程的标识符。
返回值：成功，返回0
失败，返回错误代码
注意：大部分pthread_开头的函数成功时返回0，失败时返回错误码（而不是-1）
注意：使用fork创建进程后，进程马上就启动，但是是和父进程同时执行fork后的代码。
使用pthread_create创建线程后，新线程马上就启动，即执行对应的线程处理函数。
2）线程的终止: pthread_exit
原型：void pthread_exit (void *retval);
功能：在线程函数内部调用该函数。终止该线程，并通过参数retval返回一个指针。该指针不能指向该线程的局部变量。
3）等待指定线程结束: pthread_join
原型：int pthread_join (pthread_t th, void ** thread_return);
功能：类似与进程中的waitpid等待指定的线程结束，并使参数指向该线程函数的返回值（用pthread_exit返回的值);
参数：th, 指定等待的线程;
thread_return, 指向该线程函数的返回值,线程函数的返回值类型为void*，故该参数的类型为void**;
4）使用线程程序的编译
(1) 编译时，定义宏_REENTRANT
即： gcc -D_REENTRANT （#define REENTRANT)
功能：告诉编译器，编译时需要可重入功能。即使得，在编译时，编译部分函数的可重入版本。
注意：在单线程程序中，整个程序都是顺序执行的，一个函数在同一时刻只能被一个函数调用，但在多线程中，由于并发性，一个函数可能同时被多个函数调用，此时这个函数就成了临界资源，很容易造成调用函数处理结果的相互影响，如果一个函数在多线程并发的环境中每次被调用产生的结果是不确定的，我们就说这个函数是"不可重入的"/"线程不安全"的。
(2) 编译时，指定线程库
即: gcc -lpthread
功能：使用系统默认的NPTL线程库, 即在默认路径中寻找库文件libpthread.so 默认路径为/usr/lib和/usr/local/lib
当系统默认使用的不是NPTL线程库时（系统较老，2003年以前）
指定：gcc -L/usr/lib/nptl -lpthread
补充： -L 指定库文件所在的目录;
-l 指定库文件的名称(-lpthread ,指库文件名为libpthread.so);
总结：一般使用如下形式即可
gcc -D_REENTRANT -lpthread mythread.c -o mythread
实例 main1.c:
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
int my_global;
void* my_thread_handle(void *arg)
{
int val;
val = *((int*)arg);
printf("new thread begin, arg=%d\n", val);
my_global += val;
sleep(3);
pthread_exit(&my_global);
// 不再执行
printf("new thread end\n");
}
int main(void)
{
pthread_t mythread;
int arg;
int ret;
void *thread_return;
arg = 100;
my_global = 1000;
printf("my_global=%d\n", my_global);
printf("ready create thread...\n");
ret = pthread_create(&mythread, 0, my_thread_handle, &arg);
if (ret != 0) {
printf("create thread failed!\n");
exit(1);
}
printf("wait thread finished...\n");
ret = pthread_join(mythread, &thread_return);
if (ret != 0) {
printf("pthread_join failed!\n");
exit(1);
}
printf("wait thread end, return value is %d\n", *((int*)thread_return));
printf("my_global=%d\n", my_global);
printf("create thread finished!\n");
}
5）线程的同步
1）线程的互斥 - 指某一资源同时只允许一个访问者对其进行访问，具有唯一性和排它性。但互斥无法限制访问者对资源的访问顺序，即访 问是无序的。
2）线程的同步 - 指在互斥的基础上（大多数情况），通过其它机制实现访问者对资源的有序访问。
-
问题:
同一个进程内的各个线程，共享该进程内的全局变量如果多个线程同时对某个全局变量进行访问时，就可能导致竞态?解决办法： 对临界区使用信号量、或互斥量。
-
信号量和互斥量的选择,对于同步和互斥，使用信号量或互斥量都可以实现。
使用时，选择更符合语义的手段：
如果要求最多只允许一个线程进入临界区，则使用互斥量;
如果要求多个线程之间的执行顺序满足某个约束，则使用信号量;
3. 信号量:
1）什么是信号量
此时所指的“信号量”是指用于同一个进程内多个线程之间的信号量。即POSIX信号量，而不是System V信号量（用于进程之间的同步)
用于线程的信号量的原理，与用于进程之间的信号量的原理相同。都有P操作、V操作。
信号量的表示：sem_t 类型
2) 信号量的初始化:sem_init
原型：int sem_init (sem_t *sem， int pshared, unsigned int value);
功能：对信号量进行初始化
参数：sem, 指向被初始化的信号量
pshared, 0：表示该信号量是该进程内使用的“局部信号量”， 不再被其它进程共享。非0：该信号量可被其他进程共享，Linux不支持这种信号量
value, 信号量的初值。>= 0
返回值：成功，返回0
失败， 返回错误码
3) 信号量的P操作:sem_wait
原型：int sem_wait (sem_t *sem);
返回值：成功，返回0
失败， 返回错误码
4) 信号量的V操作:sem_post
原型：int sem_post (sem_t *sem);
返回值：成功，返回0
失败， 返回错误码
5) 信号量的删除: sem_destroy
原型：int sem_destroy (sem_t *sem);
返回值：成功，返回0
失败， 返回错误码
- 实例:
主线程循环输入字符串，把字符串存放到一个全局缓存中。新线程从全局缓存中读取字符串，统计该字符串的长度。
直到用户输入end;
实例main2.c:
#include <pthread.h>
#include <semaphore.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#define BUFF_SIZE 80
char buff[BUFF_SIZE];
sem_t sem;
static void* str_thread_handle(void *arg)
{
while(1) {
//P(sem)
if (sem_wait(&sem) != 0) {
printf("sem_wait failed!\n");
exit(1);
}
printf("string is: %slen=%d\n", buff, strlen(buff));
if (strncmp(buff, "end", 3) == 0) {
break;
}
}
}
int main(void)
{
int ret;
pthread_t str_thread;
void *thread_return;
ret = sem_init(&sem, 0, 0);
if (ret != 0) {
printf("sem_init failed!\n");
exit(1);
}
ret = pthread_create(&str_thread, 0, str_thread_handle, 0);
if (ret != 0) {
printf("pthread_create failed!\n");
exit(1);
}
while (1) {
fgets(buff, sizeof(buff), stdin);
//V(sem)
if (sem_post(&sem) != 0) {
printf("sem_post failed!\n");
exit(1);
}
if (strncmp(buff, "end", 3) == 0) {
break;
}
}
ret = pthread_join(str_thread, &thread_return);
if (ret != 0) {
printf("pthread_join failed!\n");
exit(1);
}
ret = sem_destroy(&sem);
if (ret != 0) {
printf("sem_destroy failed!\n");
exit(1);
}
return 0;
}
- 练习
创建2个线程（共有主线程、线程1、线程2共3个线程）
主线程阻塞式等待用户输入字符串
主线程每接收到一个字符串之后， 线程1就马上对该字符串进行处理。
线程1的处理逻辑为：统计该字符串的个数，并记录当时的时间。
线程1把该字符串处理完后，线程2马上就把处理结果写入文件result.txt
直到用户输入exit.
multi_pthread.c:
待实现...
4. 互斥量
1）什么是互斥量效果上等同于初值为1的信号量;
互斥量的使用：类型为 pthread_mutex_t
2）互斥量的初始化:pthread_mutex_init
原型：int pthread_mutex_init(pthread_mutex_t *mutex, pthread_mutexattr_t *attr);
参数：mutex， 指向被初始化的互斥量
attr, 指向互斥量的属性一般取默认属性（当一个线程已获取互斥量后，该线程再次获取该信号量，将导致死锁!)
3) 互斥量的获取: pthread_mutex_lock
原型：int pthread_mutex_lock (pthread_mutex_t *mutex);
4）互斥量的释放:pthread_mutex_unlock
原型：int pthread_mutex_unlock (pthread_mutex_t *mutex);
5）互斥量的删除: pthread_mutex_destroy
原型: int pthread_mutex_destroy (pthread_mutex_t *mutex);
- 实例
最简单的互斥量使用 main3.c:
#include <pthread.h>
#include <semaphore.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#define BUFF_SIZE 80
int global_value = 1000;
pthread_mutex_t lock;
static void* str_thread_handle(void *arg)
{
int i = 0;
for (i=0; i<10; i++) {
//pthread_mutex_lock(&lock);
if (global_value > 0) {
// work
sleep(1);
printf("soled ticket(%d) to ChildStation(%d)\n",
global_value, i+1);
}
global_value--;
//pthread_mutex_unlock(&lock);
sleep(1);
}
}
int main(void)
{
int ret;
pthread_t str_thread;
void *thread_return;
int i;
ret = pthread_mutex_init(&lock, 0);
if (ret != 0) {
printf("pthread_mutex_init failed!\n");
exit(1);
}
ret = pthread_create(&str_thread, 0, str_thread_handle, 0);
if (ret != 0) {
printf("pthread_create failed!\n");
exit(1);
}
for (i=0; i<10; i++) {
//pthread_mutex_lock(&lock);
if (global_value > 0) {
// work
sleep(1);
printf("soled ticket(%d) to MainStation(%d)\n",
global_value, i+1);
}
global_value--;
//pthread_mutex_unlock(&lock);
sleep(1);
}
ret = pthread_join(str_thread, &thread_return);
if (ret != 0) {
printf("pthread_join failed!\n");
exit(1);
}
ret = pthread_mutex_destroy(&lock);
if (ret != 0) {
printf("pthread_mutex_destroy failed!\n");
exit(1);
}
return 0;
}
test.c:
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <semaphore.h>
void *thread_function(void *arg);
#define WORK_SIZE 1024
char work_area[WORK_SIZE];
int time_to_exit = 0;
int main()
{
int res;
pthread_t a_thread;
void *thread_result;
res = pthread_create(&a_thread, NULL, thread_function, NULL);
if (res != 0) {
perror("Thread creation failed");
exit(EXIT_FAILURE);
}
printf("Input some text. Enter 'end' to finish\n");
while(!time_to_exit) {
fgets(work_area, WORK_SIZE, stdin);
while(1) {
if (work_area[0] != '\0') {
sleep(1);
}
else {
break;
}
}
}
printf("\nWaiting for thread to finish...\n");
res = pthread_join(a_thread, &thread_result);
if (res != 0) {
perror("Thread join failed");
exit(EXIT_FAILURE);
}
printf("Thread joined\n");
exit(EXIT_SUCCESS);
}
void *thread_function(void *arg)
{
sleep(1);
while(strncmp("end", work_area, 3) != 0) {
printf("You input %d characters\n", strlen(work_area) -1);
work_area[0] = '\0';
sleep(1);
while (work_area[0] == '\0' ) {
sleep(1);
}
}
time_to_exit = 1;
work_area[0] = '\0';
pthread_exit(0);
}
(1) 分析程序的功能
(2) 分析程序存在的隐患
(3) 用互斥量解决该程序的隐患 (不能删除、修改代码，只能添加代码）
修改后为main5.c:
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <semaphore.h>
void *thread_function(void *arg);
pthread_mutex_t work_mutex; /* protects both work_area and time_to_exit */
#define WORK_SIZE 1024
char work_area[WORK_SIZE];
int time_to_exit = 0;
int main() {
int res;
pthread_t a_thread;
void *thread_result;
res = pthread_mutex_init(&work_mutex, NULL);
if (res != 0) {
perror("Mutex initialization failed");
exit(EXIT_FAILURE);
}
res = pthread_create(&a_thread, NULL, thread_function, NULL);
if (res != 0) {
perror("Thread creation failed");
exit(EXIT_FAILURE);
}
pthread_mutex_lock(&work_mutex);
printf("Input some text. Enter 'end' to finish\n");
while(!time_to_exit) {
fgets(work_area, WORK_SIZE, stdin);
pthread_mutex_unlock(&work_mutex);
while(1) {
pthread_mutex_lock(&work_mutex);
if (work_area[0] != '\0') {
pthread_mutex_unlock(&work_mutex);
sleep(1);
}
else {
break;
}
}
}
pthread_mutex_unlock(&work_mutex);
printf("\nWaiting for thread to finish...\n");
res = pthread_join(a_thread, &thread_result);
if (res != 0) {
perror("Thread join failed");
exit(EXIT_FAILURE);
}
printf("Thread joined\n");
pthread_mutex_destroy(&work_mutex);
exit(EXIT_SUCCESS);
}
void *thread_function(void *arg) {
sleep(1);
pthread_mutex_lock(&work_mutex);
while(strncmp("end", work_area, 3) != 0) {
printf("You input %d characters\n", strlen(work_area) -1);
work_area[0] = '\0';
pthread_mutex_unlock(&work_mutex);
sleep(1);
pthread_mutex_lock(&work_mutex);
while (work_area[0] == '\0' ) {
pthread_mutex_unlock(&work_mutex);
sleep(1);
pthread_mutex_lock(&work_mutex);
}
}
time_to_exit = 1;
work_area[0] = '\0';
pthread_mutex_unlock(&work_mutex);
pthread_exit(0);
}
6）线程条件变量
与互斥锁不同，条件变量是用来等待而不是用来上锁的。条件变量用来自动阻塞一个线程，直到某特殊情况发生为止。通常条件变量和互斥锁同时使用。条件变量使我们可以睡眠等待某种条件出现。条件变量是利用线程间共享的全局变量进行同步的一种机制，主要包括两个动作：一个线程等待"条件变量的条件成立"而挂起；另一个线程使"条件成立"（给出条件成立信号）。条件的检测是在互斥锁的保护下进行的。如果一个条件为假，一个线程自动阻塞，并释放等待状态改变的互斥锁。如果另一个线程改变了条件，它发信号给关联的条件变量，唤醒一个或多个等待它的线程，重新获得互斥锁，重新评价条件。如果两进程共享可读写的内存，条件变量可以被用来实现这两进程间的线程同步。
2. 条件变量初始化 pthread_cond_init
原型：int pthread_cond_init (pthread_cond_t *cond, const pthread_condattr_t *attr);
参数：cond, 条件变量指针
attr 条件变量高级属性
man 安装： apt-get install manpages-posix-dev
3. 唤醒一个等待线程 pthread_cond_signal 通知条件变量,唤醒一个等待者
原型： int pthread_cond_signal (pthread_cond_t *cond);
参数：cond, 条件变量指针
4.唤醒所有等待该条件变量的线程pthread_cond_broadcast 广播条件变量
原型： int pthread_cond_broadcast (pthread_cond_t *cond);
参数：cond, 条件变量指针
5.等待条件变量/超时被唤醒pthread_cond_timedwait 等待条件变量cond被唤醒,直到由一个信号或广播，或绝对时间abstime到 * 才唤醒该线程
原型： int pthread_cond_timedwait (pthread_cond_t *cond, pthread_mutex_t *mutex, const struct timespec *abstime);
参数：cond, 条件变量指针
pthread_mutex_t *mutex 互斥量
const struct timespec *abstime 等待被唤醒的绝对超时时间
6.等待条件变量被唤醒pthread_cond_wait 等待条件变量cond被唤醒(由一个信号或者广播)
原型： int pthread_cond_wait (pthread_cond_t *cond, pthread_mutex_t *mutex);
参数：cond, 条件变量指针
pthread_mutex_t *mutex 互斥量
常见错误码： [EINVAL] cond或mutex无效,
[EINVAL] 同时等待不同的互斥量
[EINVAL] 主调线程没有占有互斥量
7. 释放/销毁条件变量pthread_cond_destroy 待销毁的条件变量
原型： int pthread_cond_destroy (pthread_cond_t *cond);
参数：cond, 条件变量指针
8.实例main6.c:
#include <stdio.h>
#include <pthread.h>
pthread_mutex_t mutex;
pthread_cond_t cond;
void *thread1(void *arg)
{
while (1) {
printf("thread1 is running\n");
pthread_mutex_lock(&mutex);
pthread_cond_wait(&cond, &mutex);
printf("thread1 applied the condition\n");
pthread_mutex_unlock(&mutex);
sleep(4);
}
}
void *thread2(void *arg)
{
while (1) {
printf("thread2 is running\n");
pthread_mutex_lock(&mutex);
pthread_cond_wait(&cond, &mutex);
printf("thread2 applied the condition\n");
pthread_mutex_unlock(&mutex);
sleep(2);
}
}
int main()
{
pthread_t thid1, thid2;
printf("condition variable study!\n");
pthread_mutex_init(&mutex, NULL);
pthread_cond_init(&cond, NULL);
pthread_create(&thid1, NULL, (void *)thread1, NULL);
pthread_create(&thid2, NULL, (void *)thread2, NULL);
do {
pthread_cond_signal(&cond);
sleep(1);
} while (1);
return 0;
}
main7.c pthread_cond_wait 中释放和加锁机制理解:
#include <stdio.h>
#include <pthread.h>
pthread_mutex_t mutex;
pthread_cond_t cond;
void *thread1(void *arg)
{
while (1) {
printf("thread1 is running\n");
pthread_mutex_lock(&mutex);
printf("thread1 lock..\n");
pthread_cond_wait(&cond, &mutex);
printf("thread1 applied the condition\n");
printf("thread1 unlock..\n");
pthread_mutex_unlock(&mutex);
sleep(4);
}
}
void *thread2(void *arg)
{
while (1) {
printf("thread2 is running\n");
pthread_mutex_lock(&mutex);
printf("thread2 lock..\n");
pthread_cond_wait(&cond, &mutex);
printf("thread2 applied the condition\n");
printf("thread2 unlock..\n");
pthread_mutex_unlock(&mutex);
sleep(2);
}
}
int main()
{
pthread_t thid1, thid2;
printf("condition variable study!\n");
pthread_mutex_init(&mutex, NULL);
pthread_cond_init(&cond, NULL);
pthread_create(&thid1, NULL, (void *)thread1, NULL);
pthread_create(&thid2, NULL, (void *)thread2, NULL);
do {
sleep(10);
pthread_cond_signal(&cond);
} while (1);
return 0;
}
高并发线程池实现: Nginx 线程池剖析
1.传统的并发处理
2.引入线程池
3.线程池实现剖析
并发基本概念:
所谓并发编程是指在在同一台计算机上“同时”处理多个任务。并发是在同一实体上的多个事件。
多线程能解决的问题？
单个进程或线程同时只能处理一个任务，如果有很多请求需要同时处理怎么办？
解决方案： 运用多进程或多线程技术解决
缺 陷：
-
创建和销毁线程上花费的时间和消耗的系统资源，甚至可能要比花在处理实际的用户请求的时间和资源要多得多
-
活动的线程需要消耗系统资源，如果启动太多，会导致系统由于过度消耗内存或“切换过度”而导致系统资源不足
