本文通过一道程序面试题，使用不同的编程语言来实现，检验每种语言的简单版本与优化后版本的运行速度分别是多少，横向对比 Python、Go、C++、C、Rust 等编程语言的性能，

作者 | Ben Hoyt       译者 | 弯月

出处 | https://benhoyt.com/writings/count-words/

出品 | CSDN（ID：CSDNnews）

最近几年，我经常担任编程面试的面试官，而我经常给应聘者出的一道题目是：

编写一个程序，计算标准输入中每个单词的出现频率，然后按照频率从高到低输出频率。例如，给定下列输入：

The foo the foo the

defenestration the

程序应当输出：

the 4

foo 2

defenestration 1

我认为这是一个非常好的面试题，它比 FizzBuzz 要难一点，但并不像“在白板上翻转二叉树”那般折磨人。而且在真实世界中，程序员也可能会编写类似的程序，它还能考察程序员对于文件 I/O、哈希表（映射）以及相应语言的排序函数的的掌握情况。排序部分有一点难度，因为大多数语言的哈希表是无序的，即使有序，也是按照键的顺序或插入顺序，而不是按照值的顺序排列。

在应聘者给出基本的解决方案后，我还会进一步追问：首字母大写的情况怎么处理？标点怎么处理？出现频率相同的单词怎样排序？性能瓶颈可能出现在哪里？算法时间复杂度是多少？内存使用率是多少？处理 1GB 的文件大概需要多长时间？对于 1TB 的文件是否能正常运行？等等。或者你可以从“软件工程”的方面来追问，询问诸如错误处理、可测试性、将其转换成命令行工具等问题。

最基本的解决方案就是逐行读取文件，转换成小写，将每一行分割成单词，然后利用哈希表统计频率。统计完成后，将哈希表转换成单词和个数组成的列表，按照个数排序（从大到小），然后输出。

对于 Python，最简单的方案就是使用 dict，如下所示（省略了 import 等）：

counts = {}
for line in sys.stdin:
    words = line.lower().split()
for word in words:
        counts[word] = counts.get(word, 0) + 1
pairs = sorted(counts.items(), key=lambda kv: kv[1], reverse=True)
for word, count in pairs:
print(word, count)

如果应聘者非常熟悉 Python，他们可能会使用 collections.defaultdict 甚至 collections.Counter（后面的代码就用到了 Counter）。遇到这种情况，我就会询问这个类的工作原理是什么，或者要求他们使用普通的字典来解决。

巧合的是，几十年前两名计算机科学家曾经就这个问题进行了一场决斗。

1986 年，Jon Bentley 要求 Donald Knuth 编写一段程序解决该问题，演示“文学编程”，然后 Knuth 写了一篇 10 页长的论文来解决这个问题（https://www.cs.tufts.edu/~nr/cs257/archive/don-knuth/pearls-2.pdf）。然后 Doug McIlroy（Unix 流水线的发明者）仅用了一行 Unix shell 程序，利用 tr、sort 和 uniq 解决了这个问题。

这个问题我也研究了一段时间，我想看看用各种语言写出来会是什么样子，以及每种语言的简单版本与优化后版本的运行速度分别是多少。

注意：本文包含大量的代码，源代码 GitHub 地址（https://github.com/benhoyt/countwords）。或者你也可以跳至性能评测结果。

问题描述和约束条件

每个程序必须读取标准输入，然后按照频率从高到低的顺序，输出由空格分隔的单词的出现频率。为了简单一致起见，下面是我附加的一些约束条件：

• 大小写：程序必须统一将单词转换成小写，也就是说 “The the THE” 在输出中应当显示为 “the 3”。

• 单词：任何由空白分隔的部分都算单词，忽略标点符号。这样做的确会降低程序的可用性，但我不想把这个问题变成一个分词问题。

• ASCII：在空白处理和大小写操作方面，可以仅考虑 ASCII 字符。绝大多数优化版本都会使用这个条件。

• 排序：如果两个单词的频率一样，那么它们在输出中的顺序无关紧要。我通过一个规范化脚本来检查输出结果的正确性。

• 多线程：程序应当在单台机器的单个线程上运行（尽管面试时我经常会问多线程的问题）。

• 内存：不要将整个文件读入内存。逐行缓冲是允许的，也可以按块读入，最大缓冲为 64KB。但是，将整个单词计数的哈希表放到内存中是允许的（我们假设输入是真实的语言，而不是完全由随机字符组成的单词）。

• 文本：假设输入文件为文本文件，每一行的长度很合理，每行的长度小于缓冲大小。

• 安全性：即使是优化的版本，也尽量不要使用不安全的语言特性，不要使用汇编。

• 哈希表：不要自己实现哈希表（除了优化的 C语言版本之外）。

• 标准库：仅使用语言的标准库函数。

测试输入文件为重复了十次的詹姆士王译本的圣经。源文件可以从 Gutenberg.org 获得，但将其中的智能双引号替换成了 ASCII 的双引号字符，并使用 cat 重复了十次，最终获得了一个 43MB 大小的输入文件。

下面开始写代码！代码的顺序为我编写的先后顺序。

Python

Python 的简单版使用了 collections.Counter。Python 的 collections 库非常好用。这可以说是最简单的一个实现：

simple.py
counts = collections.Counter()
for line in sys.stdin:
    words = line.lower().split()
    counts.update(words)
for word, count in counts.most_common():
print(word, count)

这个版本能处理 Unicode，在实际工作中我多半会这样写。实际上这个解决方案效率非常高，因为所有底层的东西都是由 C 语言完成的，包括读取文件、转换成小写、按照空白分割、更新计数器，以及 Counter.most_common 实现的排序。

但是我们来试着优化一下！Python 自带了性能评测模块，名为 cProfile。使用非常简单，只需用 python3 -m cProfile 运行程序即可。我注释掉了最后的 print 语句，避免输出干扰程序的性能。下面是输出结果（添加了 -s tottime 选项按照每个函数的总执行时间排序）：

