摘要：使用生成器函數是一個實現管道機制問題，你想以數據管道類似管道的方式迭代處理數據。當這些生成器被連在一起后，每個會將一個多帶帶的數據元素傳遞給迭代處理管道的下一階段。

假設現在有如下業務場景

某生產系統的日志文件如下，并且在持續增加...

[ncms@UPZZGAP02 logs]$ pwd
/home/ncms/ncms/logs
[ncms@UPZZGAP02 logs]$ ll
總用量 797020
-rw-rw-r-- 1 ncms ncms 495465795 11月 30 17:10 ansible.log
-rw-rw-r-- 1 ncms ncms   2251937 11月 30 17:10 celery_beat.log
-rw-rw-r-- 1 ncms ncms     16003 11月 15 10:26 celery_flower.log
-rw-rw-r-- 1 ncms ncms   7042114 11月 30 17:10 celery_worker.log
-rw-r--r-- 1 ncms ncms  24665873 11月 30 17:10 db_error.log
-rw-r--r-- 1 ncms ncms  52428571 11月 28 18:46 db_error.log.1
-rw-r--r-- 1 ncms ncms  52428691 11月 24 06:43 db_error.log.2
-rw-r--r-- 1 ncms ncms  22410652 11月 19 15:16 db_error.log.3
-rw-r--r-- 1 ncms ncms  28064985 11月 30 17:10 db_info.log
-rw-r--r-- 1 ncms ncms  52426630 11月 28 13:29 db_info.log.1
-rw-r--r-- 1 ncms ncms  52427357 11月 24 03:48 db_info.log.2
-rw-r--r-- 1 ncms ncms  24276767 11月 19 15:16 db_info.log.3
-rw-rw-r-- 1 ncms ncms     42490 11月 30 13:06 ncms_access.log
-rw-rw-r-- 1 ncms ncms     24072 10月 30 15:33 ncms_error.log
-rw-rw-r-- 1 ncms ncms   1350318 11月 30 16:38 nginx_access.log
-rw-rw-r-- 1 ncms ncms      1685 11月  7 18:15 nginx_error.log
-rw-rw-r-- 1 ncms ncms     24001 11月 15 10:27 supervisord.log
-rw-rw-r-- 1 ncms ncms    645742 11月 30 16:38 uwsgi.log
[ncms@UPZZGAP02 logs]$ du -sh *
473M    ansible.log
2.2M    celery_beat.log
16K    celery_flower.log
6.8M    celery_worker.log
24M    db_error.log
51M    db_error.log.1
51M    db_error.log.2
22M    db_error.log.3
27M    db_info.log
51M    db_info.log.1
51M    db_info.log.2
24M    db_info.log.3
44K    ncms_access.log
24K    ncms_error.log
1.3M    nginx_access.log
4.0K    nginx_error.log
24K    supervisord.log
632K    uwsgi.log
[ncms@UPZZGAP02 logs]$

其中有應用、數據庫、Celery、Nginx、uwsgi、supervisord、Ansible的日志，Ansible.log有473M，未來很定會更大。現在需要使用某些關鍵字對日志進行查找分析，應該如何做？

最簡單粗暴的方式就是使用grep之類的命令，遞歸查找所有的.log文件，但這樣會耗費大量內存，影響機器性能。

可以考慮使用數據管道 (類似 Unix 管道) 的方式迭代處理數據。使用Python生成器函數是一個實現管道機制

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# __author__ = "liao gao xiang"

import os
import fnmatch
import gzip
import bz2
import re

# 問題，你想以數據管道 (類似 Unix 管道) 的方式迭代處理數據。比如，你有個大量的數據
# 需要處理，但是不能將它們一次性放入內存中。可以使用生成器實現數據處理管道
""" 文件格式如下
foo/
access-log-012007.gz
access-log-022007.gz
access-log-032007.gz
...
access-log-012008
bar/
access-log-092007.bz2
...
access-log-022008
"""


def gen_find(filepat, top):
    """
    查找符合Shell正則匹配的目錄樹下的所有文件名
    :param filepat: shell正則
    :param top: 目錄路徑
    :return: 文件絕對路徑生成器
    """
    for path, _, filenames in os.walk(top):
        for file in fnmatch.filter(filenames, filepat):
            yield os.path.join(path, file)


