从基础到高级详解Python多容器迭代完全指南

# 列表迭代
fruits = ['apple', 'banana', 'cherry']
print("列表迭代:")
for fruit in fruits:
    print(fruit)

# 元组迭代
colors = ('red', 'green', 'blue')
print("\n元组迭代:")
for color in colors:
    print(color)

# 集合迭代
unique_nums = {1, 2, 3, 4, 5}
print("\n集合迭代:")
for num in unique_nums:
    print(num)

# 字典迭代
person = {'name': 'Alice', 'age': 30, 'city': 'New York'}
print("\n字典迭代:")
for key, value in person.items():
    print(f"{key}: {value}")

# 文本文件迭代
print("文件迭代:")
with open('data.txt', 'r') as file:
    for line_num, line in enumerate(file, 1):
        print(f"行 {line_num}: {line.strip()}")

# CSV文件迭代
import csv
print("\nCSV文件迭代:")
with open('data.csv', 'r') as csvfile:
    reader = csv.DictReader(csvfile)
    for row in reader:
        print(f"记录: {row}")

def universal_iterator(container):
    """统一容器迭代接口"""
    if isinstance(container, dict):
        return container.items()
    elif isinstance(container, (list, tuple, set)):
        return enumerate(container)
    elif hasattr(container, '__iter__'):
        return container
    else:
        raise TypeError("不支持的类型")

# 使用示例
data_structures = [
    ['a', 'b', 'c'],
    ('x', 'y', 'z'),
    {'name': 'Alice', 'age': 30},
    {1, 2, 3}
]

print("统一容器迭代:")
for container in data_structures:
    print(f"\n容器类型: {type(container).__name__}")
    for key, value in universal_iterator(container):
        print(f"  {key}: {value}")

def recursive_iter(container):
    """递归迭代嵌套容器"""
    if isinstance(container, (list, tuple, set)):
        for item in container:
            yield from recursive_iter(item)
    elif isinstance(container, dict):
        for key, value in container.items():
            yield (key, value)
            yield from recursive_iter(value)
    else:
        yield container

# 使用示例
nested_data = {
    'name': 'Alice',
    'scores': [90, 85, [95, 92]],
    'address': {
        'city': 'New York',
        'zipcodes': (10001, 10002)
    }
}

print("递归迭代嵌套容器:")
for item in recursive_iter(nested_data):
    print(item)

class TreeNode:
    """树节点容器"""
    def __init__(self, value):
        self.value = value
        self.children = []
    
    def add_child(self, node):
        self.children.append(node)
    
    def __iter__(self):
        """深度优先迭代器"""
        yield self.value
        for child in self.children:
            yield from child

# 使用示例
root = TreeNode('A')
b = TreeNode('B')
c = TreeNode('C')
d = TreeNode('D')
root.add_child(b)
root.add_child(c)
b.add_child(d)

print("树容器迭代:")
for value in root:
    print(value)  # A, B, D, C

class Graph:
    """图容器"""
    def __init__(self):
        self.nodes = {}
    
    def add_node(self, name):
        self.nodes[name] = []
    
    def add_edge(self, src, dest):
        self.nodes[src].append(dest)
    
    def __iter__(self):
        """广度优先迭代器"""
        from collections import deque
        visited = set()
        queue = deque()
        
        for node in self.nodes:
            if node not in visited:
                visited.add(node)
                queue.append(node)
                while queue:
                    current = queue.popleft()
                    yield current
                    for neighbor in self.nodes[current]:
                        if neighbor not in visited:
                            visited.add(neighbor)
                            queue.append(neighbor)

# 使用示例
graph = Graph()
graph.add_node('A')
graph.add_node('B')
graph.add_node('C')
graph.add_node('D')
graph.add_edge('A', 'B')
graph.add_edge('A', 'C')
graph.add_edge('B', 'D')
graph.add_edge('C', 'D')

print("图容器迭代:")
for node in graph:
    print(node)  # A, B, C, D

import sqlite3

