Lazy Free特性

惰性删除或延迟释放(Lazy Free),指在删除KEY时，采用异步方式延迟释放EKY所使用的内存，将该操作交给单独的子线程BIO(backgroup I/O)进行处理，避免在同步方式删除KEY对Redis主线程的长期占用而影响系统可用性。

在删除超大KEY如单个EKY占用内存过多或单个KEY包含过多元素时，同步删除方式会导致Redis服务长期不可用，设置会引发自从主动故障切换。

在Redis 4.0版本开始提供惰性删除特性。

Lazy Free使用场景

主动使用惰性删除特性
被动使用惰性删除特性

主动惰性删除操作

UNLINK命令

使用UNLINK删除集合键时，会按照集合键的元素去估算释放该KEY的成本，
如果释放成本超过LAZYFREE_THRESHOLD，则会采用Lazy Free方式进行处理。

FLUSHALL/FLUSHDB

通过ASYNC选项来设置FLUSHALL操作或FLUSHDB操作是否采用Lazy Free方式处理。

被动使用惰性删除

被动使用惰性删除主要有下面四类场景,并通过四个参数进行控制：

## 在内存到达最大内存需要逐出数据时使用
## 建议关闭，避免内存未及时释放
lazyfree-lazy-eviction no

## 在KEY过期时使用
## 建议开启
lazyfree-lazy-expire no

## 隐式删除服务器数据时，如RENAME操作
## 建议开启
lazyfree-lazy-server-del no

## 在对从库进行全量数据同步时
## 建议关闭
slave-lazy-flush no

UNLINK命令涉及代码

void delCommand(client *c) {
    delGenericCommand(c,0);
}

void unlinkCommand(client *c) {
    delGenericCommand(c,1);
}

/* This command implements DEL and LAZYDEL. */
void delGenericCommand(client *c, int lazy) {
    int numdel = 0, j;

    for (j = 1; j < c->argc; j++) {
        expireIfNeeded(c->db,c->argv[j]);
        int deleted  = lazy ? dbAsyncDelete(c->db,c->argv[j]) :
                              dbSyncDelete(c->db,c->argv[j]);
        if (deleted) {
            signalModifiedKey(c->db,c->argv[j]);
            notifyKeyspaceEvent(NOTIFY_GENERIC,
                "del",c->argv[j],c->db->id);
            server.dirty++;
            numdel++;
        }
    }
    addReplyLongLong(c,numdel);
}

/* Delete a key, value, and associated expiration entry if any, from the DB.
 * If there are enough allocations to free the value object may be put into
 * a lazy free list instead of being freed synchronously. The lazy free list
 * will be reclaimed in a different bio.c thread. */
#define LAZYFREE_THRESHOLD 64
int dbAsyncDelete(redisDb *db, robj *key) {
    /* Deleting an entry from the expires dict will not free the sds of
     * the key, because it is shared with the main dictionary. */
    if (dictSize(db->expires) > 0) dictDelete(db->expires,key->ptr);

    /* If the value is composed of a few allocations, to free in a lazy way
     * is actually just slower... So under a certain limit we just free
     * the object synchronously. */
    /* 进行Unlink操作，但不进行FREE操作 */
    dictEntry *de = dictUnlink(db->dict,key->ptr);
    if (de) {
        robj *val = dictGetVal(de);
        /* 计算lazy free当前对象的成本 */
        size_t free_effort = lazyfreeGetFreeEffort(val);

        /* If releasing the object is too much work, do it in the background
         * by adding the object to the lazy free list.
         * Note that if the object is shared, to reclaim it now it is not
         * possible. This rarely happens, however sometimes the implementation
         * of parts of the Redis core may call incrRefCount() to protect
         * objects, and then call dbDelete(). In this case we'll fall
         * through and reach the dictFreeUnlinkedEntry() call, that will be
         * equivalent to just calling decrRefCount(). */
        /* 当lazy free成本超过64时，使用后台线程处理 */
        if (free_effort > LAZYFREE_THRESHOLD && val->refcount == 1) {
            /* 对需要lazy free的对象数量+1 */
            atomicIncr(lazyfree_objects,1);
            /* 使用BIO子线程后台处理 */
            bioCreateBackgroundJob(BIO_LAZY_FREE,val,NULL,NULL);
            /* 将对象设置为NULL */
            dictSetVal(db->dict,de,NULL);
        }
    }

    /* Release the key-val pair, or just the key if we set the val
     * field to NULL in order to lazy free it later. */
    if (de) {
        dictFreeUnlinkedEntry(db->dict,de);
        if (server.cluster_enabled) slotToKeyDel(key);
        return 1;
    } else {
        return 0;
    }
}

