对于海量数据处理业务，我们通常需要一个索引数据结构，用来帮助查询，快速判断数据记录是否存在，这种数据结构通常又叫过滤器(filter)。考虑这样一个场景，上网的时候需要在浏览器上输入URL，这时浏览器需要去判断这是否一个恶意的网站，它将对本地缓存的成千上万的URL索引进行过滤，如果不存在，就放行，如果（可能）存在，则向远程服务端发起验证请求，并回馈客户端给出警告。

索引的存储又分为有序和无序，前者使用关联式容器，比如B树，后者使用哈希算法。这两类算法各有优劣：比如，关联式容器时间复杂度稳定O(logN)，且支持范围查询；又比如哈希算法的查询、增删都比较快O(1)，但这是在理想状态下的情形，遇到碰撞严重的情况，哈希算法的时间复杂度会退化到O(n)。因此，选择一个好的哈希算法是很重要的。

时下一个非常流行的哈希索引结构就是bloom filter，它类似于bitmap这样的hashset，所以空间利用率很高。其独特的地方在于它使用多个哈希函数来避免哈希碰撞，如图所示（来源wikipedia），bit数组初始化为全0，插入x时，x被3个哈希函数分别映射到3个不同的bit位上并置1，查询x时，只有被这3个函数映射到的bit位全部是1才能说明x可能存在，但凡至少出现一个0表示x肯定不存在。

但是，bloom filter的这种位图模式带来两个问题：一个是误报（false positives），在查询时能提供“一定不存在”，但只能提供“可能存在”，因为存在其它元素被映射到部分相同bit位上，导致该位置1，那么一个不存在的元素可能会被误报成存在；另一个是漏报（false nagatives），同样道理，如果删除了某个元素，导致该映射bit位被置0，那么本来存在的元素会被漏报成不存在。由于后者问题严重得多，所以bloom filter必须确保“definitely no”从而容忍“probably yes”，不允许元素的删除。

关于元素删除的问题，一个改良方案是对bloom filter引入计数，但这样一来，原来每个bit空间就要扩张成一个计数值，空间效率上又降低了。

Cuckoo Hashing

为了解决这一问题，本文引入了一种新的哈希算法——cuckoo filter，它既可以确保该元素存在的必然性，又可以在不违背此前提下删除任意元素，仅仅比bitmap牺牲了微量空间效率。先说明一下，这个算法的思想来源是一篇CMU论文，笔者按照其思路用C语言做了一个简单实现（Github），附上对一段文本数据进行导入导出的正确性测试。

接下来我会结合自己的示例代码讲解哈希算法的实现。我们先来看看cuckoo hashing有什么特点，它的哈希函数是成对的（具体的实现可以根据需求设计），每一个元素都是两个，分别映射到两个位置，一个是记录的位置，另一个是备用位置。这个备用位置是处理碰撞时用的，这就要说到cuckoo这个名词的典故了，中文名叫布谷鸟，这种鸟有一种即狡猾又贪婪的习性，它不肯自己筑巢，而是把蛋下到别的鸟巢里，而且它的幼鸟又会比别的鸟早出生，布谷幼鸟天生有一种残忍的动作，幼鸟会拼命把未出生的其它鸟蛋挤出窝巢，今后以便独享“养父母”的食物。借助生物学上这一典故，cuckoo hashing处理碰撞的方法，就是把原来占用位置的这个元素踢走，不过被踢出去的元素还要比鸟蛋幸运，因为它还有一个备用位置可以安置，如果备用位置上还有人，再把它踢走，如此往复。直到被踢的次数达到一个上限，才确认哈希表已满，并执行rehash操作。如下图所示（图片来源）：

我们不禁要问发生哈希碰撞之前的空间利用率是多少呢？不幸地告诉你，一维数组的哈希表上跟其它哈希函数没什么区别，也就50%而已。但如果是二维的呢？

一个改进的哈希表如下图所示，每个桶（bucket）有4路槽位（slot）。当哈希函数映射到同一个bucket中，在其它三路slot未被填满之前，是不会有元素被踢的，这大大缓冲了碰撞的几率。笔者自己的简单实现上测过，采用二维哈希表（4路slot）大约80%的占用率（CMU论文数据据说达到90%以上，应该是扩大了slot关联数目所致）。

Cuckoo Filter设计与实现

cuckoo hashing的原理介绍完了，下面就来演示一下笔者自己实现的一个cuckoo filter应用，简单易用为主，不到500行C代码。应用场景是这样的：假设有一段文本数据，我们把它通过cuckoo filter导入到一个虚拟的flash中，再把它导出到另一个文本文件中。flash存储的单元页面是一个log_entry，里面包含了一对key/value，value就是文本数据，key就是这段大小的数据的SHA1值（照理说SHA1是可以通过数据源生成，没必要存储到flash，但这里主要为了测试而故意设计的，万一key和value之间没有推导关系呢）。

#define SECTOR_SIZE    (1 << 10)
#define DAT_LEN        (SECTOR_SIZE - 20)  /* minus sha1 size */

/* The log entries store key-value pairs on flash and the
 * size of each entry is assumed just one sector fit.
 */
struct log_entry {
        uint8_t sha1[20];
        uint8_t data[DAT_LEN];
};

顺便说明一下DAT_LEN设置，之前我们设计了一个虚拟flash（用malloc模拟出来），由于flash的单位是按页大小SECTOR_SIZE读写，这里假设每个log_entry正好一个页大小，当然可以根据实际情况调整。

以上是flash的存储结构，至于哈希表里的slot有三个成员tag，status和offset，分别是哈希值，状态值和在flash的偏移位置。其中status有三个枚举值：AVAILIBLE，OCCUPIED，DELETED，分别表示这个slot是空闲的，占用的还是被删除的。至于tag，按理说应该有两个哈希值，对应两个哈希函数，但其中一个已经对应bucket的位置上了，所以我们只要保存另一个备用bucket的位置就行了，这样万一被踢，只要用这个tag就可以找到它的另一个安身之所。

enum { AVAILIBLE, OCCUPIED, DELETED, };

/* The in-memory hash bucket cache is to filter keys (which is assumed SHA1) via
 * cuckoo hashing function and map keys to log entries stored on flash.
 */
struct hash_slot_cache {
        uint32_t tag : 30;  /* summary of key */
        uint32_t status : 2;  /* FSM */
        uint32_t offset;  /* offset on flash memory */
};

乍看之下size有点大是吗？没关系，你也可以根据情况调整数据类型大小，比如uint16_t，这里仅仅为了测试正确性。

至于哈希表以及bucket和slot的创建见初始化代码。buckets是一个二级指针，每个bucket指向4个slot大小的缓存，即4路slot，那么bucket_num也就是slot_num的1/4。这里我们故意把slot_num调小了点，为的是测试rehash的发生。

#define ASSOC_WAY  (4)  /* 4-way association */

struct hash_table {
    struct hash_slot_cache **buckets;
    struct hash_slot_cache *slots;
    uint32_t slot_num;
    uint32_t bucket_num;
};

int cuckoo_filter_init(size_t size)
{
    ...
    /* Allocate hash slots */
    hash_table.slot_num = nvrom_size / SECTOR_SIZE;
    /* Make rehashing happen */
    hash_table.slot_num /= 4;
    hash_table.slots = calloc(hash_table.slot_num, sizeof(struct hash_slot_cache));
    if (hash_table.slots == NULL) {
        return -1;
    }

    /* Allocate hash buckets associated with slots */
    hash_table.bucket_num = hash_table.slot_num / ASSOC_WAY;
    hash_table.buckets = malloc(hash_table.bucket_num * sizeof(struct hash_slot_cache *));
    if (hash_table.buckets == NULL) {
        free(hash_table.slots);
        return -1;
    }
    for (i = 0; i < hash_table.bucket_num; i++) {
        hash_table.buckets[i] = &hash_table.slots[i * ASSOC_WAY];
    }
}

下面是哈希函数的设计，这里有两个，前面提到既然key是20字节的SHA1值，我们就可以分别是对key的低32位和高32位进行位运算，只要bucket_num满足2的幂次方，我们就可以将key的一部分同bucket_num – 1相与，就可以定位到相应的bucket位置上，注意bucket_num随着rehash而增大，哈希函数简单的好处是求哈希值十分快。

#define cuckoo_hash_lsb(key, count)  (((size_t *)(key))[0] & (count - 1))
#define cuckoo_hash_msb(key, count)  (((size_t *)(key))[1] & (count - 1))

终于要讲解cuckoo filter最重要的三个操作了——查询、插入还有删除。查询操作是简单的，我们对传进来的参数key进行两次哈希求值tag[0]和tag[1]，并先用tag[0]定位到bucket的位置，从4路slot中再去对比tag[1]。只有比中了tag后，由于只是key的一部分，我们再去从flash中验证完整的key，并把数据在flash中的偏移值read_addr输出返回。相应的，如果bucket[tag[0]]的4路slot都没有比中，我们再去bucket[tag[1]]中比对（代码略），如果还比不中，可以肯定这个key不存在。这种设计的好处就是减少了不必要的flash读操作，每次比对的是内存中的tag而不需要完整的key。

static int cuckoo_hash_get(struct hash_table *table, uint8_t *key, uint8_t **read_addr)
{
    int i, j;
    uint8_t *addr;
    uint32_t tag[2], offset;
    struct hash_slot_cache *slot;

    tag[0] = cuckoo_hash_lsb(key, table->bucket_num);
    tag[1] = cuckoo_hash_msb(key, table->bucket_num);

    /* Filter the key and verify if it exists. */
    slot = table->buckets[tag[0]];
    for (i = 0; i bucket_num) == slot[i].tag) {
        if (slot[i].status == OCCUPIED) {
            offset = slot[i].offset;
            addr = key_verify(key, offset);
            if (addr != NULL) {
                if (read_addr != NULL) {
                    *read_addr = addr;
                }
                break;
            }
        } else if (slot[i].status == DELETED) {
            return DELETED;
        }
    }
    ...
}

接下来先将简单的删除操作，之所以简单是因为delete除了将相应slot的状态值设置一下之外，其实什么都没有干，也就是说它不会真正到flash里面去把数据清除掉。为什么？很简单，没有必要。还有一个原因，flash的写操作之前需要擦除整个页面，这种擦除是会折寿的，所以很多flash支持随机读，但必须保持顺序写。

static void cuckoo_hash_delete(struct hash_table *table, uint8_t *key)
{
    uint32_t i, j, tag[2];
    struct hash_slot_cache *slot;

    tag[0] = cuckoo_hash_lsb(key, table->bucket_num);
    tag[1] = cuckoo_hash_msb(key, table->bucket_num);

    slot = table->buckets[tag[0]];
    for (i = 0; i bucket_num) == slot[i].tag) {
        slot[i].status = DELETED;
        return;
    }
    ...
}

了解了flash的读写特性，你就知道为啥插入操作在flash层面要设计成append。不过我们这里不讨论过多flash细节，哈希表层面的插入逻辑其实跟查询差不多，我就不贴代码了。这里要贴的是如何判断并处理碰撞，其实这里也没啥玄机，就是用old_tag和old_offset保存一下临时变量，以便一个元素被踢出去之后还能找到备用的安身之所。但这里会有一个判断，每次踢人都会计数，当alt_cnt大于512时候表示哈希表真的快满了，这时候需要rehash了。

static int cuckoo_hash_collide(struct hash_table *table, uint32_t *tag, uint32_t *p_offset)
{
    int i, j, k, alt_cnt;
    uint32_t old_tag[2], offset, old_offset;
    struct hash_slot_cache *slot;

    /* Kick out the old bucket and move it to the alternative bucket. */
    offset = *p_offset;
    slot = table->buckets[tag[0]];
    old_tag[0] = tag[0];
    old_tag[1] = slot[0].tag;
    old_offset = slot[0].offset;
    slot[0].tag = tag[1];
    slot[0].offset = offset;
    i = 0 ^ 1;
    k = 0;
    alt_cnt = 0;

KICK_OUT:
    slot = table->buckets[old_tag[i]];
    for (j = 0; j < ASSOC_WAY; j++) {
        if (offset == INVALID_OFFSET && slot[j].status == DELETED) {
            slot[j].status = OCCUPIED;
            slot[j].tag = old_tag[i ^ 1];
            *p_offset = offset = slot[j].offset;
            break;
        } else if (slot[j].status == AVAILIBLE) {
            slot[j].status = OCCUPIED;
            slot[j].tag = old_tag[i ^ 1];
            slot[j].offset = old_offset;
            break;
        }
    }

    if (j == ASSOC_WAY) {
        if (++alt_cnt > 512) {
            if (k == ASSOC_WAY - 1) {
                /* Hash table is almost full and needs to be resized */
                return 1;
            } else {
                k++;
            }
        }
        uint32_t tmp_tag = slot[k].tag;
        uint32_t tmp_offset = slot[k].offset;
        slot[k].tag = old_tag[i ^ 1];
        slot[k].offset = old_offset;
        old_tag[i ^ 1] = tmp_tag;
        old_offset = tmp_offset;
        i ^= 1;
        goto KICK_OUT;
    }

    return 0;
}

rehash的逻辑也很简单，无非就是把哈希表中的buckets和slots重新realloc一下，空间扩展一倍，然后再从flash中的key重新插入到新的哈希表里去。这里有个陷阱要注意，千万不能有相同的key混进来！虽然cuckoo hashing不像开链法那样会退化成O(n)，但由于每个元素有两个哈希值，而且每次计算的哈希值随着哈希表rehash的规模而不同，相同的key并不能立即检测到冲突，但当相同的key达到一定规模后，噩梦就开始了，由于rehash里面有插入操作，一旦在这里触发碰撞，又会触发rehash，这时就是一个rehash不断递归的过程，由于其中老的内存没释放，新的内存不断重新分配，整个程序就如同陷入DoS攻击一般瘫痪了。所以每次插入操作前一定要判断一下key是否已经存在过，并且对rehash里的插入使用碰撞断言防止此类情况发生。笔者在测试中不幸中了这样的彩蛋，调试了大半天才搞清楚原因，搞IT的同学们记住一定要防小人啊~

static void cuckoo_rehash(struct hash_table *table)
{
    ...
    uint8_t *read_addr = nvrom_base_addr;
    uint32_t entries = log_entries;
    while (entries--) {
        uint8_t key[20];
        uint32_t offset = read_addr - nvrom_base_addr;
        for (i = 0; i < 20; i++) {
            key[i] = flash_read(read_addr);
            read_addr++;
        }
        /* Duplicated keys in hash table which can cause eternal
         * hashing collision! Be careful of that!
         */
        assert(!cuckoo_hash_put(table, key, &offset));
        if (cuckoo_hash_get(&old_table, key, NULL) == DELETED) {
            cuckoo_hash_delete(table, key);
        }
        read_addr += DAT_LEN;
    }
    ...
}

到此为止代码的逻辑还是比较简单，使用效果如何呢？我来帮你找个大文件unqlite.c测试一下，这是一个嵌入式数据库源代码，共59959行代码。作为需要导入的文件，编译我们的cuckoo filter，然后执行：

./cuckoo_db unqlite.c output.c

你会发现生成output.c正好也是59959行代码，一分不差，probably yes终于变成了definitely yes。同时也可以看到，cuckoo filter真的很快！如果你想看hashing的整个过程，可以参照README里把调试宏打开。最后，欢迎给这个小玩意提交PR！

参考资料

Cuckoo Filter的论文和PPT：Cuckoo Filter: Practically Better Than Bloom

我做这些题目的不是为了要去应聘这些公司，而是为了锻炼一下自己的算法和编程能力。因为我开始工作的时候基本没有这样的训练算法和编程的网站，除了大学里的“算法和数据结构”里的好些最基础最基础的知识，基本上没有什么训练。所以，当我看到有人在做这些题的时候，我也蠢蠢欲动地想去刷一下。

于是，我花了3-4个月的业余时间，我把Leetcode的154道题全部做完了。（这也是最近我没有太多的时间来写博客的原因，你可以看到我之前做的那个活动中有几个算法题来自于Leetcode）有人说我时间太多了，这里声明一下，我基本上都是利用了晚上10点以后的时间来做这些题的。

LeetCode的题大致分成两类：