def gen_opener(filenames):
    """
    每打開一個文件生成就生成一個文件對象，調用下一個迭代前關閉文件
    :param filenames: 多個文件絕對路徑組成的可迭代對象
    :return: 文件對象生成器
    """
    for filename in filenames:
        if filename.endswith(".gz"):
            f = gzip.open(filename, "r", encoding="utf-8")
        elif filename.endswith(".bz2"):
            f = bz2.open(filename, "r", encoding="utf-8")
        else:
            f = open(filename, "r", encoding="utf-8")
        yield f
        f.close()


def gen_concatenate(iterators):
    """
    將輸入序列拼接成一個很長的行序列。
    :param iterators:
    :return: 返回生成器所產生的所有值
    """
    for it in iterators:
        yield from it


def gen_grep(pattern, lines):
    """
    使用正則匹配行
    :param pattern: 正則匹配
    :param lines: 多行
    :return: 結果生成器
    """
    pat = re.compile(pattern)
    for n, line in enumerate(lines, start=1):
        if pat.search(line):
            yield n, line


if __name__ == "__main__":
    filenames = gen_find("*.log", "/home/ncms/ncms/logs")
    files = gen_opener(filenames)
    lines = gen_concatenate(files)
    user_action = gen_grep("(?i)liaogaoxiang_kd", lines)

    for n, line in user_action:
        print(line)

查詢包含用戶 liaogaoxiang_kd 的所有記錄，數據結果如下：

[views:post]:2018-11-07 18:13:09.841490 -users- liaogaoxiang_kd登錄成功！

[views:get]:2018-11-07 18:16:04.681519 -users- liaogaoxiang_kd訪問了用戶信息列表

[views:post]:2018-11-07 18:16:23.866700 -users- liaogaoxiang_kd編輯了用戶的信息

[views:get]:2018-11-07 18:16:23.878949 -users- liaogaoxiang_kd訪問了用戶信息列表

[views:get]:2018-11-07 18:16:25.641090 -users- liaogaoxiang_kd訪問了用戶信息列表

[views:post]:2018-11-07 18:16:42.671377 -users- liaogaoxiang_kd編輯了用戶的信息

[views:get]:2018-11-07 18:16:42.719873 -users- liaogaoxiang_kd訪問了用戶信息列表

[views:post]:2018-11-08 11:17:42.627693 -users- liaogaoxiang_kd登錄成功！

如需查詢其它錯誤信息，只需替換gen_grep("(?i)liaogaoxiang_kd", lines)中的關鍵字即可！以管道方式處理數據可以用來解決各類其他問題，包括解析，讀取實時數據，定時 輪詢等。

為了理解上述代碼，重點是要明白 yield 語句作為數據的生產者而 for 循環語句 作為數據的消費者。當這些生成器被連在一起后，每個 yield 會將一個多帶帶的數據元 素傳遞給迭代處理管道的下一階段。這種方式一個非常好的特點是每個生成器函數很小并且都是獨立的。這樣的話就 很容易編寫和維護。很多時候，這些函數如果比較通用的話可以在其他場景重復使用。并且最終將這些組件組合起來的代碼看上去非常簡單，也很容易理解。

使用這種方式的內存效率很高。上述代碼即便是在一個超大型文件目錄中 也能工作的很好。事實上，由于使用了迭代方式處理，代碼運行過程中只需要很小很小的內存。 在調用 gen_concatenate() 函數的時候你可能會有些不太明白。這個函數的目的是將輸入序列拼接成一個很長的行序列。 itertools.chain() 函數同樣有類似的功能， 但是它需要將所有可迭代對象最為參數傳入。在上面這個例子中，你可能會寫類似這樣 的語句 lines = itertools.chain(*files) ，這將導致 gen_opener() 生成器被提前全部消費掉。但由于 gen_opener() 生成器每次生成一個打開過的文件，等到下一個迭 代步驟時文件就關閉了，因此 chain() 在這里不能這樣使用。上面的方案可以避免這 種情況。

gen_concatenate() 函數中出現過 yield from 語句，它將 yield 操作代理到父 生成器上去。語句 yield from it 簡單的返回生成器 it 所產生的所有值。

程序員交流群，干貨分享，加我拉你入群。