Linux:信号量，生产者消费者模型和日志实现

目录1.什么是信号量2.基于环形队列的生产者消费者模型3.日志的实现1.什么是信号量信号量的本质是一把计数器描述资源的数量。申请信号量的本质是用来预定资源的比如买电影票。因为信号量也是共享资源同时只能一个线程访问所以访问信号量直接设计成了原子操作。信号量的本质是一把计数器。申请信号本质就是预定资源PV操作是原子的多个线程可以同时访问临界资源的条件如果公共资源当作整体多个线程不访问临界资源的同一个位置1.申请信号量2.访问指定的一个位置3.释放信号量这样就可以多个进程并发地访问这份资源。只要申请进来之后就不用再判断有没有了因为通过信号量已经确定了肯定可以分配给你。信号量可以定义全局和局部的sem 就是要传入的信号量。value 就是有多少资源位置#include semaphore.h 信号量 int sem_init(sem_t *sem, int pshared, unsigned int value); 参数 pshared:0表示线程间共享非零表示进程间共享 value信号量初始值销毁信号量int sem_destroy(sem_t *sem);等待和发布信号量功能等待信号量会将信号量的值减1 int sem_wait(sem_t *sem); //P() 功能发布信号量表示资源使用完毕可以归还资源了。将信号量值加1。 int sem_post(sem_t *sem);//V()2.基于环形队列的生产者消费者模型上一篇生产者与消费者模型是基于queue的整体使用是安全的其空间可以动态分配这里基于固定大小的环形队列重写这个程序有两点需要注意生产者不能把消费者套一个圈消费者不能超过生产者生产者与消费者只有两个场景会指向同一个位置需要同步为空为满。其他情况他们根本就不会指向同一个位置可以同时执行不用加锁。空时只能让生产者生产满时只能让消费之消费局部需要维持互斥与同步。不同角色对资源的认识生产者需要空间 sem_space 消费者需要数据 sem_data 。假设有N个位置开始时 sem_space N sem_data 0生产者伪代码 P(sem_space) //生产 V(sem_data) 消费者伪代码: P(sem_data) //消费 V(sem_space)两者当发现没有资源时都会被阻塞。单生产单消费RingQueue代码#pragma once #include iostream #include vector #include semaphore.h const int defaultsize 5; templateclass T class RingQueue { void P(sem_t sem) { sem_wait(sem);//-- 信号量是临界资源这里操作是原子的 } void V(sem_t sem) { sem_post(sem);// } public: RingQueue(int size defaultsize):_ringqueue(size),_size(size) { //定义信号量 sem_init(_space_sem,0,_size);//有_size个空间 sem_init(_data_sem,0,0);//最开始没有数据只有空间 } void Push(const T in) { //生产 P(_space_sem);//申请到说明一定有空间 _ringqueue[_p_step] in; _p_step; _p_step % _size; V(_data_sem); } void Pop(T* out) { P(_data_sem);//申请到说明一定有数据 *out _ringqueue[_c_step]; _c_step; _c_step % _size; V(_space_sem); } ~RingQueue() { sem_destroy(_space_sem); sem_destroy(_data_sem); } private: std::vectorT _ringqueue; int _size; int _p_step;//生产者的脚步 int _c_step;//消费者的脚步 sem_t _space_sem;//生产者关心的信号量因为生产和消费有时可以并发进行所有要两个。 sem_t _data_sem;//消费者关心的信号量 };用到了两个信号量_space_sem_data_sem生产者申请等待 _space_sem消费者申请等待 _data_sem因为信号量是公共资源要互斥访问所以申请获得信号量这些操作设计是原子的释放也是原子的。获得信号量之后说明一定可以对资源进行操作。V的两个作用是“释放资源 若有等待者则唤醒它”所以单生产单消费会看到生产者生产后消费者会跟着执行的现象。与阻塞队列的不同阻塞队列是当作一个整体来使用的所以需要整体加锁这里是通过信号量局部性地访问。多生产多消费中这里生产者与消费者的互斥与同步(空满通过信号量)已经实现了但是生产者与生产者之间消费者与消费者之间的互斥并没有实现因为比如多个生产者得到信号量进入到队列内部但是内部的_p_step 是共享的所以我们需要加锁防止生产完没有 _p_step导致数据被覆盖。信号量通过“阻塞/唤醒”机制没有加锁强制了生产者和消费者在缓冲区空/满时进行“同步交接”。这种机制比加锁更轻量因为加锁通常是“主动忙等”或“上下文切换”而信号量直接管理了资源计数。当队列不为空满时如果只定义一把锁生产和消费就不可能一起跑了 所以这里定义两把锁。在环形队列中多生产者多消费者先申请信号量还是先申请锁都是可以的但是先申请信号量更快先申请锁再申请信号量当多个线程到来时他们都会在锁外等待有锁时才会再申请信号量一个锁一个信号量地来。多个线程先申请信号量等待锁时多个线程可以申请信号量之后就只需要等待锁然后访问资源就行。多生产多消费RingQueue:#pragma once #include iostream #include vector #include semaphore.h #include pthread.h const int defaultsize 5; templateclass T class RingQueue { void P(sem_t sem) { sem_wait(sem);//-- 信号量是临界资源这里操作是原子的 } void V(sem_t sem) { sem_post(sem);// } public: RingQueue(int size defaultsize):_ringqueue(size),_size(size) { //定义信号量 sem_init(_space_sem,0,_size);//有_size个空间 sem_init(_data_sem,0,0);//最开始没有数据只有空间 pthread_mutex_init(_p_mutex,nullptr); pthread_mutex_init(_c_mutex,nullptr); } void Push(const T in) { //生产 P(_space_sem);//申请到说明一定有空间 pthread_mutex_lock(_p_mutex);//写在里面效率高 _ringqueue[_p_step] in; _p_step; _p_step % _size; pthread_mutex_unlock(_p_mutex); V(_data_sem); } void Pop(T* out) { P(_data_sem);//申请到说明一定有数据 pthread_mutex_lock(_c_mutex); *out _ringqueue[_c_step]; _c_step; _c_step % _size; pthread_mutex_unlock(_c_mutex); V(_space_sem); } ~RingQueue() { sem_destroy(_space_sem); sem_destroy(_data_sem); pthread_mutex_destroy(_p_mutex); pthread_mutex_destroy(_c_mutex); } private: std::vectorT _ringqueue; int _size; int _p_step;//生产者的脚步 int _c_step;//消费者的脚步 sem_t _space_sem;//生产者关心的信号量因为生产和消费有时可以并发进行所有要两个。 sem_t _data_sem;//消费者关心的信号量 pthread_mutex_t _p_mutex;//生产者的锁 pthread_mutex_t _c_mutex;//消费者的锁 };3.日志的实现可变参数列表可以设计在函数参数中的但不能只有...前面要有一个合法参数与函数调用压栈有关。我们要使用 va_list args就是char*; 保存可变参数列表va_start(args前面的参数)原理是压栈时参数在一起排列。然后 va_arg(args类型) 就可以得到传入的该类型参数类型就是如 printf(%d,%s, , )printf 会在前面字符串提取va_end(args)指向nullptr清理参数。可变参数列表使用示例计算可变参数列表传入数的和第一个参数是数据的个数#include stdio.h #include stdarg.h int sum(int count, ...) { va_list args; int sum 0; va_start(args,count); for(int i 0;i count ;i) { int value va_arg(args,int); sum value; } va_end(args); return sum; } int main() { int result sum(3,1,3,5); printf(%d\n,result);//9 return 0; }关于日志实现时间通过 time 我们可以获取当前的时间戳time_t 类型也就是uint64_t。#include time.h struct tm* localtime(const time_t *timep);localtime 会返回一个结构体有些时间是从0开始的内容如下图所示一个简单的日志代码#include iostream #include fstream #include stdarg.h #include cstdarg #include unistd.h #include sys/types.h #include sys/stat.h #include sys/fcntl.h enum{ Debug 0, Info, Warning, Error, Fatal, }; //打印到哪里 enum { Screen 10, OneFile, ClassFile }; std::string LevelToString(int level) { switch(level) { case Debug: return Debug; case Info: return Info; case Warning: return Warning; case Error: return Error; case Fatal: return Fatal; default: return Unknow; } } const int defultstyle Screen; const std::string default_filename testlog.; const std::string logdir log; class Log { public: Log():style(defultstyle),filename(default_filename) { mkdir(logdir.c_str(),0755);//存放日志的目录 }; void Enable(int sty)//打印的方式 { style sty; } std::string TimeStampExLocalTime()//获取当前时间 { time_t currtime time(nullptr); struct tm* curr localtime(currtime); char time_buffer[128]; //格式化输入到 time_buffer snprintf(time_buffer, sizeof(time_buffer), %d-%d-%d %d:%d:%d,\ curr-tm_year1900,curr-tm_mon1,curr-tm_mday,curr-tm_hour,curr-tm_min,curr-tm_sec); return time_buffer; } void WriteLogToOneFile(const std::string logname,const std::string message) { //c方式 umask(0); int fd open(logname.c_str(),O_CREAT|O_WRONLY|O_APPEND,0666); if(fd 0) return; write(fd, message.c_str(),message.size()); close(fd); //c方式 // std::ofstream out(logname,std::ofstream::app); // if(!out.is_open()) // return; // out.write(message.c_str(),message.size()); // //outmessage; // out.close(); } void WriteLogToClassFile(const std::string levelstr,const std::stringmessage) { std::string logname logdir;//往指定路径下打印 logname /; logname filename; logname levelstr;//通过文件后缀区分文件 WriteLogToOneFile(logname,message); } void WriteLog(const std::string levelstr,const std::string message) { switch(style) { case Screen: std::cout message; break; case OneFile: WriteLogToClassFile(all,message); break; case ClassFile: WriteLogToClassFile(levelstr,message); break; default: break; } } //第一个参数表示严重程度 void LogMessage(int level,const char* format , ...) { //错误的信息 char rightbuffer[1024]; va_list args; va_start(args,format);//取的起始指针args读取后面具体参数列表 //不想用 va_arg(args)一个个看我们这里直接处理 vsnprintf(rightbuffer,sizeof(rightbuffer),format,args);//format是%s,%f,%f,args就是具体对应的值 va_end(args); //记录发生的时间等等 char leftbuffer[1024]; std::string levelstr LevelToString(level); std::string currtime TimeStampExLocalTime(); std::string idstr std::to_string(getpid()); snprintf(leftbuffer, sizeof(leftbuffer), [%s][%s][%s],\ levelstr.c_str(),currtime.c_str(),idstr.c_str()); std::string loginfo leftbuffer; loginfo rightbuffer; WriteLog(levelstr,loginfo); //printf(%s%s,leftbuffer,rightbuffer); } ~Log(){}; private: int style; std::string filename; };测试示例Main.cc也可以封装一下Log lg; class Conf//配置中心 { public: Conf(int style defultstyle) { lg.Enable(style); } ~Conf() {} }; Conf conf;本篇结束