1）基础算法的知识。这些题里面有大量的算法题，解这些题都是有套路的，不是用递归（深度优先DFS，广度优先BFS），就是要用动态规划（Dynamic Programming），或是拆半查找（Binary Search），或是回溯（Back tracing），或是分治法（Divide and Conquer），还有大量的对树，数组、链表、字符串和hash表的操作。通过做这些题能让你对这些最基础的算法的思路有非常扎实的了解和训练。对我而言，Dynamic Programming 是我的短板，尤其是一些比较复杂的问题，在推导递推公式上总是有思维的缺陷（数学是我的硬伤），通过做了这些题后，我能感到我在DP的思路上有了很大的收获。

2）编程题。比如：atoi，strstr，add two num，括号匹配，字符串乘法，通配符匹配，文件路径简化，Text Justification，反转单词等等，这些题的Edge Case, Corner Case有很多。这些题需要你想清楚了再干，只要你稍有疏忽，就会有几个case让你痛不欲生，而且一不小心就会让你的代码会写得又臭又长，无法阅读。通过做这些题，可以非常好的训练你对各种情况的考虑，以及你对程序代码组织的掌控（其实就是其中的状态变量）。还记得我在《函数式编程》中说的，程序中的状态是你程序变得复杂难维护的直接原因。

我觉得每个程序员都应该花时间和精力做这些题，因为你会从这些题中得到很大的收益。做完这些题后你一定会明白下面几个道理：

1）想清楚了再干。这个观点我以前就在《多些时间可以少些代码》说过。如果你拿到题就上去直接写代码的话，你一定会被各种case打回来了。然后呢，你一着急，你就会进入那种我在《开发团队的效率》中说的那种毫无效率case by case的开发模式，而你也进入了“平庸模式”。于是你就会出现下图那样的情况。

2) 编程是脑力劳动，急不得。这个事情在这做这些题的时候你就会发现，要么是脑子转不过来了，要么就是明明就差一点了，但程序怎么都调不对。如果你越着急的话，你就会发现你会离目标越远，而花的时间也会更多。另外，你会发现这些题基本上都是50行代码内就可以搞定的，但是为了这50行以内的代码，你要花好多时间和精力。coding  50行代码在我们的日常工作中分分钟就完成，而Leetcode里的50行代码却没那么简单，也许，用这个你就可以区别什么是码农，什么是程序员了。

3）加班要不得。因为我总是在晚上10点以后做题，所以，基本上都是在加班状态中工作。这种状态过上两三天，你就会发现，整个大脑已经不转了，而且不但不转，还会犯很多低级错误，很多事情都想不清楚，一个晚上都在和程序的状态控制做搏斗，代码写得越来越乱，越来越没条理。于是这种时候，我都会休息几天，不做题了，然后再做题的时候，就觉得非常地清楚。可见加班 是编程最致命的敌人！

我把我的C++代码放到了Github上，大家也帮我review一下，看看有没有可以改善的。

https://github.com/haoel/leetcode

好了，不多说了，我希望大家有时间都去练练LeetCode，无论是找工作还是对你的编程能力会有非常大的提高。

（全文完）

截止这篇文章发布的时候，fun.coolshell.cn的访问量UV大约有4万左右，通关人数大约有200人，但因为在活动的第二天网上就出了一些答题攻略，通过分析，实际靠自己能力通过的人数在130人左右。通过率大约不到4‰的样子。

在这里我把整个谜题和做这个活动的东西写一下，算是给自己的一个总结。

谜题的答案和花絮

fun.coolshell.cn上一共有十道谜题，要设计这些东西还真是费尽脑汁，这让我对那些设计谜题式游戏的人相当敬佩。

第0关：很多人可能一头雾水，完全不知道这是什么，其实只要Google一下，你会知道这是一个叫BrainFuck的语言。在Coolshell.cn上我也介绍了过——《BT雷人的程序语言》《BT雷人的程序语言（大全）》，要通过这关，你需要把那段程序编译一下。要编译这段程序其实很简单，Google一个在线的编译器就可以了。（关于其它更多的古怪的编程语言请参看这里：http://esolangs.org/wiki/Language_list）

第1关：这一关也是很简单的，你需要在网页上找到两个数，一个是X，一个是Y，然后求得X和Y的乘积。对于X，你可以观察一下那个数列游戏，对于Y，你可以Google一下就知道了（我在Coolshell的《如何用最有创造力的方式输出42》说过这个事）。

第2关：上面显示了一个不一样的键盘，我给了这个键盘的Wikipedia的链接。这个键盘叫Dvorak键，不同于我们的Qwert键。通过这个两个键盘的布局映射，你可以把下面那段读不懂的文字解出来（其实，你还是可以Google，有在线的转换）。把下面那段文字转成Qwert键的，你就会发现这是一段代码，这段代码非常著名，是1987年国际C语言混乱大赛一等奖的一段代码（你可Google “IOCCC 87 unix”）。（关于IOCCC你可以参看Coolshell之前的《6个变态的HelloWorld》、《如何混乱代码》、《如何写出无法维护的代码》这几篇文章）

第3关：扫描二维码以后，你会得到一个码表转换，你可以使用Shell的tr命令来转一下下面的话。转完后你就可以读懂了，读懂了你还需要使用rot13来转一下“shell”（Google一下，你会发现也有在线的转换器，另外还有其它的rot）

第4关：这是众多同学被卡在的地方。很多同学吐槽这题太坑了，别忘了这是游戏啊。我问了几个早先通关的同学，他们都说还好了，只要静一下心来多观察一下，你就会找出规律的。这个回文的模式是，一个大写字符和一个数字（顺序不限）把一个小字母套起来。于是，写成正则表达式是：

([A-Z])([0-9])[a-z]\2\1|([0-9])([A-Z])[a-z]\4\3

用shell命令可以很快地找到9个匹配，然后，像“cat”一样，取中间的小写字母组成一个单词。写成Shell命令是：

grep -o "\([A-Z]\)\([0-9]\)[a-z]\2\1\|\([0-9]\)\([A-Z]\)[a-z]\4\3" cat.txt | sed -E "s/(.)(.)(.)\2\1/\3/g" | awk '{printf("%s",$1)}' && echo ""

这题主要考的是你的观察能力和正则表达式。

第5关：如果你点了一下图片后，你就知道，这个连接http://fun.coolshell.cn/n/2014返回了一个数字，如果你把这个数字放到那个URL中，不断地替换其中的数字，你会得到一个新的数字。于是你就会得到最终的答案。

这道题本来我是想让大家写程序的，我原来设置了一共512个序列，但是考虑到服务受不了，所以，我把它降到了128个，这样保证你的程序可以在几秒钟内得到结果，而不会对我的服务器造成压力。但是我还是看到好几个同学人肉地copy+paste+回车刷了100多下，得到了最终答案。

第6关：通过中序和后序遍历还原一棵二叉树，然后再找到其最深的路径，然后得到一个字符串后，把这个字符串做为一个passcode代入那个openssl的命令行中。你就可以解密密文得到下一关的答案。

这个题，我本想设计得更隐晦一些，用一个“心脏流血”的图片来暗示openssl，然后用别的东西暗示AES-128-CBC，后来想想算了，主要还是考大家在大学里的二叉树的最基本的算法。并介绍一下openssl的shell命令行加解密的方法。

在网上的一些攻略中我看到了大家没有用程序，而是手动地花了一棵树出来。（其实，这设计这道的时候，我本来想设计成随机树，也就每个人看到的答案都不一样，我随机建树并且找最深路径的程序都写好了，但是我最终还是没有这样做，因为这无疑增加我对这个网页游戏的代码复杂度，而我又没有太多的时间，而谜题的各种形式已经够让我花精力的了，你虽然看到了10道题，但是其实我设计了一共有16道题，我反复斟酌，即不想为难大家，又不想太简单和无聊，所以最终release了这十道题）

第7关：N皇后问题，这个问题也是大学里的题。9皇后一共有352个解，你需要把这352个解代到那个sha1的公式中（需要上一关用于解密的passcode），这样你就会得到一个解。然后这就是通关口令。

第6关和第7关的算法题你要是不会写的话，Google一下，反正我们是“大自然的搬运工”，不是吗？呵呵。

第7关这题啊，我看到一个同学用穷举的1-9的排序组合的方式来向服务发请求，从123456789开始，我都看SB了，因为这关的通答案是9开头，我勒了个去！你得对我的服务器发多少次请求啊，才能得到一个200的回复啊。TNND。服了。不过这个同学我最终还是给通过了，没有判定成作弊。

第8关：Excel的列号编程，这一关写成代码其实并不难的。但我看到网上给的好些答案，大家都是用手算。也OK，这题本身就没有什么难度，但是因为这个26进制是从1开始的，写出来的代码并不非常容易，一些边界条件很容易就break掉了。这题完全考的是编码。把COOLSHELL除以SHELL的数转成字符串。然后就进入最后一关了。

然后，我又见到有个同学用了穷举的方式，TNND，其实每道题都有人在用穷举的方式，我勒个去。他从AAA开始穷举，不一会就穷举出正确答案了。尼玛！

第9关：一个猪圈和一个共济会的logo，你Google一下，你就知道答案了。这题纯粹就是介绍知识的。不知道大家有没有去wikipedia上了解了一下这个猪圈密码和共济会是怎么一回事吗？这样的密文叫图片密文，还有很多类似的图片密文的。你知道吗？有相应的字库哦。也有在线的生成器哦。（因为我最近在学各种安全的基础知识，所以了解到了这个东西）

通关：于是你就通关了。你会发现你得到了一个helloworld，这个字符串，在我一放出来这个谜题的时候，就有很多人在尝试helloworld就是那段brainfuck的代码的输出。我汗啊。还好我做了一个比较复杂的防作弊检查……

总体来说，这些关卡都不难，但是你最少也得用2-3个小时。Top100页面时统计的平均时间是10个半小时。

再说一个花絮，自从，8月3日上线后，8月4日在网上就有了相关的解答攻略，还是在V2EX上，于是出现了好些只花了几分钟就做完了的人。不过好在事先我就预料到了这个事，事先预备好了“反作弊分析”的脚本，细节不想说太多，反正就是说，我会记录你答案的整个过程和行为，以此来确保TOP100中的人基本都是用自己能力答的，当然，可能会有漏判，但至少也是写过代码的。

活动心得

因为是第一次做活动，所以有很多感想，下面写下一些做这个活动的心得，供大家参考：

1）要做好一个这样的解题游戏并不简单。

• 关卡设计：最花力气的地方就是设计每个关卡，我不能设计得太过隐晦，也不能设计得太过明显。最好是要符合参与者的能力，但又要高于平均以上水平的能力，最好在90%以上。这样会让大家有挑战感，但是又不会有挫败感。这个度相当难把握。总体而言，本次设计的谜题中还有很多可以改进的地方。但这毕竟是我的第一次，也算是我用其来感受一下应该怎么设计游戏。

• 游戏黏性：除了设计谜题，还需要针对用户可能会答错的地方来给用户一些提示，原因也是为了不让用户有挫败感，虽然用户没有答对，但是需要用这些页面来鼓励用户You made some progress，这个很重要。这会让用户对游戏更有粘性，并且更愿意有更多的投入。找到这些地方也不是一件容易的事，因为做为游戏的设计者来说，很难从一个不知到答案的角度去思考。所以需要试玩，在fun.coolshell.cn正式release之前，我找了几个人比较聪明的人来试玩了一下，对这个游戏的帮助很大。

• 游戏管理：这样的一个在线游戏自然会出一些作弊者，为了游戏的公平性，你需要剔除这些作弊者。所以，我设计了一些比较简单的记录用户所有过程的监测的算法。通过cookie和后台的http log来一同分析。这个部分也比较地花时间。我上周六的时候写这些代码写到了凌晨4点，导致脑子不清楚，出了些bug，导致在大家游戏过程中重置cookie等伤害用户体验的事件。所以说啊，不能赶啊，也不能加班啊。

2）关于怎么做一个活动的感想。

• 这次活动的背景。首先，想做这个活动的起因是这样的。我一个朋友在微博上做活动——“转发微博或@几个人怎么怎么滴就有机获得什么什么的”，我在这里把这种活动简称为“转就送”活动。于是遭到了水军的刷奖品，导致他根本分不清楚哪些是正常人，哪些不是，因为新浪微博上有大量的这要瓣机器人，所以他这次活动最后失败了。我说，你得加点难度啊，要加点智商啊。而且，我看过太多的活动都是这样的，而且很多公司的活动也是这样的，我觉得太low了。于是，我就萌生了自己尝试一下的念头。

• 我对做活动的理解。我一直觉得网上那些诸如“转就送”或是“抽奖”这样的活动都比较SB，这些人根本就不知道怎么做活动。这样做活动不需要智商，简单粗暴，效果一点也不好，活动做完了，人就走了，人们马上就忘了。我以为做活动的精髓是这样的：

• 真正的价值。其实，好的活动并不只是物品的价格，而是参与这个过程的感觉和体会。如果你让人觉得这是碰运气的，那么这个活动除了用物品价格来吸引人，也就没别的什么了。如果这个活动的参与过程是让人有成就感的，要有成就感那么就需要有一定难度的挑战，而且这种挑战也是让众人认可和佩服的，那么这个奖品的价格再小，价值也会很大。比如：Olympic Game，World Cup之流的，世界顶尖，四年一次，来之不易。这才是活动的价值。本次的fun.coolshell.cn上的活动，我希望让大家在做题的过程中学到一些东西，另外也希望做出来的人有一种成就感。

• 让人有回味。那些简单的“转就送”式的活动不会让人产生任何的回味，只会让人产生很大的反感。就像那些“让你转发，不转就死全家”的东西，相当的让人反感。真正的回味是人们对活动参与过程的讨论和交互。在fun.coolshell.cn上线后，我就看到好几个社区在讨论这些谜题，这就是所谓的回味。只有人们对过程的回味，对参与的回味，才会让这个活动真正的成功。

• 暴露活动过程。有挑战的活动，一定要有一个Who’s Who的东西，而且是随时动态更新的可以让大家查询的，这样才会从另一个侧面激发大家的热情。因为fun.coolshell.cn一开始说了只给前十个人送东西，结果在过程中，我发现了就半天时间就差不多满了，那时我在想，如果没有奖品了，剩下的人还会不会玩了？于是我飞快地开发了一个TOP100的排行榜，让大家可以看得到这个过程，虽然前十以后就没有奖品了，但是，能上这TOP100也不错。于是乎，在没有奖品情况下，依然在激发着大家的解题热情。有竞争总是一件有意思的事情，因为成就感总是来自竞争。（注：为什么top100中会有“xxxxxx”的用户，因为一开始我用的是用户提交的name，但是后来有人告诉我，这个名字可能是真名，所以，我就改成了weibo或twitter的ID，而xxxxx则是没有留下微博或twitter的）

最后吐个槽，我真的觉得那些“纯靠运气的活动”相当的SB，我看到好些公司的运营部门招了多少个所谓的高学历和高能力的人，结果干出来的运营活动的水平，其实，也就是个有小学文化水平的人就可以做的了。那些“转就送式的”、“抽奖式的”的活动，是个人都会干，根本不需要高学历的人。

其它

1）本次活动中，有一个隐藏关卡，还没有人找出来。要能达到隐藏关卡，需要完成所有的题目。

2）活动的通关页是HelloWorld，这意味着——这仅仅是个开始。

最后感谢大家为这个活动付出的时间！

（全文完）

精神不佳我就不写文章了。于是，世界杯过后，我每天都会抽出每天晚上和周末的一些碎片时间，我仿照一些前端过关的游戏，做了几个和程序员有关的迷题，也是要通关的，不过和前端知识没什么关系。这个游戏我放到了下面这个二级域名下。

http://fun.coolshell.cn/

有兴趣的朋友可以去玩玩。通关的同学我会送你们《Unix环境高级编程（第三版）》（感谢@出版圈郭志敏 赞助）或一个马克杯（感谢@linux命令行精选网 赞助）），因为奖品数量有限，所以，我会送给前十个通关的同学（后面通关的我会随机抽几个）。

最后说一下这些迷题：

1）目前一共有10个迷题。你通关会出现个Congratulations的页面和一个表单，希望你能提供一下你的联系方式（联系方式只要你的email/weibo/twitter/homepage这样你比较公开的方式）。