$ python3 -m cProfile -s tottime simple.py <kjvbible_x10.txt
6997799 function calls (6997787 primitive calls) in 3.872 seconds
Ordered by: internal time
ncalls  tottime  percall  cumtime  percall filename:lineno(function)
   998170    1.361    0.000    1.361    0.000 {built-in method _collections._count_elements}
1    0.911    0.911    3.872    3.872 simple.py:1(<module>)
   998170    0.415    0.000    0.415    0.000 {method 'split' of 'str' objects}
998171    0.405    0.000    2.388    0.000 __init__.py:608(update)
998170    0.270    0.000    0.622    0.000 {built-in method builtins.isinstance}
998170    0.182    0.000    0.351    0.000 abc.py:96(__instancecheck__)
998170    0.170    0.000    0.170    0.000 {built-in method _abc._abc_instancecheck}
998170    0.134    0.000    0.134    0.000 {method 'lower' of 'str' objects}
5290    0.009    0.000    0.018    0.000 codecs.py:319(decode)
5290    0.009    0.000    0.009    0.000 {built-in method _codecs.utf_8_decode}
1    0.007    0.007    0.007    0.007 {built-in method builtins.sorted}
7/1    0.000    0.000    0.000    0.000 {built-in method _abc._abc_subclasscheck}
1    0.000    0.000    0.007    0.007 __init__.py:559(most_common)
1    0.000    0.000    0.000    0.000 __init__.py:540(__init__)
1    0.000    0.000    3.872    3.872 {built-in method builtins.exec}
7/1    0.000    0.000    0.000    0.000 abc.py:100(__subclasscheck__)
7    0.000    0.000    0.000    0.000 _collections_abc.py:392(__subclasshook__)
1    0.000    0.000    0.000    0.000 {method 'disable' of '_lsprof.Profiler' objects}
1    0.000    0.000    0.000    0.000 {method 'items' of 'dict' objects}

可以看出：

• 998,170 是输入的总行数，因为我们逐行读入，因此 Python 循环和函数调用都执行了这么多次。

• 大部分时间花在了 simple.py 本身上，说明 Python 字节码的执行（相对来说）非常慢。主循环完全是用 Python 写的，也执行了 998,170 次。

• str.split 相对较慢，可能是因为它需要分配内存并复制很多字符串。

• Counter.update 调用了 isinstance，加起来也花了许多时间。我考虑过直接调用 C 函数 _count_elements，但这属于实现细节，我认为应当算作“不安全”的操作。

我们能做的主要就是减少 Python 主循环的执行次数，从而减少调用那些函数的次数。我们以 64KB 的块为单位来读入：

optimized.py
counts = collections.Counter()
remaining = ''
while True:
    chunk = remaining + sys.stdin.read(64*1024)
if not chunk:
break
    last_lf = chunk.rfind('\n')  # process to last LF character
if last_lf == -1:
        remaining = ''
else:
        remaining = chunk[last_lf+1:]
        chunk = chunk[:last_lf]
    counts.update(chunk.lower().split())
for word, count in counts.most_common():
    print(word, count)

之前的版本主循环每次处理 42 个字符（平均行长度），而这段程序每次处理 65536 个字符（去掉末尾不完整的行）。读取和处理的字节数是相同的，但现在绝大多数处理都在C语言中进行，而不是在 Python 循环中进行。许多优化方案都采取了这种方式，即一次性处理大量数据。

性能评测结果看起来好多了。_count_elements和str.split 函数依然花费了最长的时间，但与之前的 998170 次相比，它们现在被调用了 662 次（因为每次大概处理 64KB 数据，而不是先前的 42 字节）：

$ python3 -m cProfile -s tottime optimized.py <kjvbible_x10.txt
         7980 function calls (7968 primitive calls) in 1.280 seconds


   Ordered by: internal time


   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
      662    0.870    0.001    0.870    0.001 {built-in method _collections._count_elements}
      662    0.278    0.000    0.278    0.000 {method 'split' of 'str' objects}
        1    0.080    0.080    1.280    1.280 optimized.py:1(<module>)
      662    0.028    0.000    0.028    0.000 {method 'lower' of 'str' objects}
      663    0.010    0.000    0.016    0.000 {method 'read' of '_io.TextIOWrapper' objects}
        1    0.007    0.007    0.007    0.007 {built-in method builtins.sorted}
      664    0.004    0.000    0.004    0.000 {built-in method _codecs.utf_8_decode}
      663    0.001    0.000    0.872    0.001 __init__.py:608(update)
      664    0.001    0.000    0.005    0.000 codecs.py:319(decode)
      662    0.001    0.000    0.001    0.000 {built-in method builtins.isinstance}
      662    0.000    0.000    0.001    0.000 {built-in method _abc._abc_instancecheck}
      662    0.000    0.000    0.000    0.000 {method 'rfind' of 'str' objects}
      664    0.000    0.000    0.000    0.000 codecs.py:331(getstate)
      662    0.000    0.000    0.001    0.000 abc.py:96(__instancecheck__)
      7/1    0.000    0.000    0.000    0.000 {built-in method _abc._abc_subclasscheck}
        1    0.000    0.000    0.007    0.007 __init__.py:559(most_common)
        1    0.000    0.000    0.000    0.000 __init__.py:540(__init__)
      7/1    0.000    0.000    0.000    0.000 abc.py:100(__subclasscheck__)
        1    0.000    0.000    1.280    1.280 {built-in method builtins.exec}
        7    0.000    0.000    0.000    0.000 _collections_abc.py:392(__subclasshook__)
        1    0.000    0.000    0.000    0.000 {method 'disable' of '_lsprof.Profiler' objects}
        1    0.000    0.000    0.000    0.000 {method 'items' of 'dict' objects}

