FLUSHDB ASYNC
FLUSHALL AYSNC

//后台线程数组 大小是BIO_NUM_OPS 在redis的3.2.3版本中是常量2， 现在版本中是BIO_NUM_OPS=3
//#define BIO_NUM_OPS       3
//表示支持三种任务。对于每种任务，对应一个list用于存储任务
static pthread_t bio_threads[BIO_NUM_OPS]; 
//bio_mutex和bio_condvar用于控制并发
static pthread_mutex_t bio_mutex[BIO_NUM_OPS];
static pthread_cond_t bio_newjob_cond[BIO_NUM_OPS];
static pthread_cond_t bio_step_cond[BIO_NUM_OPS];
//存储任务
static list *bio_jobs[BIO_NUM_OPS];
//存储pending状态的任务数量
static unsigned long long bio_pending[BIO_NUM_OPS];

//初始化锁与条件变量
    for (j = 0; j < BIO_NUM_OPS; j++) {
        pthread_mutex_init(&bio_mutex[j],NULL);
        pthread_cond_init(&bio_newjob_cond[j],NULL);
        pthread_cond_init(&bio_step_cond[j],NULL);
        bio_jobs[j] = listCreate();
        bio_pending[j] = 0;
    }
    ......
    ......
    //初始化线程 生成制定BIO_NUM_OPS数量的线程 线程传入一个编号j，0表明关闭文件，1表明aof初始化，2就是删除大Key
    for (j = 0; j < BIO_NUM_OPS; j++) {
        void *arg = (void*)(unsigned long) j;
        if (pthread_create(&thread,&attr,bioProcessBackgroundJobs,arg) != 0) {
            serverLog(LL_WARNING,"Fatal: Can't initialize Background Jobs.");
            exit(1);
        }
        bio_threads[j] = thread;
    }

struct bio_job {
    //job任务结构
    int fd; /* Fd for file based background jobs */
    lazy_free_fn *free_fn; /* Function that will free the provided arguments */
    void *free_args[]; /* List of arguments to be passed to the free function */
};

//提交Aof任务到队列
void bioCreateFsyncJob(int fd) {
    //申请任务job空间
    struct bio_job *job = zmalloc(sizeof(*job));
    job->fd = fd;
    //提交任务到队列
    bioSubmitJob(BIO_AOF_FSYNC, job);
}
//提交Lazy free任务到队列
void bioCreateLazyFreeJob(lazy_free_fn free_fn, int arg_count, ...) {
    va_list valist;
    /* Allocate memory for the job structure and all required
     * arguments */
    struct bio_job *job = zmalloc(sizeof(*job) + sizeof(void *) * (arg_count));
    job->free_fn = free_fn;
    va_start(valist, arg_count);
    for (int i = 0; i < arg_count; i++) {
        job->free_args[i] = va_arg(valist, void *);
    }
    va_end(valist);
    bioSubmitJob(BIO_LAZY_FREE, job);
}
//提交关闭文件任务到队列
void bioCreateCloseJob(int fd) {
    struct bio_job *job = zmalloc(sizeof(*job));
    job->fd = fd;
    bioSubmitJob(BIO_CLOSE_FILE, job);
}
//提交任务
void bioSubmitJob(int type, struct bio_job *job) {
    pthread_mutex_lock(&bio_mutex[type]);
    listAddNodeTail(bio_jobs[type],job);
    bio_pending[type]++;
    pthread_cond_signal(&bio_newjob_cond[type]);
    pthread_mutex_unlock(&bio_mutex[type]);
}
//返回指定类型的peding状态的作业的数量
unsigned long long bioPendingJobsOfType(int type) {
    unsigned long long val;
    pthread_mutex_lock(&bio_mutex[type]);
    val = bio_pending[type];
    pthread_mutex_unlock(&bio_mutex[type]);
    return val;
}

void *bioProcessBackgroundJobs(void *arg) {
    struct bio_job *job;
    unsigned long type = (unsigned long) arg;
    //一堆校验 省略
    。。。。。。
    //加锁
    pthread_mutex_lock(&bio_mutex[type]);
    while(1) {
        listNode *ln;
        /* The loop always starts with the lock hold. */
        if (listLength(bio_jobs[type]) == 0) {
            //条件不成立,等待
            pthread_cond_wait(&bio_newjob_cond[type],&bio_mutex[type]);
            //被通知之后,中止阻塞,从新判断条件
            continue;
        }
        /* Pop the job from the queue. */
        ln = listFirst(bio_jobs[type]);
        job = ln->value;
        //取走值之后,解锁
        pthread_mutex_unlock(&bio_mutex[type]);
        //完成出队之后，根据类型调用close、redis_fsync或者free_fn函数。
        if (type == BIO_CLOSE_FILE) {
            close(job->fd);
        } else if (type == BIO_AOF_FSYNC) {
            redis_fsync(job->fd)
        } else if (type == BIO_LAZY_FREE) {
            job->free_fn(job->free_args);
        } else {
            serverPanic("Wrong job type in bioProcessBackgroundJobs().");
        }
        //释放任务空间
        zfree(job);
        // 加锁删除队列中这个任务
        pthread_mutex_lock(&bio_mutex[type]);
        listDelNode(bio_jobs[type],ln);
        bio_pending[type]--; //队列中pending的数量减一
        //如果bioWaitStepOfType()上有阻塞的线程，则解除阻塞。
        pthread_cond_broadcast(&bio_step_cond[type]);
    }
}
线程类型 三种 0 关闭文件 1 aof 2 lazy free
#define BIO_CLOSE_FILE    0 /* Deferred close(2) syscall. */
#define BIO_AOF_FSYNC     1 /* Deferred AOF fsync. */
#define BIO_LAZY_FREE     2 /* Deferred objects freeing. */
//代表有上面三种类型的线程
#define BIO_NUM_OPS       3

void flushAppendOnlyFile(int force) {
    int sync_in_progress = 0;
    ...
    if (server.aof_fsync == AOF_FSYNC_EVERYSEC)
       sync_in_progress = aofFsyncInProgress();
    ...
    /* Perform the fsync if needed. */
    if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
        ....
    } else if ((server.aof_fsync == AOF_FSYNC_EVERYSEC &&
                server.unixtime > server.aof_last_fsync)) {
        if (!sync_in_progress) {
            //后台执行fsync
            aof_background_fsync(server.aof_fd);
            server.aof_fsync_offset = server.aof_current_size;
        }
        server.aof_last_fsync = server.unixtime;
    }
    ...
}
//如果一个AOf fsync当前已经在Bio后台线程中执行，返回true
int aofFsyncInProgress(void) {
    return bioPendingJobsOfType(BIO_AOF_FSYNC) != 0;
}
//启动后台线程 执行aof的fsync操作
void aof_background_fsync(int fd) {
    //这里就是调用Bio的创建异步任务的核心函数 发起入队操作
    bioCreateFsyncJob(fd);
}

源码太多 影响到阅读了 这里只是把流程串起来哈
使用unlink删除大key 逻辑是：
void unlinkCommand(client *c) {
    delGenericCommand(c, 1);
}
delGenericCommand->dbAsyncDelete->freeObjAsync->bioCreateLazyFreeJob
看到了吧 最后调用的就是Bio系统中的bioCreateLazyFreeJob去创建任务并且入队列