2）为了突出fun，所以，这些迷题中有好些基于一些“有趣”的知识的（可能有些知识你是不知道的）。

3）我使用了英文，只希望你对英文不要害怕，英文是程序员最关键的一项技能。（虽然我的英文也一般）

4）你要通关的话，你可能需要很多的Google/Wikipedia，所以，你可能需要翻墙环境。我希望你能经常翻墙。

5）另外，如果要通关的话，你需除了有比较好的观察能力，你还需要对Linux命令行有一些了解，有一半左右的题是需要写代码才能过的，写代码的题中有字符串匹配（正则表达式），网络请求，算法和数据结构，以及一些基础的加密解密知识。

6）这些题并不难，而且谜面提示得应该是非常清楚，不过，你要做完最快也需要2-3个小时，所以，在这里还是谢谢你的时间。

祝大家玩得愉快！

————更新：2014/8/5————

本活动已结果，题的页面还在保留中……

（全文完）

关于QR Code Specification，可参看这个PDF：http://raidenii.net/files/datasheets/misc/qr_code.pdf 

基础知识

首先，我们先说一下二维码一共有40个尺寸。官方叫版本Version。Version 1是21 x 21的矩阵，Version 2是 25 x 25的矩阵，Version 3是29的尺寸，每增加一个version，就会增加4的尺寸，公式是：(V-1)*4 + 21（V是版本号） 最高Version 40，(40-1)*4+21 = 177，所以最高是177 x 177 的正方形。

下面我们看看一个二维码的样例：

定位图案

• Position Detection Pattern是定位图案，用于标记二维码的矩形大小。这三个定位图案有白边叫Separators for Postion Detection Patterns。之所以三个而不是四个意思就是三个就可以标识一个矩形了。

• Timing Patterns也是用于定位的。原因是二维码有40种尺寸，尺寸过大了后需要有根标准线，不然扫描的时候可能会扫歪了。

• Alignment Patterns 只有Version 2以上（包括Version2）的二维码需要这个东东，同样是为了定位用的。

功能性数据

• Format Information 存在于所有的尺寸中，用于存放一些格式化数据的。

• Version Information 在 >= Version 7以上，需要预留两块3 x 6的区域存放一些版本信息。

数据码和纠错码

• 除了上述的那些地方，剩下的地方存放 Data Code 数据码 和 Error Correction Code 纠错码。

数据编码

我们先来说说数据编码。QR码支持如下的编码：

Numeric mode 数字编码，从0到9。如果需要编码的数字的个数不是3的倍数，那么，最后剩下的1或2位数会被转成4或7bits，则其它的每3位数字会被编成 10，12，14bits，编成多长还要看二维码的尺寸（下面有一个表Table 3说明了这点）

Alphanumeric mode 字符编码。包括 0-9，大写的A到Z（没有小写），以及符号$ % * + – . / : 包括空格。这些字符会映射成一个字符索引表。如下所示：（其中的SP是空格，Char是字符，Value是其索引值） 编码的过程是把字符两两分组，然后转成下表的45进制，然后转成11bits的二进制，如果最后有一个落单的，那就转成6bits的二进制。而编码模式和字符的个数需要根据不同的Version尺寸编成9, 11或13个二进制（如下表中Table 3）

Byte mode, 字节编码，可以是0-255的ISO-8859-1字符。有些二维码的扫描器可以自动检测是否是UTF-8的编码。

Kanji mode 这是日文编码，也是双字节编码。同样，也可以用于中文编码。日文和汉字的编码会减去一个值。如：在0X8140 to 0X9FFC中的字符会减去8140，在0XE040到0XEBBF中的字符要减去0XC140，然后把结果前两个16进制位拿出来乘以0XC0，然后再加上后两个16进制位，最后转成13bit的编码。如下图示例：

Extended Channel Interpretation (ECI) mode 主要用于特殊的字符集。并不是所有的扫描器都支持这种编码。

Structured Append mode 用于混合编码，也就是说，这个二维码中包含了多种编码格式。

FNC1 mode 这种编码方式主要是给一些特殊的工业或行业用的。比如GS1条形码之类的。

简单起见，后面三种不会在本文 中讨论。

下面两张表中，

• Table 2 是各个编码格式的“编号”，这个东西要写在Format Information中。注：中文是1101

• Table 3 表示了，不同版本（尺寸）的二维码，对于，数字，字符，字节和Kanji模式下，对于单个编码的2进制的位数。（在二维码的规格说明书中，有各种各样的编码规范表，后面还会提到）

下面我们看几个示例，

示例一：数字编码

在Version 1的尺寸下，纠错级别为H的情况下，编码： 01234567

1. 把上述数字分成三组: 012 345 67

2. 把他们转成二进制:  012 转成 0000001100；  345 转成 0101011001；  67 转成 1000011。

3. 把这三个二进制串起来: 0000001100 0101011001 1000011

4. 把数字的个数转成二进制 (version 1-H是10 bits ): 8个数字的二进制是 0000001000

5. 把数字编码的标志0001和第4步的编码加到前面:  0001 0000001000 0000001100 0101011001 1000011

示例二：字符编码

在Version 1的尺寸下，纠错级别为H的情况下，编码: AC-42

1. 从字符索引表中找到 AC-42 这五个字条的索引 (10,12,41,4,2)

2. 两两分组: (10,12) (41,4) (2)

3.把每一组转成11bits的二进制:

(10,12) 10*45+12 等于 462 转成 00111001110
(41,4) 41*45+4 等于 1849 转成 11100111001
(2) 等于 2 转成 000010

4. 把这些二进制连接起来：00111001110 11100111001 000010

5. 把字符的个数转成二进制 (Version 1-H为9 bits ): 5个字符，5转成 000000101

6. 在头上加上编码标识 0010 和第5步的个数编码:  0010 000000101 00111001110 11100111001 000010

结束符和补齐符

假如我们有个HELLO WORLD的字符串要编码，根据上面的示例二，我们可以得到下面的编码，

我们还要加上结束符：

按8bits重排

如果所有的编码加起来不是8个倍数我们还要在后面加上足够的0，比如上面一共有78个bits，所以，我们还要加上2个0，然后按8个bits分好组：

00100000   01011011   00001011   01111000   11010001   01110010   11011100   01001101   01000011   01000000

补齐码（Padding Bytes）

最后，如果如果还没有达到我们最大的bits数的限制，我们还要加一些补齐码（Padding Bytes），Padding Bytes就是重复下面的两个bytes：11101100 00010001 （这两个二进制转成十进制是236和17，我也不知道为什么，只知道Spec上是这么写的）关于每一个Version的每一种纠错级别的最大Bits限制，可以参看QR Code Spec的第28页到32页的Table-7一表。

假设我们需要编码的是Version 1的Q纠错级，那么，其最大需要104个bits，而我们上面只有80个bits，所以，还需要补24个bits，也就是需要3个Padding Bytes，我们就添加三个，于是得到下面的编码：

00100000 01011011 00001011 01111000 11010001 01110010 11011100 01001101 01000011 01000000 11101100 00010001 11101100

上面的编码就是数据码了，叫Data Codewords，每一个8bits叫一个codeword，我们还要对这些数据码加上纠错信息。

纠错码

上面我们说到了一些纠错级别，Error Correction Code Level，二维码中有四种级别的纠错，这就是为什么二维码有残缺还能扫出来，也就是为什么有人在二维码的中心位置加入图标。

那么，QR是怎么对数据码加上纠错码的？首先，我们需要对数据码进行分组，也就是分成不同的Block，然后对各个Block进行纠错编码，对于如何分组，我们可以查看QR Code Spec的第33页到44页的Table-13到Table-22的定义表。注意最后两列：

• Number of Error Code Correction Blocks ：需要分多少个块。

• Error Correction Code Per Blocks：每一个块中的code个数，所谓的code的个数，也就是有多少个8bits的字节。

举个例子：上述的Version 5 + Q纠错级：需要4个Blocks（2个Blocks为一组，共两组），头一组的两个Blocks中各15个bits数据 + 各 9个bits的纠错码（注：表中的codewords就是一个8bits的byte）（再注：最后一例中的（c, k, r ）的公式为：c = k + 2 * r，因为后脚注解释了：纠错码的容量小于纠错码的一半）

下图给一个5-Q的示例（因为二进制写起来会让表格太大，所以，我都用了十进制，我们可以看到每一块的纠错码有18个codewords，也就是18个8bits的二进制数）

注：二维码的纠错码主要是通过Reed-Solomon error correction（里德-所罗门纠错算法）来实现的。对于这个算法，对于我来说是相当的复杂，里面有很多的数学计算，比如：多项式除法，把1-255的数映射成2的n次方（0<=n<=255）的伽罗瓦域Galois Field之类的神一样的东西，以及基于这些基础的纠错数学公式，因为我的数据基础差，对于我来说太过复杂，所以我一时半会儿还有点没搞明白，还在学习中，所以，我在这里就不展开说这些东西了。还请大家见谅了。（当然，如果有朋友很明白，也繁请教教我）

最终编码

穿插放置

如果你以为我们可以开始画图，你就错了。二维码的混乱技术还没有玩完，它还要把数据码和纠错码的各个codewords交替放在一起。如何交替呢，规则如下：

对于数据码：把每个块的第一个codewords先拿出来按顺度排列好，然后再取第一块的第二个，如此类推。如：上述示例中的Data Codewords如下：

我们先取第一列的：67， 246， 182， 70

然后再取第二列的：67， 246， 182， 70， 85，246，230 ，247

如此类推：67， 246， 182， 70， 85，246，230 ，247 ………  ……… ，38，6，50，17，7，236

对于纠错码，也是一样：

和数据码取的一样，得到：213，87，148，235，199，204，116，159，…… …… 39，133，141，236

然后，再把这两组放在一起（纠错码放在数据码之后）得到：

67, 246, 182, 70, 85, 246, 230, 247, 70, 66, 247, 118, 134, 7, 119, 86, 87, 118, 50, 194, 38, 134, 7, 6, 85, 242, 118, 151, 194, 7, 134, 50, 119, 38, 87, 16, 50, 86, 38, 236, 6, 22, 82, 17, 18, 198, 6, 236, 6, 199, 134, 17, 103, 146, 151, 236, 38, 6, 50, 17, 7, 236, 213, 87, 148, 235, 199, 204, 116, 159, 11, 96, 177, 5, 45, 60, 212, 173, 115, 202, 76, 24, 247, 182, 133, 147, 241, 124, 75, 59, 223, 157, 242, 33, 229, 200, 238, 106, 248, 134, 76, 40, 154, 27, 195, 255, 117, 129, 230, 172, 154, 209, 189, 82, 111, 17, 10, 2, 86, 163, 108, 131, 161, 163, 240, 32, 111, 120, 192, 178, 39, 133, 141, 236

这就是我们的数据区。

Remainder Bits

最后再加上Reminder Bits，对于某些Version的QR，上面的还不够长度，还要加上Remainder Bits，比如：上述的5Q版的二维码，还要加上7个bits，Remainder Bits加零就好了。关于哪些Version需要多少个Remainder bit，可以参看QR Code Spec的第15页的Table-1的定义表。

画二维码图

Position Detection Pattern

首先，先把Position Detection图案画在三个角上。（无论Version如何，这个图案的尺寸就是这么大）

Alignment Pattern

然后，再把Alignment图案画上（无论Version如何，这个图案的尺寸就是这么大）

关于Alignment的位置，可以查看QR Code Spec的第81页的Table-E.1的定义表（下表是不完全表格）

下图是根据上述表格中的Version8的一个例子（6，24，42）

Timing Pattern

接下来是Timing Pattern的线（这个不用多说了）

Format Information

再接下来是Formation Information，下图中的蓝色部分。

Format Information是一个15个bits的信息，每一个bit的位置如下图所示：（注意图中的Dark Module，那是永远出现的）

这15个bits中包括：

• 5个数据bits：其中，2个bits用于表示使用什么样的Error Correction Level， 3个bits表示使用什么样的Mask
• 10个纠错bits。主要通过BCH Code来计算

然后15个bits还要与101010000010010做XOR操作。这样就保证不会因为我们选用了00的纠错级别和000的Mask，从而造成全部为白色，这会增加我们的扫描器的图像识别的困难。

下面是一个示例：

关于Error Correction Level如下表所示：

关于Mask图案如后面的Table 23所示。

Version Information

再接下来是Version Information（版本7以后需要这个编码），下图中的蓝色部分。

Version Information一共是18个bits，其中包括6个bits的版本号以及12个bits的纠错码，下面是一个示例：

而其填充位置如下：

数据和数据纠错码

然后是填接我们的最终编码，最终编码的填充方式如下：从左下角开始沿着红线填我们的各个bits，1是黑色，0是白色。如果遇到了上面的非数据区，则绕开或跳过。

掩码图案

这样下来，我们的图就填好了，但是，也许那些点并不均衡，如果出现大面积的空白或黑块，会告诉我们扫描识别的困难。所以，我们还要做Masking操作（靠，还嫌不复杂）QR的Spec中说了，QR有8个Mask你可以使用，如下所示：其中，各个mask的公式在各个图下面。所谓mask，说白了，就是和上面生成的图做XOR操作。Mask只会和数据区进行XOR，不会影响功能区。（注：选择一个合适的Mask也是有算法的）

其Mask的标识码如下所示：（其中的i,j分别对应于上图的x,y）

下面是Mask后的一些样子，我们可以看到被某些Mask XOR了的数据变得比较零散了。

Mask过后的二维码就成最终的图了。

好了，大家可以去尝试去写一下QR的编码程序，当然，你可以用网上找个Reed Soloman的纠错算法的库，或是看看别人的源代码是怎么实现这个繁锁的编码。

（全文完）

提起buddy system相信很多人不会陌生，它是一种经典的内存分配算法，大名鼎鼎的Linux底层的内存管理用的就是它。这里不探讨内核这么复杂实现，而仅仅是将该算法抽象提取出来，同时给出一份及其简洁的源码实现，以便定制扩展。

伙伴分配的实质就是一种特殊的“分离适配”，即将内存按2的幂进行划分，相当于分离出若干个块大小一致的空闲链表，搜索该链表并给出同需求最佳匹配的大小。其优点是快速搜索合并（O(logN)时间复杂度）以及低外部碎片（最佳适配best-fit）；其缺点是内部碎片，因为按2的幂划分块，如果碰上66单位大小，那么必须划分128单位大小的块。但若需求本身就按2的幂分配，比如可以先分配若干个内存池，在其基础上进一步细分就很有吸引力了。

可以在维基百科上找到该算法的描述，大体如是：

分配内存：

1.寻找大小合适的内存块（大于等于所需大小并且最接近2的幂，比如需要27，实际分配32）

1.如果找到了，分配给应用程序。
2.如果没找到，分出合适的内存块。

1.对半分离出高于所需大小的空闲内存块
2.如果分到最低限度，分配这个大小。
3.回溯到步骤1（寻找合适大小的块）
4.重复该步骤直到一个合适的块

释放内存：

1.释放该内存块

1.寻找相邻的块，看其是否释放了。
2.如果相邻块也释放了，合并这两个块，重复上述步骤直到遇上未释放的相邻块，或者达到最高上限（即所有内存都释放了）。

上面这段文字对你来说可能看起来很费劲，没事，我们看个内存分配和释放的示意图你就知道了：

上图中，首先我们假设我们一个内存块有1024K，当我们需要给A分配70K内存的时候，

1. 我们发现1024K的一半大于70K，然后我们就把1024K的内存分成两半，一半512K。
2. 然后我们发现512K的一半仍然大于70K，于是我们再把512K的内存再分成两半，一半是128K。
3. 此时，我们发现128K的一半小于70K，于是我们就分配为A分配128K的内存。

后面的，B，C，D都这样，而释放内存时，则会把相邻的块一步一步地合并起来（合并也必需按分裂的逆操作进行合并）。

我们可以看见，这样的算法，用二叉树这个数据结构来实现再合适不过了。

我在网上分别找到cloudwu和wuwenbin写的两份开源实现和测试用例。实际上后一份是对前一份的精简和优化，本文打算从后一份入手讲解，因为这份实现真正体现了“极简”二字，追求突破常规的，极致简单的设计。网友对其评价甚高，甚至可用作教科书标准实现，看完之后回过头来看cloudwu的代码就容易理解了。