/* Process the job accordingly to its type. */
if (type == BIO_CLOSE_FILE) {
	close((long)job->arg1);
} else if (type == BIO_AOF_FSYNC) {
	redis_fsync((long)job->arg1);
} else if (type == BIO_LAZY_FREE) {
	/* What we free changes depending on what arguments are set:
	 * arg1 -> free the object at pointer.
	 * arg2 & arg3 -> free two dictionaries (a Redis DB).
	 * only arg3 -> free the skiplist. */
	if (job->arg1)
		lazyfreeFreeObjectFromBioThread(job->arg1);
	else if (job->arg2 && job->arg3)
		lazyfreeFreeDatabaseFromBioThread(job->arg2,job->arg3);
	else if (job->arg3)
		lazyfreeFreeSlotsMapFromBioThread(job->arg3);
} else {
	serverPanic("Wrong job type in bioProcessBackgroundJobs().");
}

/* Release objects from the lazyfree thread. It's just decrRefCount()
 * updating the count of objects to release. */
void lazyfreeFreeObjectFromBioThread(robj *o) {
    decrRefCount(o);
    atomicDecr(lazyfree_objects,1);
}

void decrRefCount(robj *o) {
    if (o->refcount == 1) {
        switch(o->type) {
        case OBJ_STRING: freeStringObject(o); break;
        case OBJ_LIST: freeListObject(o); break;
        case OBJ_SET: freeSetObject(o); break;
        case OBJ_ZSET: freeZsetObject(o); break;
        case OBJ_HASH: freeHashObject(o); break;
        case OBJ_MODULE: freeModuleObject(o); break;
        case OBJ_STREAM: freeStreamObject(o); break;
        default: serverPanic("Unknown object type"); break;
        }
        zfree(o);
    } else {
        if (o->refcount <= 0) serverPanic("decrRefCount against refcount <= 0");
        if (o->refcount != OBJ_SHARED_REFCOUNT) o->refcount--;
    }
}

/* Free list object. */
void freeListObject(robj *o) {
    if (o->encoding == OBJ_ENCODING_QUICKLIST) {
        quicklistRelease(o->ptr);
    } else {
        serverPanic("Unknown list encoding type");
    }
}

/* Free entire quicklist. */
void quicklistRelease(quicklist *quicklist) {
    unsigned long len;
    quicklistNode *current, *next;

    current = quicklist->head;
    len = quicklist->len;
    while (len--) {
        next = current->next;

        zfree(current->zl);
        quicklist->count -= current->count;

        zfree(current);

        quicklist->len--;
        current = next;
    }
    zfree(quicklist);
}

UNLINK命令学习

unlink命令和del命令都调用unlinkCommand函数，使用参数lazy来标识是否使用Lazy Free方式。
Lazy Free方式会调用dbAsyncDelete函数来处理，处理过程中会计算Lazy Free方式释放对象的成本，只有超过特定阈值，才会采用Lazy Free方式。
Lazy Free方式会调用bioCreateBackgroundJob函数来使用BIO线程后台异步释放对象。
BIO线程后台处理时会采用DEL相同的方式来释放对象，唯一区别是使用后台线程不会阻塞其他业务操作。
当Redis对象执行UNLINK操作后，对应的KEY会被立即删除，不会被后续命令访问到，对应的VALUE采用异步方式来清理。