Python开发之迭代器&生成器的实战案例分享

迭代器&生成器

在 Python 中，迭代器和生成器都是用来遍历数据集合的工具，可以按需逐个生成或返回数据，从而避免一次性加载整个数据集合所带来的性能问题和内存消耗问题。

具体来说，迭代器是一个包含 iter() 和 next() 方法的对象，它通过 next() 方法依次返回数据集合中的每个元素，直到没有元素时引发 StopIteration 异常。迭代器可以自定义，也可以使用 Python 内置的可迭代对象，如列表、元组、字典、集合等，以及内置的迭代器函数，如 zip()、map()、filter() 等。

而生成器是一种特殊的迭代器，它使用 yield 关键字来定义，可以在需要时生成数据，从而实现按需生成、惰性计算的效果。生成器可以大大简化代码，提高性能和可读性，也可以通过生成器表达式快速创建简单的生成器。

下面分享几个贴近实际运维开发工作中的场景案例。

实战案例

1.假设需要从一个非常大的数据集合中查找满足特定条件的元素，并且只需要找到第一个符合条件的元素即可停止查找。如果直接遍历整个数据集合，可能会导致性能问题。这时可以使用迭代器和生成器来解决这个问题。

def find_first_match(data, condition):
    for item in data:
        if condition(item):
            return item
    raise ValueError('No matching item found')

large_data = [i for i in range(10000000)]  # 构造一个大数据集合
match = find_first_match(large_data, lambda x: x > 1000)  # 查找第一个大于 1000 的元素
print(match)

2.假设要实现一个函数，它接受一个字符串列表，然后返回这些字符串中所有字符的出现次数。可以使用迭代器和生成器来避免遍历整个列表，并减少内存使用量。

def char_count(words):
    counts = {}
    for word in words:
        for char in word:
            counts[char] = counts.get(char, 0) + 1
    return counts

def char_count_lazy(words):
    def char_gen(words):
        for word in words:
            for char in word:
                yield char
    counts = {}
    for char in char_gen(words):
        counts[char] = counts.get(char, 0) + 1
    return counts

3.需要遍历一个多级嵌套的 JSON 对象，查找其中某个特定的值。假设 JSON 对象很大，因此不能一次性加载到内存中。这时可以使用生成器来逐级遍历 JSON 对象。

def find_value(json_obj, target_key):
    if isinstance(json_obj, dict):
        for key, value in json_obj.items():
            if key == target_key:
                yield value
            else:
                yield from find_value(value, target_key)
    elif isinstance(json_obj, list):
        for item in json_obj:
            yield from find_value(item, target_key)

json_data = {
    "name": "tantianran",
    "age": 30,
    "cars": [
        {"model": "BMW", "year": 2000},
        {"model": "Tesla", "year": 2020}
    ],
    "location": {
        "address": "123 Main St",
        "city": "New York",
        "state": "NY"
    }
}

for value in find_value(json_data, "year"):
    print(value)

4.如果需要读取一个非常大的文件，并对其中的每一行进行处理，但是由于文件太大，无法一次性将整个文件读入内存中。这时可以使用生成器来实现逐行读取文件并逐行处理的操作。

def process_lines(file_name):
    with open(file_name) as f:
        for line in f:
            # 对每一行进行处理，这里只是简单地打印出来
            print(line.strip())

large_file = 'data.txt'
process_lines(large_file)

5.假设有一个大型日志文件，其中包含了数千万行日志记录。需要对这个日志文件进行分析，找出所有包含特定关键字的日志记录，并进行统计。如果直接读取整个日志文件到内存中，可能会导致内存不足的问题。这时可以使用迭代器和生成器来解决这个问题。

def log_file_reader(log_file_path):
    with open(log_file_path) as f:
        for line in f:
            yield line.strip()

def log_analyzer(log_file_path, keyword):
    log_reader = log_file_reader(log_file_path)
    count = 0
    for line in log_reader:
        if keyword in line:
            count += 1
    return count

log_file_path = 'logs.txt'
keyword = 'error'
error_count = log_analyzer(log_file_path, keyword)
print(f'The number of error logs is: {error_count}')

来源：https://mp.weixin.qq.com/s/g9A4rMKscHt8gKJq34sKQw