分配器的整体思想是，通过一个数组形式的完全二叉树来监控管理内存，二叉树的节点用于标记相应内存块的使用状态，高层节点对应大的块，低层节点对应小的块，在分配和释放中我们就通过这些节点的标记属性来进行块的分离合并。如图所示，假设总大小为16单位的内存，我们就建立一个深度为5的满二叉树，根节点从数组下标[0]开始，监控大小16的块；它的左右孩子节点下标[1~2]，监控大小8的块；第三层节点下标[3~6]监控大小4的块……依此类推。

在分配阶段，首先要搜索大小适配的块，假设第一次分配3，转换成2的幂是4，我们先要对整个内存进行对半切割，从16切割到4需要两步，那么从下标[0]节点开始深度搜索到下标[3]的节点并将其标记为已分配。第二次再分配3那么就标记下标[4]的节点。第三次分配6，即大小为8，那么搜索下标[2]的节点，因为下标[1]所对应的块被下标[3~4]占用了。

在释放阶段，我们依次释放上述第一次和第二次分配的块，即先释放[3]再释放[4]，当释放下标[4]节点后，我们发现之前释放的[3]是相邻的，于是我们立马将这两个节点进行合并，这样一来下次分配大小8的时候，我们就可以搜索到下标[1]适配了。若进一步释放下标[2]，同[1]合并后整个内存就回归到初始状态。

还是看一下源码实现吧，首先是伙伴分配器的数据结构：

struct buddy2 {
  unsigned size;
  unsigned longest[1];
};

这里的成员size表明管理内存的总单元数目（测试用例中是32），成员longest就是二叉树的节点标记，表明所对应的内存块的空闲单位，在下文中会分析这是整个算法中最精妙的设计。此处数组大小为1表明这是可以向后扩展的（注：在GCC环境下你可以写成longest[0]，不占用空间，这里是出于可移植性考虑），我们在分配器初始化的buddy2_new可以看到这种用法。

struct buddy2* buddy2_new( int size ) {
  struct buddy2* self;
  unsigned node_size;
  int i;

  if (size < 1 || !IS_POWER_OF_2(size))
    return NULL;

  self = (struct buddy2*)ALLOC( 2 * size * sizeof(unsigned));
  self->size = size;
  node_size = size * 2;

  for (i = 0; i < 2 * size - 1; ++i) {
    if (IS_POWER_OF_2(i+1))
      node_size /= 2;
    self->longest[i] = node_size;
  }
  return self;
}

整个分配器的大小就是满二叉树节点数目，即所需管理内存单元数目的2倍。一个节点对应4个字节，longest记录了节点所对应的的内存块大小。

内存分配的alloc中，入参是分配器指针和需要分配的大小，返回值是内存块索引。alloc函数首先将size调整到2的幂大小，并检查是否超过最大限度。然后进行适配搜索，深度优先遍历，当找到对应节点后，将其longest标记为0，即分离适配的块出来，并转换为内存块索引offset返回，依据二叉树排列序号，比如内存总体大小32，我们找到节点下标[8]，内存块对应大小是4，则offset = (8+1)*4-32 = 4，那么分配内存块就从索引4开始往后4个单位。

int buddy2_alloc(struct buddy2* self, int size) {
  unsigned index = 0;
  unsigned node_size;
  unsigned offset = 0;

  if (self==NULL)
    return -1;

  if (size <= 0)
    size = 1;
  else if (!IS_POWER_OF_2(size))
    size = fixsize(size);

  if (self->longest[index] < size)
    return -1;

  for(node_size = self->size; node_size != size; node_size /= 2 ) {
    if (self->longest[LEFT_LEAF(index)] >= size)
      index = LEFT_LEAF(index);
    else
      index = RIGHT_LEAF(index);
  }

  self->longest[index] = 0;
  offset = (index + 1) * node_size - self->size;

  while (index) {
    index = PARENT(index);
    self->longest[index] =
      MAX(self->longest[LEFT_LEAF(index)], self->longest[RIGHT_LEAF(index)]);
  }

  return offset;
}

在函数返回之前需要回溯，因为小块内存被占用，大块就不能分配了，比如下标[8]标记为0分离出来，那么其父节点下标[0]、[1]、[3]也需要相应大小的分离。将它们的longest进行折扣计算，取左右子树较大值，下标[3]取4，下标[1]取8，下标[0]取16，表明其对应的最大空闲值。

在内存释放的free接口，我们只要传入之前分配的内存地址索引，并确保它是有效值。之后就跟alloc做反向回溯，从最后的节点开始一直往上找到longest为0的节点，即当初分配块所适配的大小和位置。我们将longest恢复到原来满状态的值。继续向上回溯，检查是否存在合并的块，依据就是左右子树longest的值相加是否等于原空闲块满状态的大小，如果能够合并，就将父节点longest标记为相加的和（多么简单！）。

void buddy2_free(struct buddy2* self, int offset) {
  unsigned node_size, index = 0;
  unsigned left_longest, right_longest;

  assert(self && offset >= 0 && offset < size);

  node_size = 1;
  index = offset + self->size - 1;

  for (; self->longest[index] ; index = PARENT(index)) {
    node_size *= 2;
    if (index == 0)
      return;
  }

  self->longest[index] = node_size;

  while (index) {
    index = PARENT(index);
    node_size *= 2;

    left_longest = self->longest[LEFT_LEAF(index)];
    right_longest = self->longest[RIGHT_LEAF(index)];

    if (left_longest + right_longest == node_size)
      self->longest[index] = node_size;
    else
      self->longest[index] = MAX(left_longest, right_longest);
  }
}

上面两个成对alloc/free接口的时间复杂度都是O(logN)，保证了程序运行性能。然而这段程序设计的独特之处就在于使用加权来标记内存空闲状态，而不是一般的有限状态机，实际上longest既可以表示权重又可以表示状态，状态机就毫无必要了，所谓“少即是多”嘛！反观cloudwu的实现，将节点标记为UNUSED/USED/SPLIT/FULL四个状态机，反而会带来额外的条件判断和管理实现，而且还不如数值那样精确。从逻辑流程上看，wuwenbin的实现简洁明了如同教科书一般，特别是左右子树的走向，内存块的分离合并，块索引到节点下标的转换都是一步到位，不像cloudwu充斥了大量二叉树的深度和长度的间接计算，让代码变得晦涩难读，这些都是longest的功劳。一个“极简”的设计往往在于你想不到的突破常规思维的地方。

这份代码唯一的缺陷就是longest的大小是4字节，内存消耗大。但cloudwu的博客上有人提议用logN来保存值，这样就能实现uint8_t大小了，看，又是一个“极简”的设计！

说实话，很难在网上找到比这更简约更优雅的buddy system实现了——至少在Google上如此。

二叉树(Binary Tree)的前序、中序和后续遍历是算法和数据结构中的基本问题，基于递归的二叉树遍历算法更是递归的经典应用。

假设二叉树结点定义如下：

// C++
struct Node {
    int value;
    Node *left;
    Node *right;
}

中序递归遍历算法：

// C++
void inorder_traverse(Node *node) {
    if (NULL != node->left) {
        inorder_traverse(node->left);
    }
    do_something(node);
    if (NULL != node->right) {
        inorder_traverse(node->right);
    }
}

前序和后序遍历算法类似。

但是，仅有遍历算法是不够的，在许多应用中，我们还需要对遍历本身进行抽象。假如有一个求和的函数sum，我们希望它能应用于链表，数组，二叉树等等不同的数据结构。这时，我们可以抽象出迭代器(Iterator)的概念，通过迭代器把算法和数据结构解耦了，使得通用算法能应用于不同类型的数据结构。我们可以把sum函数定义为：

int sum(Iterator it)

链表作为一种线性结构，它的迭代器实现非常简单和直观，而二叉树的迭代器实现则不那么容易，我们不能直接将递归遍历转换为迭代器。究其原因，这是因为二叉树递归遍历过程是编译器在调用栈上自动进行的，程序员对这个过程缺乏足够的控制。既然如此，那么我们如果可以自己来控制整个调用栈的进栈和出栈不是就达到控制的目的了吗？我们先来看看二叉树遍历的非递归算法：

// C++
void inorder_traverse_nonrecursive(Node *node) {
    Stack stack;
    do {
        // node代表当前准备处理的子树，层层向下把左孩子压栈，对应递归算法的左子树递归
        while (NULL != node) {
            stack.push(node);
            node = node->left;
        }
        do {
            Node *top = stack.top();
            stack.pop(); //弹出栈顶，对应递归算法的函数返回
            do_something(top);
            if (NULL != top->right) {
                node = top->right; //将当前子树置为刚刚遍历过的结点的右孩子，对应递归算法的右子树递归
                break;
            }
        }
        while (!stack.empty());
    }
    while (!stack.empty());
}

通过基于栈的非递归算法我们获得了对于遍历过程的控制，下面我们考虑如何将其封装为迭代器呢？ 这里关键在于理解遍历的过程是由栈的状态来表示的，所以显然迭代器内部应该包含一个栈结构，每次迭代的过程就是对栈的操作。假设迭代器的接口为：

// C++
class Iterator {
    public:
        virtual Node* next() = 0;
};

下面是一个二叉树中序遍历迭代器的实现：

//C++
class InorderIterator : public Iterator {
    public:
        InorderIterator(Node *node) {
            Node *current = node;
            while (NULL != current) {
                mStack.push(current);
                current = current->left;
            }
        }
        virtual Node* next() {
            if (mStack.empty()) {
                return NULL;
            }
            Node *top = mStack.top();
            mStack.pop();
            if (NULL != top->right) {
                Node *current = top->right;
                while (NULL != current) {
                    mStack.push(current);
                    current = current->left;
                }
            }
            return top;
         }
    private:
        std::stack<Node*> mStack;
};

下面我们再来考察一下这个迭代器实现的时间和空间复杂度。很显然，由于栈中最多需要保存所有的结点，所以其空间复杂度是O(n)的。那么时间复杂度呢？一次next()调用也最多会进行n次栈操作，而整个遍历过程需要调用n次next()，那么是不是整个迭代器的时间复杂度就是O(n^2)呢？答案是否定的！因为每个结点只会进栈和出栈一次，所以整个迭代过程的时间复杂度依然为O(n)。其实，这和递归遍历的时空复杂度完全一样。

除了上面显式利用栈控制代码执行顺序外，在支持yield语义的语言（C#, Python等)中，还有更为直接的做法。下面基于yield的二叉树中序遍历的Python实现：

// Python
def inorder(t):
    if t:
        for x in inorder(t.left):
            yield x
        yield t.label
        for x in inorder(t.right):
            yield x

yield与return区别的一种通俗解释是yield返回时系统会保留函数调用的状态，下次该函数被调用时会接着从上次的执行点继续执行，这是一种与栈语义所完全不同的流程控制语义。我们知道Python的解释器是C写的，但是C并不支持yield语义，那么解释器是如何做到对yield的支持的呢？ 有了上面把递归遍历变换为迭代遍历的经验，相信你已经猜到Python解释器一定是对yield代码进行了某种变换。如果你已经能够实现递归变非递归，不妨尝试一下能否写一段编译程序将yield代码变换为非yield代码。

我深深地发现，对于我国这样从小被父母和老师安排各种事情长大的人，当有一天，父母和老师都跟不上的时候，我们几乎完全不知道怎么去做选择。而我最近也离开了亚马逊，换了一个工作。又正值年底，就像去年的那篇《三个故事和三个问题》一样，让我想到写一篇这样的文章。

几个例子

当我们在面对各种对选择的影响因子的时候，如：城市，公司规模，公司性质，薪水，项目，户口，技术，方向，眼界…… 你总会发现，你会在几个公司中纠结一些东西，举几个例子：

• 某网友和我说，他们去上海腾讯，因为腾讯的规模很大，但却发现薪水待遇没有豆瓣高（低的还不是一点），如果以后要换工作的话，起薪点直接关系到了以后的高工资。我说那就去豆瓣吧，他说豆瓣在北京，污染那么严重，又没有户口，生存环境不好。我说去腾讯吧，他说腾讯最近组织调整，不稳定。我说那就去豆瓣吧，慢公司，发展很稳当。他说，豆瓣的盈利不清楚，而且用Python，自己不喜欢。我说，那就去腾讯吧，……

• 还有一网友和我说，他想回老家，因为老家的人脉关系比较好，能混得好。但又想留在大城市，因为大城市可以开眼界。

• 另一网友和我说，他想进外企，练练英语，开开眼界，但是又怕在外企里当个螺丝钉，想法得不到实施。朋友拉他去创业，觉得创业挺好的，锻炼大，但是朋友做的那个不知道能不能做好。

• 还有一网友在创新工场的某团队和考研之间抉择，不知道去创新工场行不行，觉得那个项目一般，但是感觉那个团队挺有激情的，另一方面觉得自己的学历还不够，读个研应该能找到更好的工作。

• 还有一些朋友问题我应该学什么技术？不应该学什么技术？或是怎么学会学得最快，技术的路径应该是什么？有的说只做后端不做前端，有的说，只做算法研究，不做工程，等等，等等。因为他们觉得人生有限，术业有专攻。

• 等等，等等……

我个人觉得，如果是非计算机科班出生的人不会做选择，不知道怎么走也罢了，但是我们计算机科班出生的人是学过算法的，懂算法的人应该是知道怎么做选择的。

排序算法

你不可能要所有的东西，所以你只能要你最重要的东西，你要知道什么东西最重要，你就需要对你心内的那些欲望和抱负有清楚的认识，不然，你就会在纠结中度过。

所以，在选择中纠结的人有必要参考一下排序算法。

• 首先，你最需要参考的就是“冒泡排序”——这种算法的思路就是每次冒泡出一个最大的数。所以，你有必要问问你自己，面对那些影响你选择的因子，如果你只能要一个的话，你会要哪个？而剩下的都可以放弃。于是，当你把最大的数，一个一个冒泡出来的时候，并用这个决策因子来过滤选项的时候，你就能比较容易地知道知道你应该选什么了。这个算法告诉我们，人的杂念越少，就越容易做出选择。

• 好吧，可能你已茫然到了怎么比较两个决策因子的大小，比如：你分不清楚，工资>业务前景吗？业务前景>能力提升吗？所以你完全没有办法进行冒泡法。那你，你不妨参考一个“快速排序”的思路——这个算法告诉我们，我们一开始并不需要找到最大的数，我们只需要把你价值观中的某个标准拿出来，然后，把可以满足这个价值的放到右边，不能的放到左边去。比如，你的标准是：工资大于5000元&&业务前景长于3年的公司，你可以用这个标准来过滤你的选项。然后，你可以再调整这个标准再继续递归下去。这个算法告诉我们，我们的选择标准越清晰，我们就越容易做出选择。

这是排序算法中最经典的两个算法了，面试必考。相信你已烂熟于心中了。所以，我觉得你把这个算法应用于你的人生选择也应该不是什么问题。关于在于，你是否知道自己想要的是什么？

排序算法的核心思想就是，让你帮助你认清自己最需要的是什么，认清自己最想要的是什么，然后根据这个去做选择。

贪婪算法

所谓贪婪算法，是一种在每一步选择中都采取在当前状态下最好或最优（即最有利）的选择（注意：是当前状态下），从而希望导致结果是最好或最优的算法。贪婪算法最经典的一个例子就是哈夫曼编码。

对于人类来说，一般人在行为处事的时候都会使用到贪婪算法，

• 比如在找零钱的时候，如果要找补36元，我们一般会按这样的顺序找钱：20元，10元，5元，1元。

• 或者我们在过十字路口的时候，要从到对角线的那个街区时，我们也会使用贪婪算法——哪边的绿灯先亮了我们就先过到那边去，然后再转身90度等红灯再过街。

这样的例子有很多。对于选择中，大多数人都会选用贪婪算法，因为这是一个比较简单的算法，未来太复杂了，只能走一步看一步，在当前的状况下做出最利于自己的判断和选择即可。

有的人会贪婪薪水，有的人会贪婪做的项目，有的人会贪婪业务，有的人会贪婪职位，有的人会贪婪自己的兴趣……这些都没什么问题。贪婪算法并没有错，虽然不是全局最优解，但其可以让你找到局部最优解或是次优解。其实，有次优解也不错了。贪婪算法基本上是一种急功近利的算法，但是并不代表这种算法不好，如果贪婪的是一种长远和持续，又未尝不可呢？。

