redis异步磁盘同步，关闭文件原理代码,C代码库,德仔网

代码语言
代码分类
【C】 redis异步磁盘同步，关闭文件原理
作者:share_good / 发布于2015/8/10/ 1136

	#include <stdio.h>
	#include <stdlib.h>
	#include <string.h>
	
	#define pint(a) printf("%d\n",(a))
	#define pstr(a) printf("%s\n",(a))
	
	#include "./util/bio.h"
	#include "./util/redislog.h"
	#include "./base/list.h"
	
	
	
	void flush_to_disk(void *arg){
	 flush((long long)arg);
	}
	void async_close_file(void *arg){
	 close((long long arg));
	}
	
	#define FLUHSH_TO_DISK 0
	#define ASYNC_CLOSE_FILE 1
	int main(int argc, char **argv){
	 redis_set_log_path("./");
	 bio_init();
	
	 bio_create_background_job(0,FLUHSH_TO_DISK,(void*)(long)i);
	 bio_create_background_job(1,ASYNC_CLOSE_FILE,(void*)(long)i);
	
	}

	

	

	

	bio.h

	#ifndef REDIS_BIO_H_
	#define REDIS_BIO_H_
	/*redis后台io操作线程个数*/
	#define REDIS_BIO_BUM_OPTS 2
	
	/*io type*/
	#define REDIS_BIO_CLOSE_FILE 0 /*close(fd)系统调用*/
	#define REDIS_BIO_AOF_FSYNC 1 /*fsync系统调用*/
	
	
	/**
	* 1.初始化每个线程的互斥量，和条件变量
	* 2.设置线程的statck大小保证每个线程执行有足够的空间
	* 3.创建线程
	*/
	void bio_init();
	
	/*创建线程任务，并设置回调给定参数*/
	void bio_create_background_job(int type,void (*callback)(void*),void *arg);
	
	unsigned long long bio_pending_jobs_of_type(int type); /*获取每个线程的任务数*/
	
	void bio_kill_threads(); /*等待所有线程结束，然后杀死线程*/
	
	
	#endif

	

	

	bio.c

	

	#include <pthread.h>
	#include <stdio.h>
	#include <stdlib.h>
	#include <time.h>
	#include <errno.h>
	#include <signal.h>
	#include "bio.h"
	#include "../base/list.h"
	#include "../util/redislog.h"
	
	
	/*有多少中类型的任务就创建多少个线程*/
	static pthread_t bio_thread[REDIS_BIO_BUM_OPTS]; /*保存每个线程id*/
	static pthread_mutex_t bio_thread_mutex[REDIS_BIO_BUM_OPTS]; /*每个线程的互斥量*/
	static pthread_cond_t bio_thread_cond[REDIS_BIO_BUM_OPTS]; /*每个线程的条件变量*/
	static list * bio_thread_job[REDIS_BIO_BUM_OPTS]; /*每个线程的任务队列*/
	static unsigned long long bio_thread_job_nums[REDIS_BIO_BUM_OPTS]; /*每个线程任务队列中任务数*/
	
	static void* bio_background_process(void *arg);
	
	typedef struct bio_job{
	 void (*proc)(void*);
	 time_t create_time;
	 void *arg;
	}bio_job;
	
	void bio_create_background_job(int type,void (*callback)(void*),void *arg){
	 bio_job *job = (bio_job*)malloc(sizeof(bio_job));
	 job->proc = callback;
	 job->arg = arg;
	 job->create_time = time(NULL);
	 pthread_mutex_lock(&bio_thread_mutex[type]);
	 list_node_head_add(bio_thread_job[type],job);
	 bio_thread_job_nums[type]++;
	 pthread_cond_signal(&bio_thread_cond[type]);
	 pthread_mutex_unlock(&bio_thread_mutex[type]);
	}
	
	/*每个线程拥有独立的线程空间，默认大小为128k，为了保证线程有足够栈空间来执行函数，我们把线程栈大小设置为4M*/
	#define REDIS_THREAD_STACK_SIZE 4*1024*1024
	void bio_init(){
	 pthread_attr_t attr;
	 pthread_t thread;
	 size_t stack_size;
	
	 /*初始化互斥变量和条件变量*/
	 int i;
	 for(i=0;i<REDIS_BIO_BUM_OPTS;i++){
	 pthread_mutex_init(&bio_thread_mutex[i],NULL);
	 pthread_cond_init(&bio_thread_cond[i],NULL);
	 bio_thread_job_nums[i] =0;
	 bio_thread_job[i] = list_create();
	 }
	
	 /*设置线程栈的大小*/
	 pthread_attr_init(&attr);
	 pthread_attr_getstacksize(&attr,&stack_size);
	 if(stack_size<REDIS_THREAD_STACK_SIZE)
	 pthread_attr_setstacksize(&attr,stack_size);
	
	 /*创建线程*/
	 void *type;
	 for(i=0;i<REDIS_BIO_BUM_OPTS;i++){
	 type = (void *)(unsigned long)i;
	 if(pthread_create(&thread,&attr,bio_background_process,type)!=0){
	 redis_log(REDIS_WARNING,"%s,%d,%s",__FILE__,__LINE__,strerror(errno));
	 }
	 bio_thread[i] = thread;
	 }
	
	}
	
	static void* bio_background_process(void *arg){
	 bio_job *job;
	 sigset_t sigset;
	 unsigned long type = (unsigned long)arg;
	
	 /*设置当前线程的属性是可以被取消的,而且是异步的*/
	 pthread_setcancelstate(PTHREAD_CANCEL_ENABLE,NULL);
	 pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS,NULL);
	
	 pthread_mutex_lock(&bio_thread_mutex[type]);
	
	 /*保证线程的安全执行需要屏蔽一些信号*/
	 /*屏蔽中止信号*/
	 sigemptyset(&sigset);
	 sigaddset(&sigset,SIGALRM);
	 if(pthread_sigmask(SIG_BLOCK,&sigset,NULL)!=0){
	 redis_log(REDIS_WARNING,"%s,%d,%s",__FILE__,__LINE__,strerror(errno));
	 }
	
	 /*开始执行任务*/
	 while(1){
	 list_node *node;
	 if(list_length(bio_thread_job[type])==0){
	 /*开锁 触发 上锁*/
	 redis_log(REDIS_WARNING,"type=%d,signal waiting..",type);
	 pthread_cond_wait(&bio_thread_cond[type],&bio_thread_mutex[type]);
	 redis_log(REDIS_WARNING,"type=%d,signal coming",type);
	 continue;
	 }
	 node = list_head(bio_thread_job[type]);
	 job = (bio_job*)node->val;
	 pthread_mutex_unlock(&bio_thread_mutex[type]);
	
	 /*执行操作*/
	 job->proc(job->arg);
	 pthread_mutex_lock(&bio_thread_mutex[type]);
	 list_node_del(bio_thread_job[type],node);
	 bio_thread_job_nums[type] --;
	 }
	}
	
	/*等待所有线程结束，然后杀死线程*/
	void bio_kill_threads(){
	 int i;
	 for(i=0;i<REDIS_BIO_BUM_OPTS;i++){
	 if(pthread_cancel(bio_thread[i])==0){ /*向线程发送kill信号成功*/
	 if(pthread_join(bio_thread[i],NULL)!=0){
	 redis_log(REDIS_WARNING,"%s,%d, thread %d pthread_join failure %s",__FILE__,__LINE__,bio_thread[i],strerror(errno));
	 }else{
	 redis_log(REDIS_WARNING,"%s,%d, thread %d pthread_join success !",__FILE__,__LINE__,bio_thread[i]);
	 }
	 /*销毁队列任务队列*/
	 free(bio_thread_job[i]);
	 }
	 }
	}
	
	/*获取每个线程的任务数*/
	unsigned long long bio_pending_jobs_of_type(int type){
	 unsigned long long val;
	 pthread_mutex_lock(&bio_thread_mutex[type]);
	 val = bio_thread_job_nums[type];
	 pthread_mutex_unlock(&bio_thread_mutex[type]);
	 return val;
	}
试试其它关键字
　异步磁盘　　关闭文件　
同语言下
可能有用的
share_good贡献的其它代码(12)