我还发现，在 Python 程序中，读取和处理 bytes 和 str 的性能并没有显著区别（utf_8_decode 在上面的列表中位于非常靠下的位置）。此外，只要缓冲区大小超过 2KB，速度就不会比 64KB 缓冲区慢。我注意到许多系统的默认缓冲区大小是 4KB，可能就是由于这个原因。

我尝试了许多其他方法来提高性能，但这是我用标准Python能达到的最好结果。（我尝试用 PyPy 优化编译器运行，但由于某种原因，它非常慢。）Python不适合在字节层面进行优化（反而会导致 10 倍至 100 倍的速度下降），任何字符处理都应该在 C 层面进行。

 

Go

简单的 Go 版本可以使用 buffio.Scanner，并使用 ScanWords 来分割单词。Go 没有 Python 的 collections.Counter，但使用 map[string] 进行计数也非常简单，然后通过单词和计数组成的切片进行排序操作：

simple.go


func main() {
    scanner := bufio.NewScanner(os.Stdin)
    scanner.Split(bufio.ScanWords)
    counts := make(map[string]int)
    for scanner.Scan() {
        word := strings.ToLower(scanner.Text())
        counts[word]++
    }
    if err := scanner.Err(); err != nil {
        fmt.Fprintln(os.Stderr, err)
        os.Exit(1)
    }


    var ordered []Count
    for word, count := range counts {
        ordered = append(ordered, Count{word, count})
    }
    sort.Slice(ordered, func(i, j int) bool {
        return ordered[i].Count > ordered[j].Count
    })


    for _, count := range ordered {
        fmt.Println(string(count.Word), count.Count)
    }
}


type Count struct {
    Word  string
    Count int
}

Go 的简单版本比 Python 的简单版本要快很多，但只比优化过的 Python 版本快一点点（而且代码行数几乎是两倍，因为Go 有很多样板代码，还要考虑很多底层处理）。

Go 的性能评测工具需要在程序开始处添加几行代码（import "runtime/pprof" 略过）：

f, err := os.Create("cpuprofile")
if err != nil {
    fmt.Fprintf(os.Stderr, "could not create CPU profile: %v\n", err)
    os.Exit(1)
}
if err := pprof.StartCPUProfile(f); err != nil {
    fmt.Fprintf(os.Stderr, "could not start CPU profile: %v\n", err)
    os.Exit(1)
}
defer pprof.StopCPUProfile()

运行程序后就可以通过下述命令查看 CPU 的使用情况：

$ go tool pprof -http=:7777 cpuprofile 
Serving web UI on http://localhost:7777

结果很有意思，尽管并不是完全出乎意料，处理每个单词的循环占据了几乎所有时间。还有相当多的时间花在了扫描器上，而向map中插入字符串所需的内存分配也花了不少时间。我们来优化一下这几个方面。

为了改善扫描过程，我们在读取文件的同时进行扫描和大小写转换。为了减少内存分配，我们采用map[string]*int 来替代 map[string]int，这样每个单词只需要分配一次内存，而不是在每次增量的时候都要分配。

我尝试了几次后才得到了这个结果。中间的一个步骤是依然使用 bufio.Scanner，但采用自定义的分割函数 scanWordsASCII。但是，尽管的确快了一些，但去掉 bufio.Scanner 也并不难。我做的另一个尝试是使用自定义的哈希表，但我认为这已经超出了本问题的范围，而且也并不比map[string]*int 快多少。

optimized.go


func main() {
    var word []byte
    buf := make([]byte, 64*1024)
    counts := make(map[string]*int)
    for {
        // Read input in 64KB blocks till EOF.
        n, err := os.Stdin.Read(buf)
        if err != nil && err != io.EOF {
            fmt.Fprintln(os.Stderr, err)
            os.Exit(1)
        }
        if n == 0 {
            break
        }


        // Count words in the buffer.
        for i := 0; i < n; i++ {
            c := buf[i]


            // Found a whitespace char, count last word.
            if c <= ' ' {
                if len(word) > 0 {
                    increment(counts, word)
                    word = word[:0] // reset word buffer
                }
                continue
            }


            // Convert to ASCII lowercase as we go.
            if c >= 'A' && c <= 'Z' {
                c = c + ('a' - 'A')
            }


            // Add non-space char to word buffer.
            word = append(word, c)
        }
    }


    // Count last word, if any.
    if len(word) > 0 {
        increment(counts, word)
    }


    // Convert to slice of Count, sort by count descending, print.
    ordered := make([]Count, 0, len(counts))
    for word, count := range counts {
        ordered = append(ordered, Count{word, *count})
    }
    sort.Slice(ordered, func(i, j int) bool {
        return ordered[i].Count > ordered[j].Count
    })
    for _, count := range ordered {
        fmt.Println(count.Word, count.Count)
    }
}


func increment(counts map[string]*int, word []byte) {
    if p, ok := counts[string(word)]; ok {
        // Word already in map, increment existing int via pointer.
        *p++
        return
    }
    // Word not in map, insert new int.
    n := 1
    counts[string(word)] = &n
}

现在评测结果非常扁平，几乎所有时间都花在了主循环或 map 访问中：

Go 能提供更接近底层的控制（而且你可以更加深入，比如将I/O映射到内存，自定义哈希表等）。但是，程序员的时间也非常宝贵，所以需要取舍。在实际工作中，我会依然使用 bufio.scanner 和 ScanWords、bytes.ToLower，以及优化方案中的 map[string]*int。

C++

自从我最后一次使用 C++ 以来，它有了很多的改进：C++11 添加了许多优秀的功能，然后 C++14、17和20又添加了更多功能。现在的 C++ 比旧时的 C++ 更加复杂了，尽管错误信息依然很乱。下面是我写的一个简单版本：

simple.cpp