动态规划

但是我们知道，对于大部分的问题，贪婪法通常都不能找出最优解，因为他们一般没有测试所有可能的解。因为贪婪算法是一种短视的行为，只会跟据当前的形式做判断，也就是过早做决定，因而没法达到最佳解。

动态规划和贪婪算法的最大不同是，贪婪算法做出选择，不能在过程优化。动态规划则会保存以前的运算结果，并根据以前的结果对当前进行选择，会动态优化功能。

动态规划算法至少告诉我们两个事：

1）承前启后非常重要，当你准备去做遍历的时候，你的上次的经历不但能开启你以后的经历，而且还能为后面的经历所用。你的每一步都没有浪费。

2）是否可以回退也很重要。这意思是——如果你面前有两个选择，一个是A公司一个是B公司，如果今天你选了A公司，并不是你完全放弃了B公司。而是，你知道从A公司退出来去B公司，会比从B公司退出来去A公司要容易一些。

比如说：你有两个offer，一个是Yahoo，一个是Baidu，上述的第一点会让我们思考，我以前的特长和能力更符合Yahoo还是Baidu？而Yahoo和Baidu谁能给我开启更大的平台？上述的第二点告诉我们，是进入Yahoo后如果没有选好，是否还能再选择Baidu公司？还是进入Baidu公司后能容易回退到Yahoo公司？

Dijkstra最短路径

最短路径是一个Greedy + DP的算法。相当经典。这个算法的大意如下：

1）在初始化的时候，所有的结点都和我是无穷大，默认是达不到的。

2）从离自己最近的结点开始贪婪。

3）走过去，看看又能到达什么样的结点，计算并更新到所有目标点的距离。

4）再贪婪与原点最短的结点，如此反复。

这个算法给我们带来了一些这样的启示：

• 有朋友和我说过他想成为一个架构师，或是某技术领域的专家，并会踏踏实实的向这个目标前进，永不放弃。我还是鼓励了他，但我也告诉他了这个著名的算法，我说，这个算法告诉你，架构师或某领域的专家对你来说目前的距离是无穷大，他们放在心中，先看看你能够得着的东西。所谓踏实，并不是踏踏实实追求你的目标，而是踏踏实实把你够得着看得见的就在身边的东西干好。我还记得我刚参加工作，从老家出来的时候，从来没有想过要成为一个技术牛人，也从来没有想过我的博客会那么的有影响力，在做自己力所能及，看得见摸得着的事情，我就看见什么技术就学什么，学着学着就知道怎么学更轻松，怎么学更扎实，这也许就是我的最短路径。

• 有很多朋友问我要不要学C++，或是问我学Python还是学Ruby，是不是不用学前端，等等。这些朋友告诉我，他们不可能学习多个语言，学了不用也就忘了，而且术业有专攻。这并没有什么不对的，只是我个人觉得，学习一个东西没有必要只有两种状态，一种是不学，另一种是精通。了解一个技术其实花不了多少时间，我学C++的目的其实是为了更懂Java，学TCP/IP协议其实是为了更懂Socket编程，很多东西都是连通和相辅相成的，学好了C/C++/Unix/TCP等这些基础技术后，我发现到达别的技术路径一下缩短了（这就是为什么我用两天时间就可以了解Go语言的原因）。这就好像这个算法一样，算法效率不高，也许达到你的目标，你在一开始花了很长时间，遍历了很多地方，但是，这也许这就是你的最短路径（比起你达不到要好得多）。

算法就是Trade-Off

你根本没有办法能得到所有你想得到的东西，任何的选择都意味着放弃——当你要去获得一个东西的时候，你总是需要放弃一些东西。人生本来就是一个跷跷板，一头上，另一头必然下。这和我们做软件设计或算法设计一样，用时间换空间，用空间换时间，还有CAP理论，总是有很多的Trade-Off，正如这个短语的原意一样——你总是要用某种东西去交易某种东西。

我们都在用某种东西在交易我们的未来，有的人用自己的努力，有的人用自己的思考，有的人用自己的年轻，有的人用自己的自由，有的人用自己的价值观，有的人用自己的道德…… …… 有的人在交换金钱，有的人在交换眼界，有的人在交换经历，有的人在交换地位，有的人在交换能力，有的人在交换自由，有的人在交换兴趣，有的人在交换虚荣心，在交换安逸享乐…… ……

每个人有每个人的算法，每个算法都有每个算法的purpose，就算大家在用同样的算法，但是每个人算法中的那些变量、开关和条件都不一样，得到的结果也不一样。我们就是生活在Matrix里的一段程序，我们每个人的算法决定着我们每个人的选择，我们的选择决定了我们的人生。

2012年就要过去了，祝大家新年快乐！

（全文完）

我们有一个程序，叫ShuffleArray()，是用来洗牌的，我见过N多千变万化的ShuffleArray()，但是似乎从来没人去想过怎么去测试这个算法。所以，我在面试中我经常会问应聘者如何测试ShuffleArray()，没想到这个问题居然难倒了很多有多年编程经验的人。对于这类的问题，其实，测试程序可能比算法更难写，代码更多。而这个问题正好可以加强一下我在《我们需要专职的QA吗？》中我所推崇的——开发人员更适合做测试的观点。

我们先来看几个算法（第一个用递归二分随机抽牌，第二个比较偷机取巧，第三个比较通俗易懂）

递归二分随机抽牌

有一次是有一个朋友做了一个网页版的扑克游戏，他用到的算法就是想模拟平时我们玩牌时用手洗牌的方式，是用递归+二分法，我说这个程序恐怕不对吧。他觉得挺对的，说测试了没有问题。他的程序大致如下（原来的是用Javascript写的，我在这里凭记忆用C复现一下）：

//递归二分方法
const size_t MAXLEN = 10;
const char TestArr[MAXLEN] = {'A','B','C','D','E','F','G','H','I','J'};

static char RecurArr[MAXLEN]={0};
static int cnt = 0;
void ShuffleArray_Recursive_Tmp(char* arr, int len)
{
    if(cnt > MAXLEN || len <=0){
        return;
    }

    int pos = rand() % len;
    RecurArr[cnt++] = arr[pos];
    if (len==1) return;
    ShuffleArray_Recursive_Tmp(arr, pos);
    ShuffleArray_Recursive_Tmp(arr+pos+1, len-pos-1);
}

void ShuffleArray_Recursive(char* arr, int len)
{
    memset(RecurArr, 0, sizeof(RecurArr));
    cnt=0;
    ShuffleArray_Recursive_Tmp(arr, len);
    memcpy(arr, RecurArr, len);
}

void main()
{
    char temp[MAXLEN]={0};
    for(int i=0; i<5; i++) {
        strncpy(temp, TestArr, MAXLEN);
        ShuffleArray_Recursive((char*)temp, MAXLEN);
    }
}

随便测试几次，还真像那么回事：

第一次：D C A B H E G F I J
第二次：A G D B C E F J H I
第三次：A B H F C E D G I J
第四次：J I F B A D C E H G
第五次：F B A D C E H G I J

快排Hack法

让我们再看一个hack 快排的洗牌程序（只看算法，省去别的代码）：

int compare( const void *a, const void *b )
{
    return rand()%3-1;
}

void ShuffleArray_Sort(char* arr, int len)
{
    qsort( (void *)arr, (size_t)len, sizeof(char), compare );
}

运行个几次，感觉得还像那么回事：

第一次：H C D J F E A G B I
第二次：B F J D C E I H G A
第三次：C G D E J F B I A H
第四次：H C B J D F G E I A
第五次：D B C F E A I H G J

看不出有什么破绽。

大多数人的实现

下面这个算法是大多数人的实现，就是for循环一次，然后随机交换两个数

void ShuffleArray_General(char* arr, int len)
{
    const int suff_time = len;
    for(int idx=0; idx<suff_time; idx++) {
        int i = rand() % len;
        int j = rand() % len;
        char temp = arr[i];
        arr[i] = arr[j];
        arr[j] = temp;
    }
}

跑起来也还不错，洗得挺好的。

第一次：G F C D A J B I H E
第二次：D G J F E I A H C B
第三次：C J E F A D G B H I
第四次：H D C F A E B J I G
第五次：E A J F B I H G D C

但是上述三个算法哪个的效果更好？好像都是对的。一般的QA或是程序员很有可能就这样把这个功能Pass了。但是事情并没有那么简单……

如何测试

在做测试之前，我们还需要了解一下一个基本知识——PC机上是做不出真随机数的，只能做出伪随机数。真随机数需要硬件支持。但是不是这样我们就无法测试了呢，不是的。我们依然可以测试。

我们知道，洗牌洗得好不好，主要是看是不是够随机。那么如何测试随机性呢？

试想，我们有个随机函数rand()返回1到10中的一个数，如果够随机的话，每个数返回的概率都应该是一样的，也就是说每个数都应该有10分之1的概率会被返回。

一到概率问题，我们只有一个方法来做测试，那就是用统计的方式。也就是说，你调用rand()函数100次，其中，每个数出现的次数大约都在10次左右。（注意：我用了左右，这说明概率并不是很准确的）不应该有一个数出现了15次以上，另一个在5次以下，要是这样的话，这个函数就是错的。

举一反三，测试洗牌程序也一样，需要通过概率的方式来做统计，是不是每张牌出现在第一个位置的次数都是差不多的。

于是，这样一来上面的程序就可以很容易做测试了。

下面是测试结果（测试样本1000次——列是每个位置出现的次数，行是各个字符的统计，出现概率应该是1/10，也就是100次）：

递归随机抽牌的方法

很明显，这个洗牌程序太有问题。算法是错的！

     1    2    3    4    5    6    7    8    9    10
----------------------------------------------------
A | 101  283  317  208   65   23    3    0    0    0
B | 101  191  273  239  127   54   12    2    1    0
C | 103  167  141  204  229  115   32    7    2    0
D | 103  103   87  128  242  195  112   26    3    1
E | 104   83   62   67  116  222  228   93   22    3
F |  91   58   34   60   69  141  234  241   65    7
G |  93   43   35   19   44  102  174  274  185   31
H |  94   28   27   27   46   68   94  173  310  133
I | 119   27   11   30   28   49   64   96  262  314
J |  91   17   13   18   34   31   47   88  150  511

快排Hack法

看看对角线（从左上到右下）上的数据，很离谱！所以，这个算法也是错的。

      1    2    3    4    5    6    7    8    9    10
-----------------------------------------------------
A |   74  108  123  102   93  198   40   37   52  173
B |  261  170  114   70   49   28   37   76  116   79
C |  112  164  168  117   71   37   62   96  116   57
D |   93   91  119  221  103   66   91   98   78   40
E |   62   60   82   90  290  112   95   98   71   40
F |   46   60   63   76   81  318   56   42   70  188
G |   72   57   68   77   83   39  400  105   55   44
H |   99   79   70   73   87   34  124  317   78   39
I |  127  112  102   90   81   24   57   83  248   76
J |   54   99   91   84   62  144   38   48  116  264

大多数人的算法

我们再来看看大多数人的算法。还是对角线上的数据有问题，所以，还是错的。

      1    2    3    4    5    6    7    8    9    10
-----------------------------------------------------
A |  178   98   92   82  101   85   79  105   87   93
B |   88  205   90   94   77   84   93   86  106   77
C |   93   99  185   96   83   87   98   88   82   89
D |  105   85   89  190   92   94  105   73   80   87
E |   97   74   85   88  204   91   80   90  100   91
F |   85   84   90   91   96  178   90   91  105   90
G |   81   84   84  104  102  105  197   75   79   89
H |   84   99  107   86   82   78   92  205   79   88
I |  102   72   88   94   87  103   94   92  187   81
J |   87  100   90   75   76   95   72   95   95  215

正确的算法

下面，我们来看看性能高且正确的算法—— Fisher_Yates算法

void ShuffleArray_Fisher_Yates(char* arr, int len)
{
    int i = len, j;
    char temp;

    if ( i == 0 ) return;
    while ( --i ) {
        j = rand() % (i+1);
        temp = arr[i];
        arr[i] = arr[j];
        arr[j] = temp;
    }
}

这个算法不难理解，看看测试效果（效果明显比前面的要好）：

      1    2    3    4    5    6    7    8    9    10
-----------------------------------------------------
A |  107   98   83  115   89  103  105   99   94  107
B |   91  106   90  102   88  100  102   97  112  112
C |  100  107   99  108  101   99   86   99  101  100
D |   96   85  108  101  117  103  102   96  108   84
E |  106   89  102   86   88  107  114  109  100   99
F |  109   96   87   94   98  102  109  101   92  102
G |   94   95  119  110   97  112   89  101   89   94
H |   93  102  102  103  100   89  107  105  101   98
I |   99  110  111  101  102   79  103   89  104  102
J |  105  112   99   99  108  106   95   95   99   82

但是我们可以看到还是不完美。因为我们使用的rand()是伪随机数，不过已经很不错的。最大的误差在20%左右。

我们再来看看洗牌100万次的统计值，你会看到误差在6%以内了。这个对于伪随机数生成的程序已经很不错了。

      1       2     3       4      5      6      7      8     9      10
-------------------------------------------------------------------------
A | 100095  99939 100451  99647  99321 100189 100284  99565 100525  99984
B |  99659 100394  99699 100436  99989 100401  99502 100125 100082  99713
C |  99938  99978 100384 100413 100045  99866  99945 100025  99388 100018
D |  99972  99954  99751 100112 100503  99461  99932  99881 100223 100211
E | 100041 100086  99966  99441 100401  99958  99997 100159  99884 100067
F | 100491 100294 100164 100321  99902  99819  99449 100130  99623  99807
G |  99822  99636  99924 100172  99738 100567 100427  99871 100125  99718
H |  99445 100328  99720  99922 100075  99804 100127  99851 100526 100202
I | 100269 100001  99542  99835 100070  99894 100229 100181  99718 100261
J | 100268  99390 100399  99701  99956 100041 100108 100212  99906 100019

如何写测试案例

测试程序其实很容易写了。就是，设置一个样本大小，做一下统计，然后计算一下误差值是否在可以容忍的范围内。比如：

• 样本：100万次
• 最大误差：10%以内
• 平均误差：5%以内 （或者：90%以上的误差要小于5%）

注意

其实，以上的测试只是测试了牌在各个位置的概率。这个还不足够好。因为还可能会现在有Patten的情况。如：每次洗牌出来的都是一个循环顺序数组。这完全可以满足我上面的测试条件。但是那明显是错的。所以，还需要统计每种排列的出现的次数，看看是不是均匀。但是，如果这些排列又是以某种规律出现的呢？看来，这没完没了了。

测试的确是一个很重要，并不简单的事情。谢谢所有参与讨论的人。

附录

之前忘贴了一个模拟我们玩牌洗牌的算法，现补充如下：

void ShuffleArray_Manual(char* arr, int len)
{
    int mid = len / 2;

    for (int n=0; n<5; n++){

        //两手洗牌
        for (int i=1; i<mid; i+=2){
            char tmp = arr[i];
            arr[i] = arr[mid+i];
            arr[mid+i] = tmp;
        }

        //随机切牌
        char *buf = (char*)malloc(sizeof(char)*len);

        for(int j=0; j<5; j++) {
            int start= rand() % (len-1) + 1;
            int numCards= rand()% (len/2) + 1;

            if (start + numCards > len ){
                numCards = len - start;
            }

            memset(buf, 0, len);
            strncpy(buf, arr, start);
            strncpy(arr, arr+start, numCards);
            strncpy(arr+numCards, buf, start);
        }
        free(buf);

    }
}

我们来看看测试结果：（10万次）效果更好一些，误差在2%以内了。

      1       2     3       4      5      6      7      8     9      10
-------------------------------------------------------------------------
A |  10002   9998   9924  10006  10048  10200   9939   9812  10080   9991
B |   9939   9962  10118  10007   9974  10037  10149  10052   9761  10001
C |  10054  10100  10050   9961   9856   9996   9853  10016   9928  10186
D |   9851   9939   9852  10076  10208  10003   9974  10052   9992  10053
E |  10009   9915  10050  10037   9923  10094  10078  10059   9880   9955
F |  10151  10115  10113   9919   9844   9896   9891   9904  10225   9942
G |  10001  10116  10097  10030  10061   9993   9891   9922   9889  10000
H |  10075  10033   9866   9857  10170   9854  10062  10078  10056   9949
I |  10045   9864   9879  10066   9930   9919  10085  10104  10095  10013
J |   9873   9958  10051  10041   9986  10008  10078  10001  10094   9910

