线程同步
多线程访问共享资源的问题
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <pthread.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <ctype.h>
void tfn(int arg) {
srand(0);
while(1) {
printf("HELLO ");
sleep(rand() % 3);
printf("WORLD \n");
sleep(rand() % 3);
}
}
int main() {
pthread_t tid;
pthread_create(&tid, NULL, &tfn, NULL);
srand(0);
while(1) {
printf("hello ");
sleep(rand() % 3);
printf("world \n");
sleep(rand() % 3);
}
return 0;
}
err = pthread_create(&ntid, NULL, thr_fn, "new thread: ");
if (err != 0) {
fprintf(stderr, "can't create thread: %s\n", strerror(err));
exit(1);
}
由于pthread_create的错误码不保存在errno中，因此不能直接用perror(3)打印错误信 息，可以先用strerror(3)把错误码转换成错误信息再打印。
使用pthread_mutex_t实现线程间同步
每个线程在对资源操作前都尝试先加锁，成功加锁才能操作，操作结束解锁
同一时刻，只能有一个线程持有该锁。当 A 线程对某个全局变量加锁访问，B 在访问前尝试加锁，拿不到锁，B 阻塞。C 线程不去加锁，而直接访问该全局变量，依然能够访问，但会出现数据混乱。
互斥锁
int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutexattr_t *attr);
可以直接 使用宏进行初始化。pthead_mutex_t muetx = PTHREAD_MUTEX_INITIALIZER;
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); 解锁
以上 5 个函数的返回值都是：成功返回 0， 失败返回错误号
主线程和子线程同时打印hell 和world字样，使用lock确保这两个打印是原子性的
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <pthread.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <ctype.h>
pthread_mutex_t mutex;
void tfn(int arg) {
srand(0);
while(1) {
pthread_mutex_lock(&mutex);
printf("HELLO ");
sleep(rand() % 3);
printf("WORLD \n");
pthread_mutex_unlock(&mutex);
sleep(rand() % 3);
}
}
int main() {
pthread_t tid;
pthread_mutex_init(&mutex, NULL);
pthread_create(&tid, NULL, &tfn, NULL);
srand(0);
while(1) {
pthread_mutex_lock(&mutex);
printf("hello ");
sleep(rand() % 3);
printf("world \n");
pthread_mutex_unlock(&mutex);
sleep(rand() % 3);
}
pthread_mutex_destroy(&mutex);
return 0;
}
读写锁
其特性为：写独占，读共享，写锁优先级高
1.读写锁是“写模式加锁”时， 解锁前，所有对该锁加锁的线程都会被阻塞。
2.读写锁是“读模式加锁”时， 如果线程以读模式对其加锁会成功；如果线程以写模式加锁会阻塞。
3.读写锁是“读模式加锁”时， 既有试图以写模式加锁的线程，也有试图以读模式加锁的线程。那么读写锁会阻塞随后的读模式锁请求。优先满足写模式锁。读锁、写锁并行阻塞，写锁优先级高
int pthread_rwlock_init(pthread_rwlock_t *lock, const pthread_rwlockattr_t *attr);
pthread_rwlock_t 读写锁变量，传出参数，attr读写锁相关属性，默认设为NULL即可
int pthread_rwlock_destroy(pthread_rwlock_t *lock); 销毁一个读写锁
int pthread_rwlock_rdlock(pthread_rwlock_t *lock); 读锁
int pthread_rwlock_tryrdlock(pthread_rwlock_t *lock); 尝试加读锁
int pthread_rwlock_wrlock(pthread_rwlock_t *lock); 写锁
int pthread_rwlock_trywrlock(pthread_rwlock_t *lock); 尝试加写锁
int pthread_rwlock_unlock(pthread_rwlock_t *lock); 解锁
以上函数，成功返回0， 失败返回错误号
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <pthread.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <ctype.h>
pthread_rwlock_t lock;
int count;
void tfn_write(int arg) {
while (1)
{
usleep(2000);
pthread_rwlock_wrlock(&lock);
printf("write thread %d global = %d\n", arg, ++count);
pthread_rwlock_unlock(&lock);
usleep(3000);
}
}
void tfn_read(int arg) {
while (1)
{
usleep(2000);
pthread_rwlock_rdlock(&lock);
printf("read thread %d global = %d\n", arg, ++count);
pthread_rwlock_unlock(&lock);
usleep(3000);
}
}
int main() {
pthread_t tid[8];
pthread_rwlock_init(&lock, NULL);
for (size_t i = 0; i < 3; i++)
{
pthread_create(&tid[i], NULL, &tfn_write, (void *)i);
}
for (size_t i = 0; i < 5; i++)
{
pthread_create(&tid[i+3], NULL, &tfn_read, (void *)i);
}
for (size_t i = 0; i < 8; i++)
{
pthread_join(tid[i], NULL);
}
pthread_rwlock_destroy(&lock);
return 0;
}
条件变量
int pthread_cond_init(pthread_cond_t *cond, const pthread_condattr_t *attr);
cond 条件变量，传出参数， cond 默认传NULL, 可以用宏来初始化一个默认的条件变量PTHREAD_COND_INITIALIZER
int pthread_cond_destroy(pthread_cond_t *cond);
int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);
1. 阻塞等待条件变量 cond（参 1）满足
2. 释放已掌握的互斥锁（解锁互斥量）相当于 pthread_mutex_unlock(&mutex); 1.2.两步为一个原子操作。
3. 当被唤醒，pthread_cond_wait 函数返回时，解除阻塞并重新申请获取互斥pthread_mutex_lock(&mutex);
int pthread_cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex,
const struct timespec *abstime);
int pthread_cond_signal(pthread_cond_t *cond);
唤醒至少一个阻塞在条件变量上的线程
int pthread_cond_broadcast(pthread_cond_t *cond);
唤醒全部阻塞在条件变量上的线程
动态初始化
pthread_mutex_t mutex;
pthread_mutext_init(&mutex, NULL);
静态初始化
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER
pthread_cond_wait
int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);
阻塞等待一个条件变量
- 阻塞等待条件变量cond满足条件
- 释放已掌握的互斥锁，相当于pthread_mutex_unlock
- 当被唤醒，pthread_cond_wait函数返回时，解除阻塞，并重新申请获取互斥锁，pthread_mutex_lock
使用条件变量完成生产者消费者模型
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include<errno.h>
#include <pthread.h>
struct msg{
int num;
struct msg *next;
};
struct msg *head;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; //定义并且初始化互斥量
pthread_cond_t has_data = PTHREAD_COND_INITIALIZER; // 定义初始化一个条件变量
void* produser(void *arg) {
while (1)
{
struct msg *mp = malloc(sizeof(struct msg));
mp->num = rand() % 1000 + 1; // 1-1000的随机数
printf("--produce %d\n", mp->num);
pthread_mutex_lock(&mutex); // 加锁互斥量
mp->next = head; //写公共区域
head = mp;
pthread_mutex_unlock(&mutex); // 解锁互斥量
pthread_cond_signal(&has_data); // 唤醒阻塞在条件变量has_data上的线程。
sleep(rand() % 3);
}
return (void *)NULL;
}
void* consumer(void *arg) {
while (1)
{
struct msg *mp;
pthread_mutex_lock(&mutex);
while (head == NULL)
{
// 阻塞等待条件变量
pthread_cond_wait(&has_data, &mutex);
}
mp = head;
head = mp->next;
pthread_mutex_unlock(&mutex);
printf("----------consumer id:%lu %d\n",pthread_self(), mp->num);
free(mp);
sleep(rand() % 3);
}
return (void *)NULL;
}
void err_thread(int ret, char *str) {
if (ret !=0)
{
fprintf(stderr, "%s:%s\n", str, strerror(ret));
pthread_exit(NULL);
}
}
int main() {
pthread_t pid, cid;
int ret;
srand(time(NULL));
ret = pthread_create(&pid, NULL, produser, NULL); // 生产者
if (ret !=0)
{
err_thread(ret, "pthread_create produser");
}
ret = pthread_create(&cid, NULL, consumer, NULL); // 消费者
if (ret !=0) {
err_thread(ret, "pthread_create consumer");
}
ret = pthread_create(&cid, NULL, consumer, NULL); // 消费者
if (ret !=0) {
err_thread(ret, "pthread_create consumer");
}
ret = pthread_create(&cid, NULL, consumer, NULL); // 消费者
if (ret !=0) {
err_thread(ret, "pthread_create consumer");
}
pthread_join(pid, NULL);
pthread_join(cid, NULL);
return 0;
}
image.png
信号量
由于互斥锁的粒度比较大，如果我们希望在多个线程间对某个对象的部分数据进行共享，使用互斥锁是没有办法实现的，只能将这个数据对象锁住，这样虽然达到了多个线程操作共享数据时保证数据正确定的目的，无形中导致线程的并发性下降，线程从并行执行，变成串行执行，与直接使用单进程无异。
信号量，是相对这种的一种处理方式，既能保证同步，数据不混乱，又能提高线程并发。
主要应用函数
#include <semaphore.h>
int sem_init(sem_t *sem, int pshared, unsigned int value);
pshared 0 ：用于线程间同步
1 : 用于进程间同步
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);
以上几个函数都是成功返回0，失败返回-1，同时设置errno
信号量基本操作
sem_wait 1. 信号量>0 则信号量做减1操作
| 2. 信号量==0， 造成线程阻塞
|
|对应
sem_post 将信号量加1，同时唤醒阻塞在信号量上的线程。
因此信号量的初值，决定了占用信号量线程的个数。
信号量同步线程操作
创建三个线程a,b,c，使用信号量，让线程c先运行，然后在运行线程b，最后运行线程a
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include<errno.h>
#include <pthread.h>
#include <semaphore.h>
sem_t sem1;
sem_t sem2;
void * a_fun(void *arg) {
sem_wait(&sem1);
printf("thread a running\n");
return (void *)0;
}
void * b_fun(void *arg) {
sem_wait(&sem2);
printf("thread b running\n");
//释放线程a
sem_post(&sem1);
return (void *)0;
}
void * c_fun(void *arg) {
printf("thread c running\n");
// 释放线程b，对信号量sem2做加1操作，让阻塞的线程b继续运行
sem_post(&sem2);
return (void *)0;
}
void err_thread(int ret, char *str) {
if (ret !=0)
{
fprintf(stderr, "%s:%s\n", str, strerror(ret));
pthread_exit(NULL);
}
}
int main() {
pthread_t a_pid, b_pid,c_pid;
//线程信号量初始化,初始值为0
sem_init(&sem1, 0,0);
sem_init(&sem2, 0,0);
int ret;
srand(time(NULL));
ret = pthread_create(&a_pid, NULL, a_fun, NULL);
if (ret !=0)
{
err_thread(ret, "pthread_create produser");
}
ret = pthread_create(&b_pid, NULL, b_fun, NULL);
if (ret !=0)
{
err_thread(ret, "pthread_create produser");
}
ret = pthread_create(&c_pid, NULL, c_fun, NULL);
if (ret !=0)
{
err_thread(ret, "pthread_create produser");
}
pthread_join(a_pid, NULL);
pthread_join(b_pid, NULL);
pthread_join(c_pid, NULL);
//销毁线程信号量
sem_destroy(&sem1);
sem_destroy(&sem2);
return 0;
}
image.png
网友评论