int main() {
    std::string word;
    std::unordered_map<std::string, int> counts;
    while (std::cin >> word) {
        std::transform(word.begin(), word.end(), word.begin(),
            [](unsigned char c){ return std::tolower(c); });
        ++counts[word];
    }
    if (std::cin.bad()) {
        std::cerr << "error reading stdin\n";
        return 1;
    }


    std::vector<std::pair<std::string, int>> ordered(counts.begin(),
        counts.end());
    std::sort(ordered.begin(), ordered.end(),
        [](auto const& a, auto const& b) { return a.second>b.second; });


    for (auto const& count : ordered) {
        std::cout << count.first << " " << count.second << "\n";
    }
}

在进行优化时，我做的第一件事情就是启用了编译优化（g++ -O2）。我喜欢 Go 的原因之一就是你不需要操心这些，因为优化是永远启用的。

我注意到 I/O 比较慢。有一行魔法代码，只要加到程序开头，禁止在每次 I/O 操作之后与 C 标准库的函数进行同步，就能提高性能。这行代码几乎可以将速度提高到两倍：

ios::sync_with_stdio(false);


GCC可以使用gprof生成性能评测报告。下面是部分结果：


index % time    self  children    called     name
                                  13             frame_dummy [1]
[1]    100.0    0.01    0.00       0+13      frame_dummy [1]
                                  13             frame_dummy [1]
-----------------------------------------------
                0.00    0.00   32187/32187       std::vector<std::pair\
<std::__cxx11::basic_string<char, std::char_traits<char>, std::allocat\
or<char> >, int>, std::allocator<std::pair<std::__cxx11::basic_string<\
char, std::char_traits<char>, std::allocator<char> >, int> > >::vector\
<std::__detail::_Node_iterator<std::pair<std::__cxx11::basic_string<ch\
ar, std::char_traits<char>, std::allocator<char> > const, int>, false,\
 true>, void>(std::__detail::_Node_iterator<std::pair<std::__cxx11::ba\
sic_string<char, std::char_traits<char>, std::allocator<char> > const,\
 int>, false, true>, std::__detail::_Node_iterator<std::pair<std::__cx\
x11::basic_string<char, std::char_traits<char>, std::allocator<char> >\
 const, int>, false, true>, std::allocator<std::pair<std::__cxx11::bas\
ic_string<char, std::char_traits<char>, std::allocator<char> >, int> >\
 const&) [11]
[8]      0.0    0.00    0.00   32187         void std::__cxx11::basic_\
string<char, std::char_traits<char>, std::allocator<char> >::_M_constr\
uct<char*>(char*, char*, std::forward_iterator_tag) [8]
-----------------------------------------------
                0.00    0.00       1/1           __libc_csu_init [17]
[9]      0.0    0.00    0.00       1         _GLOBAL__sub_I_main [9]
...

这里出现了 C++ 模板。可能我比较古板，与 std::basic_istream<char, std::char_traits<char> >& std::operator>><char, std::char_traits<char>, std::allocator<char> >(std::basic_istream<char, std::char_traits<char> >&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> >&)相比，我更喜欢 malloc 和 scanf。

我很不喜欢分析这些结果，所以我直接去编写优化后的 C 版本了。当然 C++ 版本依然有许多可以改进的地方，但我认为最后肯定会越来越底层，越来越像C（至少以我相当有限的现代 C++ 知识而言会这样），所以你可以直接看下面的 C 版本那就好了。

在 C 版本中我使用了 Valgrind 的评测工具（Callgrind），参见下面一节的注释。其实我还可以使用 Linux 的 perf 工具（具体来说是 perf record 和 perf report 命令），它的结果要比 gprof 好一些。

C

C 是一门永不过时的语言：快速、不安全、简单。我依然很喜欢 C，因为它不像 C++ 那样难懂，而且我可以随便深入底层。C 语言非常通用（Linux 内核、Redis、PostgreSQL、SQLite 等许多许多库都是用 C 写的），而且永远不会消亡。所以我们来编写一个 C 语言版本。

不幸的是，C 的标准库并没有哈希表数据结构。但是我们还有 libc，它提供了 hcreate 和 hsearch 哈希表函数，所以这里我们破例允许使用非标准库的 libc 函数。在优化版本中我们会编写自己的哈希表。

hcreate 有一个不方便的地方是，你必须提前指定哈希表的最大尺寸。我知道唯一的单词数量不会超过 30000，所以这里我们暂时选择 60000。

simple.c


#define MAX_UNIQUES 60000


typedef struct {
    char* word;
    int count;
} count;


// Comparison function for qsort() ordering by count descending.
int cmp_count(const void* p1, const void* p2) {
    int c1 = ((count*)p1)->count;
    int c2 = ((count*)p2)->count;
    if (c1 == c2) return 0;
    if (c1 < c2) return 1;
    return -1;
}


int main() {
    // The hcreate hash table doesn't provide a way to iterate, so
    // store the words in an array too (also used for sorting).
    count* words = calloc(MAX_UNIQUES, sizeof(count));
    int num_words = 0;


    // Allocate hash table.
    if (hcreate(MAX_UNIQUES) == 0) {
        fprintf(stderr, "error creating hash table\n");
        return 1;
    }


    char word[101]; // 100-char word plus NUL byte
    while (scanf("%100s", word) != EOF) {
        // Convert word to lower case in place.
        for (char* p = word; *p; p++) {
            *p = tolower(*p);
        }


        // Search for word in hash table.
        ENTRY item = {word, NULL};
        ENTRY* found = hsearch(item, FIND);
        if (found != NULL) {
            // Word already in table, increment count.
            int* pn = (int*)found->data;
            (*pn)++;
        } else {
            // Word not in table, insert it with count 1.
            item.key = strdup(word); // need to copy word
            if (item.key == NULL) {
                fprintf(stderr, "out of memory in strdup\n");
                return 1;
            }
            int* pn = malloc(sizeof(int));
            if (pn == NULL) {
                fprintf(stderr, "out of memory in malloc\n");
                return 1;
            }
            *pn = 1;
            item.data = pn;
            ENTRY* entered = hsearch(item, ENTER);
            if (entered == NULL) {
                fprintf(stderr, "table full, increase MAX_UNIQUES\n");
                return 1;
            }


            // And add to words list for iterating.
            words[num_words].word = item.key;
            num_words++;
        }
    }


    // Iterate once to add counts to words list, then sort.
    for (int i = 0; i < num_words; i++) {
        ENTRY item = {words[i].word, NULL};
        ENTRY* found = hsearch(item, FIND);
        if (found == NULL) { // shouldn't happen
            fprintf(stderr, "key not found: %s\n", item.key);
            return 1;
        }
        words[i].count = *(int*)found->data;
    }
    qsort(&words[0], num_words, sizeof(count), cmp_count); 


    // Iterate again to print output.
    for (int i = 0; i < num_words; i++) {
        printf("%s %d\n", words[i].word, words[i].count);
    }


    return 0;
}

