懒惰删除的巨大牺牲

懒惰删除的最初实现不是异步线程

异步线程方案其实也相当复杂

> sadd src1 value1 value2 value3
(integer) 3
> sadd src2 value3 value4 value5
(integer) 3
> sunionstore dest src1 src2
(integer) 5
> smembers dest
1) "value2"
2) "value3"
3) "value1"
4) "value4"
5) "value5"

异步删除的实现

struct bio_job {
    time_t time;  // 时间字段暂时没有使用，应该是预留的
    void *arg1, *arg2, *arg3;
};

/* 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)
        // 释放一个普通对象，string/set/zset/hash 等等，用于普通对象的异步删除
        lazyfreeFreeObjectFromBioThread(job->arg1);
    else if (job->arg2 && job->arg3)
        // 释放全局 redisDb 对象的 dict 字典和 expires 字典，用于 flushdb
        lazyfreeFreeDatabaseFromBioThread(job->arg2,job->arg3);
    else if (job->arg3)
        // 释放 Cluster 的 slots_to_keys 对象，参见源码篇的「基数树」小节
        lazyfreeFreeSlotsMapFromBioThread(job->arg3);

void lazyfreeFreeObjectFromBioThread(robj *o) {
    decrRefCount(o); // 降低对象的引用计数，如果为零，就释放
    atomicDecr(lazyfree_objects,1); // lazyfree_objects 为待释放对象的数量，用于统计
}

// 减少引用计数
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;  // 释放 hash 对象，继续追踪
        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--; // 引用计数减 1
    }
}

// 释放 hash 对象
void freeHashObject(robj *o) {
    switch (o->encoding) {
    case OBJ_ENCODING_HT:
        // 释放字典，我们继续追踪
        dictRelease((dict*) o->ptr);
        break;
    case OBJ_ENCODING_ZIPLIST:
        // 如果是压缩列表可以直接释放
        // 因为压缩列表是一整块字节数组
        zfree(o->ptr);
        break;
    default:
        serverPanic("Unknown hash encoding type");
        break;
    }
}

// 释放字典，如果字典正在迁移中，ht[0] 和 ht[1] 分别存储旧字典和新字典
void dictRelease(dict *d)
{
    _dictClear(d,&d->ht[0],NULL); // 继续追踪
    _dictClear(d,&d->ht[1],NULL);
    zfree(d);
}

// 这里要释放 hashtable 了
// 需要遍历第一维数组，然后继续遍历第二维链表，双重循环
int _dictClear(dict *d, dictht *ht, void(callback)(void *)) {
    unsigned long i;

    /* Free all the elements */
    for (i = 0; i < ht->size && ht->used > 0; i++) {
        dictEntry *he, *nextHe;

        if (callback && (i & 65535) == 0) callback(d->privdata);

        if ((he = ht->table[i]) == NULL) continue;
        while(he) {
            nextHe = he->next;
            dictFreeKey(d, he); // 先释放 key
            dictFreeVal(d, he); // 再释放 value
            zfree(he); // 最后释放 entry
            ht->used--;
            he = nextHe;
        }
    }
    /* Free the table and the allocated cache structure */
    zfree(ht->table); // 可以回收第一维数组了
    /* Re-initialize the table */
    _dictReset(ht);
    return DICT_OK; /* never fails */
}

队列安全

void bioCreateBackgroundJob(int type, void *arg1, void *arg2, void *arg3) {
    struct bio_job *job = zmalloc(sizeof(*job));

    job->time = time(NULL);
    job->arg1 = arg1;
    job->arg2 = arg2;
    job->arg3 = arg3;
    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]); // 释放锁
}

// 异步线程执行逻辑
void *bioProcessBackgroundJobs(void *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;
        /* It is now possible to unlock the background system as we know have
         * a stand alone job structure to process.*/
        pthread_mutex_unlock(&bio_mutex[type]); // 释放锁

        // 这里是处理过程，为了省纸，就略去了
        ...
        
        // 释放任务对象
        zfree(job);

        ...
        
        // 再次加锁继续处理下一个元素
        pthread_mutex_lock(&bio_mutex[type]);
        // 因为任务已经处理完了，可以放心从链表中删除节点了
        listDelNode(bio_jobs[type],ln);
        bio_pending[type]--; // 计数减 1
    }

1. Redis 还有其它地方用到了对象共享机制么？

2. Java 的 ConcurrentLinkQueue 具体是如何实现的？