（全文完）

信息检索概述

信息检索是当前应用十分广泛的一种技术，论文检索、搜索引擎都属于信息检索的范畴。通常，人们把信息检索问题抽象为：在文档集合D上，对于由关键词w[1] … w[k]组成的查询串q，返回一个按查询q和文档d匹配度relevance(q, d)排序的相关文档列表D’。

对于这一问题，先后出现了布尔模型、向量模型等各种经典的信息检索模型，它们从不同的角度提出了自己的一套解决方案。布尔模型以集合的布尔运算为基础，查询效率高，但模型过于简单，无法有效地对不同文档进行排序，查询效果不佳。向量模型把文档和查询串都视为词所构成的多维向量，而文档与查询的相关性即对应于向量间的夹角。不过，由于通常词的数量巨大，向量维度非常高，而大量的维度都是0，计算向量夹角的效果并不好。另外，庞大的计算量也使得向量模型几乎不具有在互联网搜索引擎这样海量数据集上实施的可行性。

tf-idf模型

目前，真正在搜索引擎等实际应用中广泛使用的是tf-idf模型。tf-idf模型的主要思想是：如果词w在一篇文档d中出现的频率高，并且在其他文档中很少出现，则认为词w具有很好的区分能力，适合用来把文章d和其他文章区分开来。该模型主要包含了两个因素：

1) 词w在文档d中的词频tf (Term Frequency)，即词w在文档d中出现次数count(w, d)和文档d中总词数size(d)的比值：

tf(w,d) = count(w, d) / size(d) 

2) 词w在整个文档集合中的逆向文档频率idf (Inverse Document Frequency)，即文档总数n与词w所出现文件数docs(w, D)比值的对数:

idf = log(n / docs(w, D)) 

tf-idf模型根据tf和idf为每一个文档d和由关键词w[1]…w[k]组成的查询串q计算一个权值，用于表示查询串q与文档d的匹配度：

tf-idf(q, d) 
= sum { i = 1..k | tf-idf(w[i], d) } 
= sum { i = 1..k | tf(w[i], d) * idf(w[i]) } 

信息检索问题的概率视角

直观上看，tf描述的是文档中词出现的频率；而idf是和词出现文档数相关的权重。我们比较容易定性地理解tf-idf的基本思想，但具体到tf-idf的一些细节却并不是那么容易说清楚为什么。比如：

1) 为什么tf是count(w, d) / size(d)？能不能是log(count(w, d) / size(d))等其他形式？

2) 为什么idf是一个log形式？

3) 为什么tf和idf之间是乘积关系，而不是加法或指数关系？

4) 为什么多个关键词的tf-idf值是加法关系，而不是乘法或者指数关系？

5) 除了tf-idf值，Google还会计算网页的PageRank值，二者相乘得到最后的权值，为什么是乘法，而不是加法或指数？

据说，最初甚至tf-idf的提出者自己也没有对诸如“为什么idf是log形式”这个问题给出有力的解释，虽然后来有人从信息论的角度对idf的log形式给出了令人信服的解释，但是剩下的其他一些疑问仍然存在。在我了解的范围内，对于tf-idf模型还没有一个真正统一完整的理论解释。在试图为tf-idf找到更好的理论解释的过程中，我意识到对tf-idf模型种种疑问的根源在于tf-idf试图表达的“查询q和文档的匹配度”本身就有一定的模糊性，什么叫做“匹配度”，这就有很大的自由发挥空间。如果说向量模型的用向量夹角来表示匹配度概念还有一定的理论基础，那么用tf-idf来表达匹配度就有点“与其说是科学，不如说是艺术”的味道。

更进一步，其实，信息检索问题的抽象方式“在文档集合D上，对于给定查询串q，返回一个按查询q和文档d匹配度relevance(q, d)排序的相关文档列表D’”本身是值得反思的。我们应当考虑抛弃“匹配度”这种模糊的目标，从根源上寻求一种具有明确数学意义的目标。如果我们从概率视角来看，把“查询串q和文档d的匹配度”问题转换为“当查询串是q时，用户期望获得文档d的概率”问题，信息检索问题就清晰多了。一方面这个概率描述是站在人的角度来看待信息检索问题的，更加贴近实际的用户体验；另一方面，概率本身是有明确数学意义的，这样我们就首先从目标上对问题进行了严格化。

下面，我将通过一个模型，从概率的视角，一边解释tf-idf的概率意义，一边指出其不合理之处。

盒子小球模型

为了分析“当查询串是q时，用户期望获得文档d的概率”问题，我首先建立了一种称为“盒子小球模型”的简化模型。盒子小球模型把词想象成各种不同颜色的小球，文档想象成装有若干小球的盒子，把“当查询串是q时，用户期望获得文档d的概率“转换为下面的问题：

有n个盒子d[1], d[2], … d[n]，每个盒子中有若干不同颜色的小球，有人随机地选择了一个盒子，并从盒子中随机地拿出了一个颜色为w[j]的小球，那么这个小球来自于盒子d[i]的概率是多少？

其实，这就是经典的条件概率问题P(d[i] | w[j])，采用贝叶斯推断将其转化为：

P(d[i] | w[j]) 
= P(d[i], w[j]) / P(w[j]) 
= P(d[i]) * P(w[j] | d[i]) / P(w[j]) 

我们注意到这个条件概率包括几个部分，P(d[i])是盒子d[i]被选中的先验概率，p(w[j])是w[j]颜色小球被选中的先验概率，P(w[j] | d[i])是在盒子d[i]中选中颜色w[j]小球的条件概率。

文档先验概率P(d)与PageRank

首先，我们来看盒子d[i]被选中的先验概率P(d[i])是什么。P(d[i])的意义是：当用户什么也没有输入的时候，它可能对文档d[i]感兴趣的概率。在没有更多信息的情况下，我们可以认为每个盒子被选中的先验概率P(d[i])是相等的，都等于1 / m，其中m表示总文档数（总盒子数），这时P(d[i])作为公共系数可被忽略。不过，在实际应用中，我们通常可以根据其他知识获得各文档的先验概率，比如，学术文献和网页通常可以基于引用度模型计算其先验概率，这些经典论文和热门网页是多数人乐于见到的。说到这里，你可能已经发现，Google PageRank本质上就是这个先验概率P(d[i])乘以某个系数！所以，PageRank实际上也被纳入这个条件概率模型中来了，这就不难解释为什么在Google的排序算法中PageRank权重和tf-idf权重是一种乘积关系而不是加或者指数关系。另一方面，在理解了文档先验概率对整个搜索结果概率的影响后，当搜索引擎中针对PageRank出现各种假链接SEO时，我们可以不拘泥于基于链接引用模型的PageRank，只要是以网页先验概率为目标，不论是采用基于链接引用的PageRank，还是基于搜索结果点击数模型，或是其他模型，都是可以的。这就是“变通”，从原理上“通”了，就可以在方法上“变”。

词的先验概率P(w)

下面我们来考察词w[j]的先验概率P(w[j])。P(w[j])的意义是：在整个文档集合中，w[j]被作为搜索关键词的概率，比如：“iPhone 5”，“青花瓷”这类词被用作搜索关键词的概率较高，而“的”，“什么”，“我们”这类高频词不大可能成为搜索关键词。那么，我们如何来定量计算P(w[j])呢？一种思路就是把w[j]在文档集中出现的频率作为其先验概率。不过，显然存在更好的方案：在大量的搜索查询中进行统计，统计方法得出P(w[j])的方法很接近P(w[j])本质的，不需要引入额外的假设。比如，一段时间内某搜索引擎的搜索总次数为10^10次，“公积金”这个词出现了100次，那么，我们可以认为先验概率P(“公积金”)就是100 / 10^10 = 10^-8。

词代表文档主题的条件概率P(w | d)

最后，我们来看条件概率P(w[j] | d[i])。P(w[j] | d[i])的意义是在文档d[i]中，人们用关键词w[j]来搜索它的概率。那么，什么样的词是人们会用来搜索一篇文档的呢？多数情况下，是那些代表一篇文档主题的词。比如，有一篇新闻是关于iPhone 5发布会的，那么“iPhone5”， “发布会”，“库克”，“苹果”这些词基本上就构成了文章的主题；那么，反过来说，如果用户想搜索这篇关于iPhone 5发布会的新闻，他就有很大的可能通过这几个词来进行搜索。我们应当注意分辨P(w[j] | d[i])与P(w[j])的区别，后者可以通过大量的查询统计得来，而前者不能与后者直接划等号，因为前者的意义是w[j]代表d[i]主题的概率。如果非要引入统计方法，那么P(w[j] | d[i])对应的统计是：当搜索关键词是w[j]且搜索结果包含d[i]时，用户点击（满意）d[i]作为搜索结果的频率。比如，用“iPhone5 发布会”的搜索，在结果中有都10000次出现了网页x，其中，用户8000次点击了网页x，那么，可以认为有80%的概率网页x的主题是关于“iPhone5 发布会”的。

词的信息量和idf

上面谈到了对P(w[j] | d[i])的计算的统计方法，但该方法有一定的局限，比如，要能进行统计首先需要文档出现在足够多的搜索结果中，需要时间和量的积累。除了统计方法外，我们可以考虑其他方法计算词w[j]代表文档d[i]主题的概率。可能有人立刻会想到要对文章进行语义分析提取关键词，给这些关键词高权重，给其他词低权重。这种想法有一定的合理性，但实现上涉及语义分析，没有成熟高效的方法。实际上，信息论为我们提供了另一条高效方案。上面谈到“的”，“什么”，“我们”这类高频词不会成为文档主题和搜索关键词的原因是它们不能提供足够的信息，而“iPhone 5”，“发布会”这样的词汇则信息量丰富。所谓信息是指对不确定性（熵）的减小程度，信息的单位是比特(bit)，信息量越大对于不确定性的减小程度越大。比如，外面可能在下雨也可能没有下雨，可能性空间大小为2，如果我们看一眼窗外，可能性空间就变成了1，那么“看见窗外在下雨”所提供的信息量就和熵的减小程度成正比，具体来讲等于log(2/1)=1。如果要用二进制编码是否下雨，需要1个bit，0代表没有下雨，1代表下雨。

但在很多场景下，各个可能性的概率并不相同，比如：欧洲杯16只球队都可能夺冠，赛前它们夺冠的先验概率并不相同，那么结果的不确定性程度实际上是小于log(16)=4。如果你没有看比赛，有人告诉你西班牙夺冠了，你可能会觉得很正常，但如果有人告诉你瑞士夺冠了，你通常会非常惊讶。这一现象的理论解释是，如果赛前西班牙夺冠概率是1/4，而瑞士夺冠概率是1/32，那么，“西班牙夺冠”的信息量为log(4)=2，即把不确定性减小为原来的1/4，而“瑞士夺冠”的信息量为log(32)=5，不确定性减小为原来的1/32，一下子接受比前者大了两倍以上的信息量，当然你会吃惊。

回到信息检索，比如，“2012美国大选”这个查询串包含了“2012”，“美国”和“大选”3个关键词，我们应该如何定量计算它们的信息量呢？根据信息的定义，词的信息量等于它对不确定性的缩小程度。如果文档总数为2^30，其中2^14篇文档出现了“美国”，那么“美国”这个词就把文档的不确定性从2^30缩小为2^14，它所包含的信息量为log(2^30/2^14)=16；而只有2^10篇文档出现了“大选”，那么大选的信息量就是log(2^30/2^10)=20，比“美国”多了4个bit。而“的”，“什么”，“我们”这些高频词对减小文档不确定性几乎没有帮助，因而信息量为0。相信你已经发现，上面idf(w)公式中的log(n / docs(w, D))实际上就是词w的信息量了。

如果我们考虑词的信息量对条件概率P(w[j] | d[i])的影响，假设“词w在文档中被选中的概率与其在文档中的出现频率和其信息量的乘积成正比”，那么上面的条件概率模型就变成：

P(d[i] | w[j]) 
= P(d[i], w[j]) / P(w[j]) 
= P(d[i]) * P(w[j] | d[i]) / P(w[j]) 
= P(d[i]) * (tf(w[j], d[i]) * idf(w[j] / sum { k = 1..size(d[i]), tf(w[k], d[i]) * idf(w[k]) }) / p(w[j]) 
= P(d[i]) * (tf-idf(w[j], d[i]) / sum { k = 1..size(d[i]), tf-idf(w[k], d[i]) }) / p(w[j]) 
= P(d[i]) * (tf-idf(w[j], d[i]) / tf-idf(d[i])) / p(w[j]) 

我们看到tf-idf已经被纳入框架内了，但是还多出文档先验概率P(d[i])，关键词先验概率P(w[j])和文档各词的总tf-idf(d[i])。普通搜索引擎是基于PageRank和tf-idf的，那么，根据这个概率模型，我们可以看出，它没有考虑文档总tf-idf(d[i])和关键词先验概率p(w[j])。如果考虑这两个因素，相信搜索效果会更好。

多关键词

上面的条件概率模型主要是针对单个关键词的情况，下面我们进一步将其扩展到多关键词情况。我们知道，在tf-idf中，多个关键词的所产生的tf-idf值是一种叠加关系，那么这是否符合条件概率模型呢？答案是否定的。在两个关键字情况下，条件概率问题转化为“如果有人从一个盒子中同时摸出颜色w[x]的小球和颜色w[y]的小球，这两个小球来自于盒子d[i]的概率是多少？”。假设从盒子中摸出各个小球事件是相互独立的情况下，即

P(w[x], w[y]) 
= P(w[x]) * P(w[y]) P(w[x], w[y] | d[i]) 
= P(w[x] | d[i]) * P(w[y] | d[i]) 

我们可以推导出条件概率：

P(d[i] | w[x], w[y]) 
= P(d[i], w[x], w[y]) / P(w[x], w[y]) 
= P(d[i]) * P(w[x], w[y] | d[i]) / P(w[x], w[y]) 
= P(d[i]) * P(w[x] | d[i]) * P(w[y] | d[i]) / (P(w[x] * P(w[y])) 
= P(d[i]) * (tf-idf(w[x], d[i]) / tf-idf(d[i])) * ((tf-idf(w[y], d[i]) / tf-idf(d[i]))) / (p(w[x]) * P(w[y])) 

可见，概率模型所得出的各个关键词的tf-idf值之间是乘积关系，这是与tf-idf模型的加法关系是不同的。这一点可能与二者是否要求“文档必须包含所有查询关键词”的基本假设有关系。在文档不包含所有关键字的这种情况下，tf-idf模型可能得出一个非0的匹配度，但条件概率模型得出的概率肯定为0。不过，如果考虑一般查询关键词数量不多（3个以内），而大量文档都同时包含这些关键词，概率模型的乘积关系是比tf-idf模型的加法关系更有理论基础。从根本上讲，这是因为tf-idf的“匹配度”是一个模棱两可的概念，而条件概率有坚实的理论基础。

总结

TF-IDF模型是搜索引擎中广泛使用的信息检索模型，但对于TF-IDF模型一直存在各种疑问。本文为信息检索问题一种基于条件概率的盒子小球模型，其核心思想是把“查询串q和文档d的匹配度问题”转化为“查询串q来自于文档d的条件概率问题”。它从概率的视角为信息检索问题定义了比TF-IDF模型所表达的匹配度更为清晰的目标。从概率模型中，我们看到查询串q来自于文档d的条件概率主要包含以下几个因素：1) 文档的先验概率P(d[i])，这与PageRank对应；2) 词w被作为搜索关键词的先验概率P(w)，这可以通过统计方法获得；3) 关键词w代表文档d主题，或以词w搜索文档d的概率，P(w | d)，除了统计方法，这可以通过tf-idf来计算。

————注：本文于2019年11月4日更新————

关于无锁队列的实现，网上有很多文章，虽然本文可能和那些文章有所重复，但是我还是想以我自己的方式把这些文章中的重要的知识点串起来和大家讲一讲这个技术。下面开始正文。