里面有相当一部分样板代码（大部分是内存分配和错误检查），但由于 C 非常底层，我认为这并不是问题。有难度的地方基本上都被隐藏起来了，例如 scanf会进行分词，hsearch 会进行哈希表操作。这段程序本身也非常快，而且非常小（在 Linux 上的可执行文件只有 17KB）。

我试着用 gprof 进行性能评测，但没有得到任何有用的结果（也许采样不够频繁？），所以我采用了 Valgrind 的评测工具 Callgrind。这是我第一次使用该工具，但看起来它非常强大。

在使用 gcc -g 编译后，运行下面的命令生成评测报告：

valgrind --tool=callgrind ./simple-c <kjvbible_x10.txt >/dev/null

不出所料，scanf 是最慢的，接下来是 hsearch。所以我们接下来的优化会激进一些。我主要做了三件事情：

• 分块读取文件，像 Go 和 Python 一样，这样可以避免 scanf 的额外开销。

• 字节仅处理一次，或者尽量减少次数。在进行分词的同时进行大小写转换并计算哈希。

• 使用 FNV-1 哈希函数实现自己的哈希表。

optimized.c


#define BUF_SIZE 65536
#define HASH_LEN 65536  // must be a power of 2
#define FNV_OFFSET 14695981039346656037UL
#define FNV_PRIME 1099511628211UL


// Used both for hash table buckets and array for sorting.
typedef struct {
    char* word;
    int word_len;
    int count;
} count;


// Comparison function for qsort() ordering by count descending.
int cmp_count(const void* p1, const void* p2) {
    int c1 = ((count*)p1)->count;
    int c2 = ((count*)p2)->count;
    if (c1 == c2) return 0;
    if (c1 < c2) return 1;
    return -1;
}


count* table;
int num_unique = 0;


// Increment count of word in hash table (or insert new word).
void increment(char* word, int word_len, uint64_t hash) {
    // Make 64-bit hash in range for items slice.
    int index = (int)(hash & (uint64_t)(HASH_LEN-1));


    // Look up key, using direct match and linear probing if not found.
    while (1) {
        if (table[index].word == NULL) {
            // Found empty slot, add new item (copying key).
            char* word_copy = malloc(word_len);
            if (word_copy == NULL) {
                fprintf(stderr, "out of memory\n");
                exit(1);
            }
            memmove(word_copy, word, word_len);
            table[index].word = word_copy;
            table[index].word_len = word_len;
            table[index].count = 1;
            num_unique++;
            return;
        }
        if (table[index].word_len == word_len &&
                memcmp(table[index].word, word, word_len) == 0) {
            // Found matching slot, increment existing count.
            table[index].count++;
            return;
        }
        // Slot already holds another key, try next slot (linear probe).
        index++;
        if (index >= HASH_LEN) {
            index = 0;
        }
    }
}


int main() {
    // Allocate hash table buckets.
    table = calloc(HASH_LEN, sizeof(count));
    if (table == NULL) {
        fprintf(stderr, "out of memory\n");
        return 1;
    }


    char buf[BUF_SIZE];
    int offset = 0;
    while (1) {
        // Read file in chunks, processing one chunk at a time.
        size_t num_read = fread(buf+offset, 1, BUF_SIZE-offset, stdin);
        if (num_read+offset == 0) {
            break;
        }


        // Find last space or linefeed in buf and process up to there.
        int space;
        for (space = offset+num_read-1; space>=0; space--) {
            char c = buf[space];
            if (c <= ' ') {
                break;
            }
        }
        int num_process = (space >= 0) ? space : (int)num_read+offset;


        // Scan chars to process: tokenize, lowercase, and hash as we go.
        int i = 0;
        while (1) {
            // Skip whitespace before word.
            for (; i < num_process; i++) {
                char c = buf[i];
                if (c > ' ') {
                    break;
                }
            }
            // Look for end of word, lowercase and hash as we go.
            uint64_t hash = FNV_OFFSET;
            int start = i;
            for (; i < num_process; i++) {
                char c = buf[i];
                if (c <= ' ') {
                    break;
                }
                if (c >= 'A' && c <= 'Z') {
                    c += ('a' - 'A');
                    buf[i] = c;
                }
                hash *= FNV_PRIME;
                hash ^= (uint64_t)c;
            }
            if (i <= start) {
                break;
            }
            // Got a word, increment count in hash table.
            increment(buf+start, i-start, hash);
        }


        // Move down remaining partial word.
        if (space >= 0) {
            offset = (offset+num_read-1) - space;
            memmove(buf, buf+space+1, offset);
        } else {
            offset = 0;
        }
    }


    count* ordered = calloc(num_unique, sizeof(count));
    for (int i=0, i_unique=0; i<HASH_LEN; i++) {
        if (table[i].word != NULL) {
            ordered[i_unique++] = table[i];
        }
    }
    qsort(ordered, num_unique, sizeof(count), cmp_count);
    for (int i=0; i<num_unique; i++) {
        printf("%.*s %d\n",
                ordered[i].word_len, ordered[i].word, ordered[i].count);
    }


    return 0;
}