def sqlite_iterator(db_path, query):
    """SQLite数据库迭代器"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute(query)
    
    while True:
        row = cursor.fetchone()
        if row is None:
            break
        yield row
    
    cursor.close()
    conn.close()

# 使用示例
print("SQLite迭代:")
for row in sqlite_iterator('example.db', 'SELECT * FROM users'):
    print(row)

from pymongo import MongoClient

def mongo_iterator(collection_name, query={}):
    """MongoDB迭代器"""
    client = MongoClient('localhost', 27017)
    db = client['test_db']
    collection = db[collection_name]
    
    cursor = collection.find(query)
    for document in cursor:
        yield document
    
    client.close()

# 使用示例
print("MongoDB迭代:")
for doc in mongo_iterator('users', {'age': {'$gt': 30}}):
    print(doc)

import threading
from queue import Queue

class ThreadSafeQueue:
    """线程安全队列容器"""
    def __init__(self):
        self.queue = Queue()
        self.lock = threading.Lock()
    
    def put(self, item):
        with self.lock:
            self.queue.put(item)
    
    def __iter__(self):
        """线程安全迭代器"""
        while not self.queue.empty():
            with self.lock:
                if not self.queue.empty():
                    yield self.queue.get()
                else:
                    break

# 使用示例
def producer(queue):
    """生产者线程"""
    for i in range(5):
        queue.put(f"Item-{i}")
        print(f"生产: Item-{i}")

def consumer(queue):
    """消费者线程"""
    for item in queue:
        print(f"消费: {item}")

print("线程安全队列迭代:")
ts_queue = ThreadSafeQueue()

# 创建线程
prod_thread = threading.Thread(target=producer, args=(ts_queue,))
cons_thread = threading.Thread(target=consumer, args=(ts_queue,))

# 启动线程
prod_thread.start()
cons_thread.start()

# 等待完成
prod_thread.join()
cons_thread.join()

import multiprocessing

def process_safe_iterator(container):
    """进程安全迭代器"""
    manager = multiprocessing.Manager()
    shared_list = manager.list(container)
    
    def worker(items):
        for item in items:
            print(f"处理: {item}")
    
    processes = []
    # 分割数据
    chunk_size = len(shared_list) // 4
    for i in range(4):
        start = i * chunk_size
        end = (i+1) * chunk_size if i < 3 else len(shared_list)
        chunk = shared_list[start:end]
        p = multiprocessing.Process(target=worker, args=(chunk,))
        processes.append(p)
        p.start()
    
    for p in processes:
        p.join()

# 使用示例
data = list(range(100))
print("多进程容器迭代:")
process_safe_iterator(data)

def chunked_file_iterator(file_path, chunk_size=1024):
    """大文件分块迭代"""
    with open(file_path, 'r') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            yield chunk

# 使用示例
print("大文件分块迭代:")
for i, chunk in enumerate(chunked_file_iterator('large_file.txt', 4096)):
    print(f"块 {i+1}: {len(chunk)}字符")

import mmap

def mmap_iterator(file_path):
    """内存映射文件迭代"""
    with open(file_path, 'r') as f:
        with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
            start = 0
            while start < len(mm):
                # 查找行尾
                end = mm.find(b'\n', start)
                if end == -1:
                    end = len(mm)
                line = mm[start:end].decode('utf-8')
                yield line
                start = end + 1

# 使用示例
print("内存映射迭代:")
for line in mmap_iterator('large_file.txt'):
    if 'error' in line:
        print(f"发现错误行: {line}")

class RangeIterator:
    """自定义范围迭代器"""
    def __init__(self, start, end, step=1):
        self.current = start
        self.end = end
        self.step = step
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.current >= self.end:
            raise StopIteration
        value = self.current
        self.current += self.step
        return value

# 使用示例
print("自定义迭代器:")
for num in RangeIterator(5, 15, 2):
    print(num)  # 5, 7, 9, 11, 13

def fibonacci_container(n):
    """斐波那契数列容器"""
    a, b = 0, 1
    count = 0
    while count < n:
        yield a
        a, b = b, a + b
        count += 1

# 使用示例
print("生成器容器:")
fib = fibonacci_container(10)
for num in fib:
    print(num)  # 0, 1, 1, 2, 3, 5, 8, 13, 21, 34

class Stack:
    """栈容器"""
    def __init__(self):
        self.items = []
    
    def push(self, item):
        self.items.append(item)
    
    def pop(self):
        return self.items.pop()
    
    def __iter__(self):
        """从顶到底迭代"""
        return reversed(self.items)

# 使用示例
stack = Stack()
stack.push(1)
stack.push(2)
stack.push(3)

print("栈迭代:")
for item in stack:
    print(item)  # 3, 2, 1

from collections import deque

class Queue:
    """队列容器"""
    def __init__(self):
        self.items = deque()
    
    def enqueue(self, item):
        self.items.append(item)
    
    def dequeue(self):
        return self.items.popleft()
    
    def __iter__(self):
        """先进先出迭代"""
        return iter(self.items)

# 使用示例
queue = Queue()
queue.enqueue('A')
queue.enqueue('B')
queue.enqueue('C')

print("队列迭代:")
for item in queue:
    print(item)  # A, B, C

def process_container(container):
    """统一处理各种容器"""
    if isinstance(container, dict):
        items = container.items()
    elif hasattr(container, '__iter__'):
        items = enumerate(container)
    else:
        raise TypeError("不支持的类型")

    for key, value in items:
        process_item(key, value)

# 大数据使用生成器
def large_data_iter():
    with open('huge_file.txt') as f:
        for line in f:
            yield line

# 避免加载整个容器
# 错误做法
big_list = list(range(1000000))
for item in big_list:
    ...

# 正确做法
for item in range(1000000):
    ...

def safe_container_iter(container):
    """安全的容器迭代"""
    try:
        for item in container:
            try:
                process(item)
            except Exception as e:
                print(f"处理错误: {e}")
    except TypeError:
        print("对象不可迭代")

# 使用局部变量加速
def optimized_iter(container):
    """优化迭代性能"""
    items = container.items() if isinstance(container, dict) else container
    iter_func = items.__iter__
    next_func = items.__next__
    while True:
        try:
            item = next_func()
            process(item)
        except StopIteration:
            break

class ConcurrentContainer:
    """并发安全容器"""
    def __init__(self):
        self.data = []
        self.lock = threading.Lock()

    def add(self, item):
        with self.lock:
            self.data.append(item)

    def __iter__(self):
        """线程安全迭代器"""
        with self.lock:
            copy = self.data[:]
        return iter(copy)

class CustomContainer:
    """
    自定义容器类

    迭代行为:
    按添加顺序迭代元素
    支持嵌套迭代

    示例:
        c = CustomContainer()
        c.add(1)
        c.add(2)
        for item in c:
            print(item)
    """
    def __init__(self):
        self.items = []

    def add(self, item):
        self.items.append(item)

    def __iter__(self):
        return iter(self.items)