关于CAS等原子操作

在开始说无锁队列之前，我们需要知道一个很重要的技术就是CAS操作——Compare & Set，或是 Compare & Swap，现在几乎所有的CPU指令都支持CAS的原子操作，X86下对应的是 CMPXCHG 汇编指令。有了这个原子操作，我们就可以用其来实现各种无锁（lock free）的数据结构。

这个操作用C语言来描述就是下面这个样子：（代码来自Wikipedia的Compare And Swap词条）意思就是说，看一看内存*reg里的值是不是oldval，如果是的话，则对其赋值newval。

int compare_and_swap (int* reg, int oldval, int newval)
{
  int old_reg_val = *reg;
  if (old_reg_val == oldval) {
     *reg = newval;
  }
  return old_reg_val;
}

我们可以看到，old_reg_val 总是返回，于是，我们可以在 compare_and_swap 操作之后对其进行测试，以查看它是否与 oldval相匹配，因为它可能有所不同，这意味着另一个并发线程已成功地竞争到 compare_and_swap 并成功将 reg 值从 oldval 更改为别的值了。

这个操作可以变种为返回bool值的形式（返回 bool值的好处在于，可以调用者知道有没有更新成功）：

bool compare_and_swap (int *addr, int oldval, int newval)
{
  if ( *addr != oldval ) {
      return false;
  }
  *addr = newval;
  return true;
}

与CAS相似的还有下面的原子操作：（这些东西大家自己看Wikipedia，也没什么复杂的）

• Fetch And Add，一般用来对变量做 +1 的原子操作
• Test-and-set，写值到某个内存位置并传回其旧值。汇编指令BST
• Test and Test-and-set，用来低低Test-and-Set的资源争夺情况

注：在实际的C/C++程序中，CAS的各种实现版本如下：

1）GCC的CAS

GCC4.1+版本中支持CAS的原子操作（完整的原子操作可参看 GCC Atomic Builtins）

bool __sync_bool_compare_and_swap (type *ptr, type oldval type newval, ...)
type __sync_val_compare_and_swap (type *ptr, type oldval type newval, ...)

2）Windows的CAS

在Windows下，你可以使用下面的Windows API来完成CAS：（完整的Windows原子操作可参看MSDN的InterLocked Functions）

 InterlockedCompareExchange ( __inout LONG volatile *Target,
                                 __in LONG Exchange,
                                 __in LONG Comperand);

3) C++11中的CAS

C++11中的STL中的atomic类的函数可以让你跨平台。（完整的C++11的原子操作可参看 Atomic Operation Library）

template< class T >
bool atomic_compare_exchange_weak( std::atomic* obj,
                                   T* expected, T desired );
template< class T >
bool atomic_compare_exchange_weak( volatile std::atomic* obj,
                                   T* expected, T desired );

无锁队列的链表实现

下面的代码主要参考于两篇论文：

• John D. Valois 1994年10月在拉斯维加斯的并行和分布系统系统国际大会上的一篇论文——《Implementing Lock-Free Queues》
• 美国纽约罗切斯特大学 Maged M. Michael 和 Michael L. Scott 在1996年3月发表的一篇论文 《Simple, Fast, and Practical Non-Blocking and Blocking ConcurrentQueue Algorithms》

（注：下面的代码并不完全与这篇论文相同）

初始化一个队列的代码很简，初始化一个dummy结点（注：在链表操作中，使用一个dummy结点，可以少掉很多边界条件的判断），如下所示：

InitQueue(Q)
{
    node = new node()
    node->next = NULL;
    Q->head = Q->tail = node;
}

我们先来看一下进队列用CAS实现的方式，基本上来说就是链表的两步操作：

1. 第一步，把tail指针的next指向要加入的结点。 tail->next = p;
2. 第二步，把tail指针移到队尾。 tail = p;

EnQueue(Q, data) //进队列
{
    //准备新加入的结点数据
    n = new node();
    n->value = data;
    n->next = NULL;

    do {
        p = Q->tail; //取链表尾指针的快照
    } while( CAS(p->next, NULL, n) != TRUE); 
    //while条件注释：如果没有把结点链在尾指针上，再试

    CAS(Q->tail, p, n); //置尾结点 tail = n;
}

我们可以看到，程序中的那个 do-while 的 Retry-Loop 中的 CAS 操作：如果 p->next 是 NULL，那么，把新结点 n 加到队尾。如果不成功，则重新再来一次！

就是说，很有可能我在准备在队列尾加入结点时，别的线程已经加成功了，于是tail指针就变了，于是我的CAS返回了false，于是程序再试，直到试成功为止。这个很像我们的抢电话热线的不停重播的情况。

但是你会看到，为什么我们的“置尾结点”的操作（第13行）不判断是否成功，因为：

1. 如果有一个线程T1，它的while中的CAS如果成功的话，那么其它所有的 随后线程的CAS都会失败，然后就会再循环，
2. 此时，如果T1 线程还没有更新tail指针，其它的线程继续失败，因为tail->next不是NULL了。
3. 直到T1线程更新完 tail 指针，于是其它的线程中的某个线程就可以得到新的 tail 指针，继续往下走了。
4. 所以，只要线程能从 while 循环中退出来，意味着，它已经“独占”了，tail 指针必然可以被更新。

这里有一个潜在的问题——如果T1线程在用CAS更新tail指针的之前，线程停掉或是挂掉了，那么其它线程就进入死循环了。下面是改良版的EnQueue()

EnQueue(Q, data) //进队列改良版 v1
{
    n = new node();
    n->value = data;
    n->next = NULL;

    p = Q->tail;
    oldp = p
    do {
        while (p->next != NULL)
            p = p->next;
    } while( CAS(p.next, NULL, n) != TRUE); //如果没有把结点链在尾上，再试

    CAS(Q->tail, oldp, n); //置尾结点
}

我们让每个线程，自己fetch 指针 p 到链表尾。但是这样的fetch会很影响性能。而且，如果一个线程不断的EnQueue，会导致所有的其它线程都去 fetch 他们的 p 指针到队尾，能不能不要所有的线程都干同一个事？这样可以节省整体的时间？

比如：直接 fetch Q->tail 到队尾？因为，所有的线程都共享着 Q->tail，所以，一旦有人动了它后，相当于其它的线程也跟着动了，于是，我们的代码可以改进成如下的实现：

EnQueue(Q, data) //进队列改良版 v2 
{
    n = new node();
    n->value = data;
    n->next = NULL;

    while(TRUE) {
        //先取一下尾指针和尾指针的next
        tail = Q->tail;
        next = tail->next;

        //如果尾指针已经被移动了，则重新开始
        if ( tail != Q->tail ) continue;

        //如果尾指针的 next 不为NULL，则 fetch 全局尾指针到next
        if ( next != NULL ) {
            CAS(Q->tail, tail, next);
            continue;
        }

        //如果加入结点成功，则退出
        if ( CAS(tail->next, next, n) == TRUE ) break;
    }
    CAS(Q->tail, tail, n); //置尾结点
}

上述的代码还是很清楚的，相信你一定能看懂，而且，这也是 Java 中的 ConcurrentLinkedQueue 的实现逻辑，当然，我上面的这个版本比 Java 的好一点，因为没有 if 嵌套，嘿嘿。

好了，我们解决了EnQueue，我们再来看看DeQueue的代码：（很简单，我就不解释了）