上面大约是 150 行代码（包括空行和注释），绝对是最长的程序，但是还好！你可以看到，利用线性扫描编写自己的哈希表并没有使用太多代码。固定表大小并不好，但是动态改变表大小会增加许多工作量，而且固定表大小并不会显著减慢运行速度，所以我将动态表大小留给读者作为练习。

可执行文件依然是 17KB（这是我喜欢C的原因）。而且，不出所料，这个版本是最快的，比优化后的 Go 版本还要快一些，因为我们编写了自己的哈希表，而且处理的字节数更少。

但是，这段代码比 GNU grep 要慢。GNU grep 最初的作者 Mike Haertel 在2010年写过一篇非常著名的文章解释为什么GNU grep这么快（https://lists.freebsd.org/pipermail/freebsd-current/2010-August/019310.html）。但是有人认为，那篇文章有些过时了。Mike 的一般性建议非常好，但是今天 GNU grep 很快的原因并不是它使用 Boyer-Moore 算法跳过了一些字节，而是因为它使用了 glibc 中速度极快的向量化版本的 memchr。

我相信这 个C 版本还能进一步优化，比如 I/O 内存映射，避免一次处理一个字节，使用更好的数据结构进行计数等等。但暂时就这样吧！

AWK

AWK 实际上非常适合这个任务：读取每一行并按空格分割单词，对于它来说是小菜一碟。AWK 做不到的事情之一就是排序（如果不使用 Gawk 特有的排序功能的话），所以我用 AWK 的管道运算符将输出发送给 sort。下面是简单版的代码：

simple.awk


{
    for (i = 1; i <= NF; i++)
        counts[tolower($i)]++
}


END {
    for (k in counts)
        print k, counts[k] | "sort -nr -k2"
}

我不知道有什么简单的办法可以在底层进行性能评测（Gawk 有一个评测模式，但只是输出每一行执行了多少次）。我也不知道怎样用 AWK 或 Gawk 按块读取输入。

优化版本的一个小改动就是针对每一行调用 tolower，而不是针对每个单词。主循环变成这样：

optimized.awk


{
    $0 = tolower($0)
    for (i = 1; i <= NF; i++)
        counts[$i]++
}
...

可以使用 gawk -b 命令运行该程序，这样可以将 Gawk 设置为“字节”模式，如此一来，它就会使用 ASCII，而不是 UTF-8。这个“优化”过的版本在 gawk -b下运行时，比直接使用 gawk 运行简单版本大约快 1.6 倍。

有意思的是，Gawk 的手册有一页正好介绍了这个词频统计问题，其示例使用 AWK 的 gsub 函数演示了如何去掉标点。

Forth

Forth 是我学习的第一门编程语言（它非常好，而且非常能拓展思维），所以我决定使用 Gforth 写一个 Forth 版本。我有很多年没使用过这个语言了。

simple.fs


200 constant max-line
create line max-line allot  \ Buffer for read-line
wordlist constant counts    \ Hash table of words to count
variable num-uniques  0 num-uniques !


\ Convert character to lowercase.
: to-lower ( C -- c )
    dup [char] A [ char Z 1+ ] literal within if
        32 +
    then ;


\ Convert string to lowercase in place.
: lower-in-place ( addr u -- )
    over + swap ?do
        i c@ to-lower i c!
    loop ;