DeQueue(Q) //出队列
{
    do{
        p = Q->head;
        if (p->next == NULL){
            return ERR_EMPTY_QUEUE;
        }
    while( CAS(Q->head, p, p->next) != TRUE );
    return p->next->value;
}

我们可以看到，DeQueue的代码操作的是 head->next，而不是 head 本身。这样考虑是因为一个边界条件，我们需要一个dummy的头指针来解决链表中如果只有一个元素，head 和 tail 都指向同一个结点的问题，这样 EnQueue 和 DeQueue 要互相排斥了。

但是，如果 head 和 tail 都指向同一个结点，这意味着队列为空，应该返回 ERR_EMPTY_QUEUE，但是，在判断 p->next == NULL 时，另外一个EnQueue操作做了一半，此时的 p->next 不为 NULL了，但是 tail 指针还差最后一步，没有更新到新加的结点，这个时候就会出现，在 EnQueue 并没有完成的时候， DeQueue 已经把新增加的结点给取走了，此时，队列为空，但是，head 与 tail 并没有指向同一个结点。如下所示：

虽然，EnQueue的函数会把 tail 指针置对，但是，这种情况可能还是会导致一些并发问题，所以，严谨来说，我们需要避免这种情况。于是，我们需要加入更多的判断条件，还确保这个问题。下面是相关的改进代码：

DeQueue(Q) //出队列，改进版
{
    while(TRUE) {
        //取出头指针，尾指针，和第一个元素的指针
        head = Q->head;
        tail = Q->tail;
        next = head->next;

        // Q->head 指针已移动，重新取 head指针
        if ( head != Q->head ) continue;
        
        // 如果是空队列
        if ( head == tail && next == NULL ) {
            return ERR_EMPTY_QUEUE;
        }
        
        //如果 tail 指针落后了
        if ( head == tail && next == NULL ) {
            CAS(Q->tail, tail, next);
            continue;
        }

        //移动 head 指针成功后，取出数据
        if ( CAS( Q->head, head, next) == TRUE){
            value = next->value;
            break;
        }
    }
    free(head); //释放老的dummy结点
    return value;
}

上面这段代码的逻辑和 Java 的 ConcurrentLinkedQueue 的 poll 方法很一致了。也是《Simple, Fast, and Practical Non-Blocking and Blocking ConcurrentQueue Algorithms》这篇论文中的实现。

CAS的ABA问题

所谓ABA（见维基百科的ABA词条），问题基本是这个样子：

1. 进程P1在共享变量中读到值为A
2. P1被抢占了，进程P2执行
3. P2把共享变量里的值从A改成了B，再改回到A，此时被P1抢占。
4. P1回来看到共享变量里的值没有被改变，于是继续执行。

虽然P1以为变量值没有改变，继续执行了，但是这个会引发一些潜在的问题。ABA问题最容易发生在lock free 的算法中的，CAS首当其冲，因为CAS判断的是指针的值。很明显，值是很容易又变成原样的。

比如上述的DeQueue()函数，因为我们要让head和tail分开，所以我们引入了一个dummy指针给head，当我们做CAS的之前，如果head的那块内存被回收并被重用了，而重用的内存又被EnQueue()进来了，这会有很大的问题。（内存管理中重用内存基本上是一种很常见的行为）

这个例子你可能没有看懂，维基百科上给了一个活生生的例子——

你拿着一个装满钱的手提箱在飞机场，此时过来了一个火辣性感的美女，然后她很暖昧地挑逗着你，并趁你不注意的时候，把用一个一模一样的手提箱和你那装满钱的箱子调了个包，然后就离开了，你看到你的手提箱还在那，于是就提着手提箱去赶飞机去了。

这就是ABA的问题。

解决ABA的问题

维基百科上给了一个解——使用double-CAS（双保险的CAS），例如，在32位系统上，我们要检查64位的内容

1）一次用CAS检查双倍长度的值，前半部是值，后半部分是一个计数器。

2）只有这两个都一样，才算通过检查，要吧赋新的值。并把计数器累加1。

这样一来，ABA发生时，虽然值一样，但是计数器就不一样（但是在32位的系统上，这个计数器会溢出回来又从1开始的，这还是会有ABA的问题）

当然，我们这个队列的问题就是不想让那个内存重用，这样明确的业务问题比较好解决，论文《Implementing Lock-Free Queues》给出一这么一个方法——使用结点内存引用计数refcnt！（论文《Simple, Fast, and Practical Non-Blocking and Blocking ConcurrentQueue Algorithms》中的实现方法也基本上是一样的，用到的是增加一个计数，可以理解为版本号）

）

SafeRead(q)
{
    loop:
        p = q->next;
        if (p == NULL){
            return p;
        }

        Fetch&Add(p->refcnt, 1);

        if (p == q->next){
            return p;
        }else{
            Release(p);
        }
    goto loop;
}

其中的 Fetch&Add和Release分是是加引用计数和减引用计数，都是原子操作，这样就可以阻止内存被回收了。

用数组实现无锁队列

本实现来自论文《Implementing Lock-Free Queues》

使用数组来实现队列是很常见的方法，因为没有内存的分部和释放，一切都会变得简单，实现的思路如下：

1）数组队列应该是一个ring buffer形式的数组（环形数组）

2）数组的元素应该有三个可能的值：HEAD，TAIL，EMPTY（当然，还有实际的数据）

3）数组一开始全部初始化成EMPTY，有两个相邻的元素要初始化成HEAD和TAIL，这代表空队列。

4）EnQueue操作。假设数据x要入队列，定位TAIL的位置，使用double-CAS方法把(TAIL, EMPTY) 更新成 (x, TAIL)。需要注意，如果找不到(TAIL, EMPTY)，则说明队列满了。

5）DeQueue操作。定位HEAD的位置，把(HEAD, x)更新成(EMPTY, HEAD)，并把x返回。同样需要注意，如果x是TAIL，则说明队列为空。

算法的一个关键是——如何定位HEAD或TAIL？

1）我们可以声明两个计数器，一个用来计数EnQueue的次数，一个用来计数DeQueue的次数。

2）这两个计算器使用使用Fetch&ADD来进行原子累加，在EnQueue或DeQueue完成的时候累加就好了。

3）累加后求个模什么的就可以知道TAIL和HEAD的位置了。

如下图所示：

 小结

以上基本上就是所有的无锁队列的技术细节，这些技术都可以用在其它的无锁数据结构上。

1）无锁队列主要是通过CAS、FAA这些原子操作，和Retry-Loop实现。

2）对于Retry-Loop，我个人感觉其实和锁什么什么两样。只是这种“锁”的粒度变小了，主要是“锁”HEAD和TAIL这两个关键资源。而不是整个数据结构。

还有一些和Lock Free的文章你可以去看看：

• Code Project 上的雄文 《Yet another implementation of a lock-free circular array queue》
• Herb Sutter的《Writing Lock-Free Code: A Corrected Queue》– 用C++11的std::atomic模板。
• IBM developerWorks的《设计不使用互斥锁的并发数据结构》

 （全文完）

我再次引用我以前的一个观点——

能解算法题并不意味着这个人就有能力就能在工作中解决问题，你可以想想，小学奥数题可能比这些题更难，但并不意味着那些奥数能手就能解决实际问题。

好了，让我们来看一个示例（这个示例是昨天在微博上的一个讨论），这个题是——“找出无序数组中第2大的数”，几乎所有的人都用了O(n)的算法，我相信对于我们这些应试教育出来的人来说，不用排序用O(n)算法是很正常的事，连我都不由自主地认为O(n)算法是这个题的标准答案。我们太习惯于标准答案了，这是我国教育最悲哀的地方。（广义的洗脑就是让你的意识依赖于某个标准答案，然后通过给你标准答案让你不会思考而控制你）

功能性需求分析

试想，如果我们在实际工作中得到这样一个题 我们会怎么做？我一定会分析这个需求，因为我害怕需求未来会改变，今天你叫我找一个第2大的数，明天你找我找一个第4大的数，后天叫我找一个第100大的数，我不搞死了。需求变化是很正常的事。分析完这个需求后，我会很自然地去写找第K大数的算法——难度一下子就增大了。

很多人会以为找第K大的需求是一种“过早扩展”的思路，不是这样的，我相信我们在实际编码中写过太多这样的程序了，你一定不会设计出这样的函数接口—— Find2ndMaxNum(int* array, int len)，就好像你不会设计出 DestroyBaghdad(); 这样的接口，而是设计一个DestoryCity( City& ); 的接口，而把Baghdad当成参数传进去！所以，你应该是声明一个叫FindKthMaxNum(int* array, int len, int kth)，把2当成参数传进去。这是最基本的编程方法，用数学的话来说，叫代数！最简单的需求分析方法就是把需求翻译成函数名，然后看看是这个接口不是很二？！

（注：不要纠结于FindMaxNum()或FindMinNum()，因为这两个函数名的业务意义很清楚了，不像Find2ndMaxNum()那么二）

非功能性需求分析

性能之类的东西从来都是非功能性需求，对于算法题，我们太喜欢研究算法题的空间和时间复杂度了。我们希望做到空间和时间双丰收，这是算法学术界的风格。所以，习惯于标准答案的我们已经失去思考的能力，只会机械地思考算法之内的性能，而忽略了算法之外的性能。

如果题目是——“从无序数组中找到第K个最大的数”，那么，我们一定会去思考用O(n)的线性算法找出第K个数。事实上，也有线性算法——STL中可以用nth_element求得类似的第n大的数，其利用快速排序的思想，从数组S中随机找出一个元素X，把数组分为两部分Sa和Sb。Sa中的元素大于等于X，Sb中元素小于X。这时有两种情况：1）Sa中元素的个数小于k，则Sb中的第k-|Sa|个元素即为第k大数；2） Sa中元素的个数大于等于k，则返回Sa中的第k大数。时间复杂度近似为O(n)。

搞学术的nuts们到了这一步一定会欢呼胜利！但是他们哪里能想得到性能的需求分析也是来源自业务的！

我们一说性能，基本上是个人都会问，请求量有多大？如果我们的FindKthMaxNum()的请求量是m次，那么你的这个每次都要O(n)复杂度的算法得到的效果就是O(n*m)，这一点，是书呆子式的学院派人永远想不到的。因为应试教育让我们不会从实际思考了。

工程式的解法

根据上面的需求分析，有软件工程经验的人的解法通常会这样：

1）把数组排序，从大到小。

2）于是你要第k大的数，就直接访问 array[k]。

排序只需要一次，O(n*log(n))，然后，接下来的m次对FindKthMaxNum()的调用全是O(1)的，整体复杂度反而成了线性的。

其实，上述的还不是工程式的最好的解法，因为，在业务中，那数组中的数据可能会是会变化的，所以，如果是用数组排序的话，有数据的改动会让我重新排序，这个太耗性能了，如果实际情况中会有很多的插入或删除操作，那么可以考虑使用B+树。

工程式的解法有以下特点：

1）很方便扩展，因为数据排好序了，你还可以方便地支持各种需求，如从第k1大到k2大的数据（那些学院派写出来的代码在拿到这个需求时又开始挠头苦想了）

2）规整的数据会简化整体的算法复杂度，从而整体性能会更好。（公欲善其事，必先利其器）

3）代码变得清晰，易懂，易维护！（学院派的和STL一样的近似O(n)复杂度的算法没人敢动）

争论

你可能会和我有以下争论，

• 如果程序员做这个算法题用排序的方式，他一定不会像你想那么多。是的，你说得对。但是我想说，很多时候，我们直觉地思考，恰恰是正确的路。因为“排序”这个思路符合人类大脑处理问题的方式，而使用学院派的方式是反大脑直觉的。反大脑直觉的，通常意味着晦涩难懂，维护成本上升。

• 就是一道面试题，我就是想测试一下你的算法技能，这也扯太多了。没问题，不过，我们要清楚我们是在招什么人？是一个只会写算法的人，还是一个会做软件的人？这个只有你自己最清楚。

• 这个算法题太容易诱导到学院派的思路了。是的这道“找出第K大的数”，其实可以变换为更为业务一点的题目——“我要和别的商户竞价，我想排在所有竞争对手报价的第K名，请写一个程序，我输入K，和一个商品名，系统告诉我应该订多少价？（商家的所有商品的报价在一数组中）”——业务分析，整体性能，算法，数据结构，增加需求让应聘者重构，这一个问题就全考了。

• 你是不是在说算法不重要，不用学？千万别这样理解我，搞得好像如果面试不面，我就可以不学。算法很重要，算法题能锻炼我们的思维，而且也有很多实际用处。我这篇文章不是让大家不要去学算法，这是完全错误的，我是让大家带着业务问题去使用算法。问你业务问题，一样会问到算法题上来。

小结

看过这上面的分析，我相信你明白我为什么反对纯算法面试题了。原因就是纯算法的面试题根本不能反应一个程序的综合素质！

那么，在面试中，我们应该要考量程序员的那些综合素质呢？我以为有下面这些东西：

1. 会不会做需求分析？怎么理解问题的？
2. 解决问题的思路是什么？想法如何？
3. 会不会对基础的算法和数据结构灵活运用？

另外，我们知道，对于软件开发来说，在工程上，难是的下面是这些挑战：

• 软件的维护成本远远大于软件的开发成本。
• 软件的质量变得越来越重要，所以，测试工作也变得越来越重要。
• 软件的需求总是在变的，软件的需求总是一点一点往上加的。
• 程序中大量的代码都是在处理一些错误的或是不正常的流程。

所以，对于编程能力上，我们应该主要考量程序员的如下能力：

1. 设计是否满足对需求的理解，并可以应对可能出现的需求变化。
2. 程序是否易读，易维护？
3. 重构代码的能力如何？
4. 会不会测试自己写好的程序？

所以，这段时间，我越来越倾向于问应聘者一些有业务意义的题，而且应增加或更改需求来看程序员的重构代码的能力，写完程序后，让应聘者设计测试案例。

比如：解析加减乘除表达式，字符串转数字，洗牌程序，口令生成器，通过ip地址找地点，英汉词典双向检索……

总之，我反对纯算法面试题！

（全文完）

Wikipedia上的KNN词条中有一个比较经典的图如下：

从上图中我们可以看到，图中的有两个类型的样本数据，一类是蓝色的正方形，另一类是红色的三角形。而那个绿色的圆形是我们待分类的数据。

• 如果K=3，那么离绿色点最近的有2个红色三角形和1个蓝色的正方形，这3个点投票，于是绿色的这个待分类点属于红色的三角形。

• 如果K=5，那么离绿色点最近的有2个红色三角形和3个蓝色的正方形，这5个点投票，于是绿色的这个待分类点属于蓝色的正方形。

我们可以看到，机器学习的本质——是基于一种数据统计的方法！那么，这个算法有什么用呢？我们来看几个示例。

产品质量判断

假设我们需要判断纸巾的品质好坏，纸巾的品质好坏可以抽像出两个向量，一个是“酸腐蚀的时间”，一个是“能承受的压强”。如果我们的样本空间如下：（所谓样本空间，又叫Training Data，也就是用于机器学习的数据）

那么，如果 X1 = 3 和 X2 = 7， 这个毛巾的品质是什么呢？这里就可以用到KNN算法来判断了。

假设K=3，K应该是一个奇数，这样可以保证不会有平票，下面是我们计算（3，7）到所有点的距离。（关于那些距离公式，可以参看K-Means算法中的距离公式）

所以，最后的投票，好的有2票，坏的有1票，最终需要测试的（3，7）是合格品。（当然，你还可以使用权重——可以把距离值做为权重，越近的权重越大，这样可能会更准确一些）

注：示例来自这里，K-NearestNeighbors Excel表格下载

预测

假设我们有下面一组数据，假设X是流逝的秒数，Y值是随时间变换的一个数值（你可以想像是股票值）

那么，当时间是6.5秒的时候，Y值会是多少呢？我们可以用KNN算法来预测之。

这里，让我们假设K=2，于是我们可以计算所有X点到6.5的距离，如：X=5.1，距离是 | 6.5 – 5.1 | = 1.4， X = 1.2 那么距离是 | 6.5 – 1.2 | = 5.3 。于是我们得到下面的表：

注意，上图中因为K=2，所以得到X=4 和 X =5.1的点最近，得到的Y的值分别为27和8，在这种情况下，我们可以简单的使用平均值来计算：

于是，最终预测的数值为：17.5

注：示例来自这里，KNN_TimeSeries Excel表格下载

插值，平滑曲线

KNN算法还可以用来做平滑曲线用，这个用法比较另类。假如我们的样本数据如下（和上面的一样）：

要平滑这些点，我们需要在其中插入一些值，比如我们用步长为0.1开始插值，从0到6开始，计算到所有X点的距离（绝对值），下图给出了从0到0.5 的数据：

下图给出了从2.5到3.5插入的11个值，然后计算他们到各个X的距离，假值K=4，那么我们就用最近4个X的Y值，然后求平均值，得到下面的表：

于是可以从0.0, 0.1, 0.2, 0.3 …. 1.1, 1.2, 1.3…..3.1, 3.2…..5.8, 5.9, 6.0 一个大表，跟据K的取值不同，得到下面的图：

注：示例来自这里，KNN_Smoothing Excel表格下载

后记

最后，我想再多说两个事，

1） 一个是机器学习，算法基本上都比较简单，最难的是数学建模，把那些业务中的特性抽象成向量的过程，另一个是选取适合模型的数据样本。这两个事都不是简单的事。算法反而是比较简单的事。

2）对于KNN算法中找到离自己最近的K个点，是一个很经典的算法面试题，需要使用到的数据结构是“最大堆——Max Heap”，一种二叉树。你可以看看相关的算法。

（全文完）

在数据挖掘中， k-Means 算法是一种 cluster analysis 的算法，其主要是来计算数据聚集的算法，主要通过不断地取离种子点最近均值的算法。

问题

K-Means算法主要解决的问题如下图所示。我们可以看到，在图的左边有一些点，我们用肉眼可以看出来有四个点群，但是我们怎么通过计算机程序找出这几个点群来呢？于是就出现了我们的K-Means算法（Wikipedia链接）

算法概要

这个算法其实很简单，如下图所示：

从上图中，我们可以看到，A, B, C, D, E 是五个在图中点。而灰色的点是我们的种子点，也就是我们用来找点群的点。有两个种子点，所以K=2。

然后，K-Means的算法如下：

1. 随机在图中取K（这里K=2）个种子点。
2. 然后对图中的所有点求到这K个种子点的距离，假如点Pi离种子点Si最近，那么Pi属于Si点群。（上图中，我们可以看到A,B属于上面的种子点，C,D,E属于下面中部的种子点）
3. 接下来，我们要移动种子点到属于他的“点群”的中心。（见图上的第三步）
4. 然后重复第2）和第3）步，直到，种子点没有移动（我们可以看到图中的第四步上面的种子点聚合了A,B,C，下面的种子点聚合了D，E）。

这个算法很简单，但是有些细节我要提一下，求距离的公式我不说了，大家有初中毕业水平的人都应该知道怎么算的。我重点想说一下“求点群中心的算法”

求点群中心的算法

一般来说，求点群中心点的算法你可以很简的使用各个点的X/Y坐标的平均值。不过，我这里想告诉大家另三个求中心点的的公式：

1）Minkowski Distance 公式 —— λ 可以随意取值，可以是负数，也可以是正数，或是无穷大。

2）Euclidean Distance 公式 —— 也就是第一个公式 λ=2 的情况

3）CityBlock Distance 公式 —— 也就是第一个公式 λ=1 的情况

这三个公式的求中心点有一些不一样的地方，我们看下图（对于第一个 λ 在 0-1之间）。

（1）Minkowski Distance     （2）Euclidean Distance    （3） CityBlock Distance

上面这几个图的大意是他们是怎么个逼近中心的，第一个图以星形的方式，第二个图以同心圆的方式，第三个图以菱形的方式。

K-Means的演示

如果你以”K Means Demo“为关键字到Google里查你可以查到很多演示。这里推荐一个演示

http://home.dei.polimi.it/matteucc/Clustering/tutorial_html/AppletKM.html

操作是，鼠标左键是初始化点，右键初始化“种子点”，然后勾选“Show History”可以看到一步一步的迭代。

注：这个演示的链接也有一个不错的 K Means Tutorial 。

K-Means ++ 算法

K-Means主要有两个最重大的缺陷——都和初始值有关：

•  K 是事先给定的，这个 K 值的选定是非常难以估计的。很多时候，事先并不知道给定的数据集应该分成多少个类别才最合适。（ ISODATA 算法通过类的自动合并和分裂，得到较为合理的类型数目 K）

• K-Means算法需要用初始随机种子点来搞，这个随机种子点太重要，不同的随机种子点会有得到完全不同的结果。（K-Means++算法可以用来解决这个问题，其可以有效地选择初始点）

我在这里重点说一下 K-Means++算法步骤：

1. 先从我们的数据库随机挑个随机点当“种子点”。
2. 对于每个点，我们都计算其和最近的一个“种子点”的距离D(x)并保存在一个数组里，然后把这些距离加起来得到Sum(D(x))。
3. 然后，再取一个随机值，用权重的方式来取计算下一个“种子点”。这个算法的实现是，先取一个能落在Sum(D(x))中的随机值Random，然后用Random -= D(x)，直到其<=0，此时的点就是下一个“种子点”。
4. 重复第（2）和第（3）步直到所有的K个种子点都被选出来。
5. 进行K-Means算法。

相关的代码你可以在这里找到“implement the K-means++ algorithm”(墙) 另，Apache 的通用数据学库也实现了这一算法

K-Means 算法应用

看到这里，你会说，K-Means算法看来很简单，而且好像就是在玩坐标点，没什么真实用处。而且，这个算法缺陷很多，还不如人工呢。是的，前面的例子只是玩二维坐标点，的确没什么意思。但是你想一下下面的几个问题：

1）如果不是二维的，是多维的，如5维的，那么，就只能用计算机来计算了。

2）二维坐标点的X, Y 坐标，其实是一种向量，是一种数学抽象。现实世界中很多属性是可以抽象成向量的，比如，我们的年龄，我们的喜好，我们的商品，等等，能抽象成向量的目的就是可以让计算机知道某两个属性间的距离。如：我们认为，18岁的人离24岁的人的距离要比离12岁的距离要近，鞋子这个商品离衣服这个商品的距离要比电脑要近，等等。

只要能把现实世界的物体的属性抽象成向量，就可以用K-Means算法来归类了。

在 《k均值聚类(K-means)》 这篇文章中举了一个很不错的应用例子，作者用亚洲15支足球队的2005年到1010年的战绩做了一个向量表，然后用K-Means把球队归类，得出了下面的结果，呵呵。

• 亚洲一流：日本，韩国，伊朗，沙特
• 亚洲二流：乌兹别克斯坦，巴林，朝鲜
• 亚洲三流：中国，伊拉克，卡塔尔，阿联酋，泰国，越南，阿曼，印尼

其实，这样的业务例子还有很多，比如，分析一个公司的客户分类，这样可以对不同的客户使用不同的商业策略，或是电子商务中分析商品相似度，归类商品，从而可以使用一些不同的销售策略，等等。

最后给一个挺好的算法的幻灯片：http://www.cs.cmu.edu/~guestrin/Class/10701-S07/Slides/clustering.pdf

（全文完）

我们直接来看示例，如果我们需要来压缩下面的字符串：

 “beep boop beer!” 

首先，我们先计算出每个字符出现的次数，我们得到下面这样一张表 :

接下来就是我们的算法——把这个Priority Queue 转成二叉树。我们始终从queue的头取两个元素来构造一个二叉树（第一个元素是左结点，第二个是右结点），并把这两个元素的priority相加，并放回Priority中（再次注意，这里的Priority就是字符出现的次数），然后，我们得到下面的数据图表：

同样，我们再把前两个取出来，形成一个Priority为2+2=4的结点，然后再放回Priority Queue中 :

继续我们的算法（我们可以看到，这是一种自底向上的建树的过程）：

最终我们会得到下面这样一棵二叉树：

此时，我们把这个树的左支编码为0，右支编码为1，这样我们就可以遍历这棵树得到字符的编码，比如：‘b’的编码是 00，’p’的编码是101， ‘r’的编码是1000。我们可以看到出现频率越多的会越在上层，编码也越短，出现频率越少的就越在下层，编码也越长。

最终我们可以得到下面这张编码表：

这里需要注意的一点是，我们的Huffman对各个字符的编码是不会冲突的，也就是说，不会存在某一个编码是另一个编码的前缀，不然的话就会大问题了。因为encode后的编码是没有分隔符的。

于是，对于我们的原始字符串  beep boop beer!

其对就能的二进制为 : 0110 0010 0110 0101 0110 0101 0111 0000 0010 0000 0110 0010 0110 1111 0110 1111 0111 0000 0010 0000 0110 0010 0110 0101 0110 0101 0111 0010 0010 0001

我们的Huffman的编码为： 0011 1110 1011 0001 0010 1010 1100 1111 1000 1001

从上面的例子中，我们可以看到被压缩的比例还是很可观的。

作者给出了源码你可以看看（ C99标准） Download the source files

（全文完）