\ Count given word in hash table.
: count-word ( addr u -- )
    2dup counts search-wordlist if
        \ Increment existing word
        >body 1 swap +!
        2drop
    else
        \ Insert new word with count 1
        2dup lower-in-place
        ['] create execute-parsing 1 ,
        1 num-uniques +!
    then ;


\ Process text in the source buffer (one line).
: process-input ( -- )
    begin
        parse-name dup
    while
        count-word
    repeat
    2drop ;


\ Less-than for words (true if count is *greater* for reverse sort).
: count< ( nt1 nt2 -- )
    >r name>interpret >body @
    r> name>interpret >body @
    > ;


\ ... Definition of "sort" elided ...


\ Append word from wordlist to array at given offset.
: append-word ( addr offset nt -- addr offset+cell true )
    >r 2dup + r> swap !
    cell+ true ;


\ Show "word count" line for each word, most frequent first.
: show-words ( -- )
    num-uniques @ cells allocate throw
    0 ['] append-word counts traverse-wordlist drop
    dup num-uniques @ sort
    num-uniques @ 0 ?do
        dup i cells + @
        dup name>string type space
        name>interpret >body @ . cr
    loop
    drop ;


: main ( -- )
    counts set-current  \ Define into counts wordlist
    begin
        line max-line stdin read-line throw
    while
        line swap ['] process-input execute-parsing
    repeat
    drop
    show-words ;

Forth 被誉为“只能写不能读”的语言，这是有理由的。我曾经很喜欢没有局部变量的思想，但在实践中，这只会导致大量的 dup over swap rot。此外，即使是 Gforth（比标准的 Forth 强大很多）也不提供 to-lower 或 sort 等非常原始的工具，所以一切都要自己写。

幸运的是，哈希表可以通过 wordlist 实现。wordlist 的本来用途是定义新的 Forth 单词，但利用 Gforth 的 execute-parsing 扩展，把它当做哈希表使用也非常好。而 skip 和 scan 也很适合进行空白分词。

至于优化，似乎只需运行 gforth-fast 而不是 gforth，就能魔法般地提高运行速度，所以我的第一个优化版本就是这么简单（尽管在性能评测中两个版本都是使用 gforth-fast 运行的）。看起来 gforth-fast 能够避免调用的额外开销，但无法为错误生成友好的栈跟踪信息。

optimized.fs


\ Start hash table at larger size
15 :noname to hashbits hashdouble ; execute


65536 constant buf-size
create buf buf-size allot  \ Buffer for read-file
wordlist constant counts   \ Hash table of words to count
variable num-uniques  0 num-uniques !


\ Convert character to lowercase.
: to-lower ( C -- c )
    dup [char] A [ char Z 1+ ] literal within if
        32 +
    then ;


\ Convert string to lowercase in place.
: lower-in-place ( addr u -- )
    over + swap ?do
        i c@ to-lower i c!
    loop ;


\ Count given word in hash table.
: count-word ( c-addr u -- )
    2dup counts find-name-in dup if
        ( name>interpret ) >body 1 swap +! 2drop
    else
        drop nextname create 1 ,
        1 num-uniques +!
    then ;


\ Process text in the buffer.
: process-string ( -- )
    begin
        parse-name dup
    while
        count-word
    repeat
    2drop ;


\ Less-than for words (true if count is *greater* for reverse sort).
: count< ( nt1 nt2 -- )
    >r name>interpret >body @
    r> name>interpret >body @
    > ;


\ ... Definition of "sort" elided ...


\ Append word from wordlist to array at given offset.
: append-word ( addr offset nt -- addr offset+cell true )
    dup name>string lower-in-place
    >r  2dup + r> swap !
    cell+ true ;


\ Show "word count" line for each word, most frequent first.
: show-words ( -- )
    num-uniques @ cells allocate throw
    0 ['] append-word counts traverse-wordlist drop
    dup num-uniques @ sort
    num-uniques @ 0 ?do
        dup i cells + @
        dup name>string type space
        name>interpret >body @ . cr
    loop
    drop ;


\ Find last LF character in string, or return -1.
: find-eol ( addr u -- eol-offset|-1 )
    begin
        1- dup 0>=
    while
        2dup + c@ 10 = if
            nip exit
        then
    repeat
    nip ;


: main ( -- )
    counts set-current  \ Define into counts wordlist
    0 >r  \ offset after remaining bytes
    begin
        \ Read from remaining bytes till end of buffer
        buf r@ + buf-size r@ - stdin read-file throw dup
    while
        \ Process till last LF
        buf over r@ + find-eol
        dup buf swap ['] process-string execute-parsing
        \ Move leftover bytes to start of buf, update offset
        dup buf + -rot  buf -rot  - r@ +
        r> drop dup >r  move
    repeat
    drop r> drop
    show-words ;

也许在非常小的嵌入式系统中我会考虑使用 Forth，但我不认为它有足够的函数库用于通用程序的编写。如果想使用更加现代化的类 Forth 语言，可以看看 Factor。

Rust

我经常使用 Andrew Gallant 的 ripgrep 代码搜索工具，我也知道他非常喜欢 Rust（及其优化），所以我在发表该文之前，询问了他是否愿意提供一个 Rust版本。

结果他不仅同意了，编写了简单版本和优化版本（类似于 Go 语言，速度也很相近），而且还提供了三个额外的版本（一些版本使用了外部依赖）：

• 一个“额外奖励”版本：类似于简单版，但能够处理 Unicode 分词，而且还有一些其他的优点。Andrew 说，这个版本很适合作为通用版本。

• 一个自定义哈希表版本。基本上是我的 C 版本的移植，速度也几乎一样快。

• 一个 trie 版本，使用了 trie 数据结构替代哈希表。Andrew 认为这个数据结构能更快一些，但事实证明它慢了一点点。

他的简单版本没有使用任何外部依赖，与前面的 Go 和 C++ 的简单版本非常相似。

rust/simple/main.rs


fn main() {
    // We don't return Result from main because it prints the debug
    // representation of the error. The code below prints the "display"
    // or human readable representation of the error.
    if let Err(err) = try_main() {
        eprintln!("{}", err);
        std::process::exit(1);
    }
}


fn try_main() -> Result<(), Box<dyn Error>> {
    let stdin = io::stdin();
    let stdin = io::BufReader::new(stdin.lock());
    let mut counts: HashMap<String, u64> = HashMap::new();
    for result in stdin.lines() {
        let line = result?;
        for word in line.split_whitespace() {
            let canon = word.to_lowercase();
            *counts.entry(canon).or_insert(0) += 1;
        }
    }


    let mut ordered: Vec<(String, u64)> = counts.into_iter().collect();
    ordered.sort_by(|&(_, cnt1), &(_, cnt2)| cnt1.cmp(&cnt2).reverse());
    for (word, count) in ordered {
        writeln!(io::stdout(), "{} {}", word, count)?;
    }
    Ok(())
}

下面是优化的版本。我引用了他的评论，代码只从 try_main 开始：

“这个版本基本上是优化过的 Go 版本低移植。它的缓冲区处理部分稍稍简单一些，我们不需要考虑最后一个换行字符的处理。（这样能节省一些工作量，但只能在每个 64KB 处理后节省，所以其实可以忽略不计。只是我认为稍稍简单一些。）

“这里只是用一个从加密角度来看，用不十分安全的哈希算法替换掉了 std 的默认哈希算法，此外并没有太多有趣的东西。

“std 默认使用了加密安全的哈希算法，速度略慢。在这段程序中，fxhash 和 fnv 的性能似乎相等，而 fxhash 在我的评测中略胜一筹。如果我们需要强制‘没有外部依赖’规则，我们也可以非常容易地自己编写fnv的实现。”

rust/optimized/main.rs


fn try_main() -> Result<(), Box<dyn Error>> {
    let stdin = io::stdin();
    let mut stdin = stdin.lock();
    let mut counts: HashMap<Vec<u8>, u64> = HashMap::default();
    let mut buf = vec![0; 64 * (1<<10)];
    let mut offset = 0;
    let mut start = None;
    loop {
        let nread = stdin.read(&mut buf[offset..])?;
        if nread == 0 {
            if offset > 0 {
                increment(&mut counts, &buf[..offset+1]);
            }
            break;
        }
        let buf = &mut buf[..offset+nread];


        for i in (0..buf.len()).skip(offset) {
            let b = buf[i];
            if b'A' <= b && b <= b'Z' {
                buf[i] += b'a' - b'A';
            }
            if b == b' ' || b == b'\n' {
                if let Some(start) = start.take() {
                    increment(&mut counts, &buf[start..i]);
                }
            } else if start.is_none() {
                start = Some(i);
            }
        }
        if let Some(ref mut start) = start {
            offset = buf.len() - *start;
            buf.copy_within(*start.., 0);
            *start = 0;
        } else {
            offset = 0;
        }
    }


    let mut ordered: Vec<(Vec<u8>, u64)> = counts.into_iter().collect();
    ordered.sort_by(|&(_, cnt1), &(_, cnt2)| cnt1.cmp(&cnt2).reverse());


    for (word, count) in ordered {
        writeln!(io::stdout(), "{} {}", std::str::from_utf8(&word)?,
                 count)?;
    }
    Ok(())
}


fn increment(counts: &mut HashMap<Vec<u8>, u64>, word: &[u8]) {
    // using 'counts.entry' would be more idiomatic here, but doing so
    // requires allocating a new Vec<u8> because of its API. Instead, we
    // do two hash lookups, but in the exceptionally common case (we see
    // a word we've already seen), we only do one and without any allocs.
    if let Some(count) = counts.get_mut(word) {
        *count += 1;
        return;
    }
    counts.insert(word.to_vec(), 1);
}

Unix shell

我们尝试一下使用基本的 Unix 命令行工具，下面是 Doug McIloy 的版本：

tr 'A-Z' 'a-z' | tr -s ' ' '\n' | sort | uniq -c | sort -nr

它非常慢，因为需要对整个文件排序，而不是使用哈希表进行计数（将整个文件读入内存实际上违反了我提出的约束条件）。但是让我惊讶的是，只要把第一个 sort 的地域选项设置为 C（即仅处理 ASCII），就能将速度提高5倍！比其他语言中更好的算法还要快！

而且如果使用 sort -S 2GB 提供更大的缓冲区，还能进一步提高其速度（这里又打破了约束条件）。有意思的是，sort 的 --parallen 选项在这个规模上并没有太大作用。所以优化后的版本如下：

tr 'A-Z' 'a-z' | tr -s ' ' '\n' | LC_ALL=C sort -S 2G | uniq -c | \
    sort -nr

对于小型文件来说这样就足够了。如果我想处理巨型文件，没办法将整个文件读入内存进行排序，那么我会采用 AWK 或 Python 版本。

输出结果与其他语言不太一样，其格式是“空格-前缀-计数-单词”，而不是“单词-计数”，如下所示：

     4 the
      2 foo
      1 defenestration

我们可以加上 awk '{ print $2, $1 }'来改正，但毕竟要使用 awk，而且 awk 有更有效的办法。

其他语言

许多读者给 benhoyt/countwords 代码库贡献了其他流行的语言。感谢你们！

下面是完整的列表：

• Bash: Jesse Hathaway - 不在评测范围内，因为运行时间超过了两分钟

• C#: John Taylor, Yuriy Ostapenko 和Osman Turan

• C# (LINQ): Osman Turan - 不评测

• C++ optimized version: Jussi Pakkanen, Adev, Nathan Myers

• Common Lisp: Brad Svercl

• Crystal: Andrea Manzini

• D: Ross Lonstein

• F#: Yuriy Ostapenko

• Go: Miguel Angel - 对 Go 的优化版本做了简化；Joshua Corbin - 添加了一个并行 Go 版本

• Java: Iulian Pleșoianu

• JavaScript: Dani Biró 和 Flo Hinze

• Julia: Alessandro Melis

• Kotlin: Kazik Pogoda

• Lua: Flemming Madsen

• Nim: csterritt 和 euantorano

• OCaml: doesntgolf

• Pascal: Osman Turan

• Perl: Charles Randall

• PHP: Max Semenik

• Ruby: Bill Mill, 和Niklas

• Rust: Andrew Gallant

• Swift: Daniel Müllenborn

• Tcl: William Ross

• Zig: ifreund ，matu3ba 和ansingh

性能结果和经验

下面是在我的笔记本电脑（64位Linux+SSD）上运行程序的结果。每个测试运行5次，取最短时间作为结果。每次运行都使用如了下命令：

time $PROGRAM <kjvbible_x10.txt >/dev/null

时间单位为秒，所以越小越好。列表按照简单版本的执行时间排序，从快到慢。（注意 grep 和 wc 实际上并没有解决统计问题，只是列出作为比较。）

我认为， 简单版本是最合适的。在实际工作中基本上会选择这些版本。我们从中得到什么结论？下面是我的一些想法：

• 几乎没有人选择优化过的 C 版本，除非你想自己编写一个 GNU wordfreq 工具。因为太容易写错了。如果你想要安全的语言和更快的版本，我推荐选择 Go 或 Rust。

• 如果你只需要快速的解决方案（很多情况下如此），那么 Python 和 Awk 非常合适。

• C++ 模板会产生大量可怕的错误信息，还会在评测报告中产生可怕的函数名，所以评测报告几乎无法阅读。

• 我依然认为这是一个非常好的面试题，尽管我不会期待候选人在白板上写出任何一个优化过的解决方案。

• 我们通常会认为 I/O 很昂贵，但这里的瓶颈并不在 I/O 上。从评测角度来看，文件很可能被缓存了，但即使并非如此，硬盘的读取速度也已经非常快了。真正的瓶颈在于分词和哈希表操作。

这个实验非常有趣！我学到了许多优化的知识，使用了 Valgrind 评测工具，还编写了久违的 Forth 代码。

原文链接：https://benhoyt.com/writings/count-words/

声明：本文由CSDN翻译，转载请注明来源。

60+专家，13个技术领域，CSDN 《IT 人才成长路线图》重磅来袭！

直接扫码或微信搜索「CSDN」公众号，后台回复关键词「路线图」，即可获取完整路线图！