JavaScript Generator与异步迭代器实战：构建可暂停的异步流程控制 | 前端进阶

一、Generator函数的革命性意义

在JavaScript异步编程的发展历程中，Generator函数是一个重要的里程碑。它首次引入了”可暂停执行”的概念，为后续的async/await语法奠定了理论基础。与传统的回调函数和Promise相比，Generator提供了更直观、更可控的异步流程管理方式。

Generator的核心价值在于：它允许函数在执行过程中多次进入和退出，保持执行上下文的状态。这种特性使得我们可以用同步的方式编写异步代码，极大地提高了代码的可读性和可维护性。

二、Generator基础：从入门到精通

1. 基本语法与执行机制

// 定义Generator函数
function* basicGenerator() {
    console.log('开始执行');
    yield '第一个暂停点';
    console.log('继续执行');
    yield '第二个暂停点';
    console.log('执行结束');
    return '最终返回值';
}

// 使用Generator
const gen = basicGenerator();

console.log(gen.next()); // { value: '第一个暂停点', done: false }
console.log(gen.next()); // { value: '第二个暂停点', done: false }
console.log(gen.next()); // { value: '最终返回值', done: true }

2. 双向数据传递

function* twoWayCommunication() {
    const name = yield '请输入您的姓名：';
    console.log(`收到姓名：${name}`);
    
    const age = yield `您好 ${name}，请输入您的年龄：`;
    console.log(`收到年龄：${age}`);
    
    return `用户信息：${name}, ${age}岁`;
}

const communicator = twoWayCommunication();

console.log(communicator.next().value); // 请输入您的姓名：
console.log(communicator.next('张三').value); // 您好 张三，请输入您的年龄：
console.log(communicator.next(25).value); // 用户信息：张三, 25岁

3. 错误处理机制

function* errorHandlingGenerator() {
    try {
        yield '第一步';
        yield '第二步';
        throw new Error('模拟错误');
    } catch (error) {
        console.log(`捕获错误：${error.message}`);
        yield '错误恢复执行';
    }
}

const errorGen = errorHandlingGenerator();
console.log(errorGen.next()); // { value: '第一步', done: false }
console.log(errorGen.next()); // { value: '第二步', done: false }
console.log(errorGen.throw(new Error('外部抛出错误'))); // 捕获错误：外部抛出错误
console.log(errorGen.next()); // { value: '错误恢复执行', done: false }

三、异步迭代器：处理异步数据流

1. 异步Generator函数

async function* asyncDataGenerator() {
    // 模拟异步数据获取
    const users = await fetchUsers();
    for (const user of users) {
        // 模拟每个用户的详细信息获取
        const details = await fetchUserDetails(user.id);
        yield details;
    }
}

// 使用for await...of循环
async function processUsers() {
    for await (const userDetails of asyncDataGenerator()) {
        console.log('处理用户:', userDetails);
    }
}

2. 自定义异步可迭代对象

class AsyncDataStream {
    constructor(dataSource) {
        this.dataSource = dataSource;
        this.index = 0;
    }
    
    async *[Symbol.asyncIterator]() {
        while (this.index  {
            setTimeout(() => {
                resolve({ id, data: `项目${id}的数据` });
            }, 100);
        });
    }
}

// 使用示例
async function main() {
    const stream = new AsyncDataStream([1, 2, 3, 4, 5]);
    
    for await (const item of stream) {
        console.log('接收到数据:', item);
    }
}

四、实战案例：构建智能任务调度系统

1. 任务调度器核心类

class TaskScheduler {
    constructor(maxConcurrent = 3) {
        this.maxConcurrent = maxConcurrent;
        this.runningTasks = new Set();
        this.taskQueue = [];
        this.isPaused = false;
    }
    
    // 添加任务
    addTask(task) {
        return new Promise((resolve, reject) => {
            this.taskQueue.push({
                task,
                resolve,
                reject
            });
            this.processQueue();
        });
    }
    
    // 处理任务队列
    async processQueue() {
        if (this.isPaused || this.runningTasks.size >= this.maxConcurrent) {
            return;
        }
        
        if (this.taskQueue.length === 0) {
            return;
        }
        
        const { task, resolve, reject } = this.taskQueue.shift();
        const taskPromise = task();
        
        this.runningTasks.add(taskPromise);
        
        try {
            const result = await taskPromise;
            resolve(result);
        } catch (error) {
            reject(error);
        } finally {
            this.runningTasks.delete(taskPromise);
            this.processQueue(); // 继续处理下一个任务
        }
    }
    
    // 暂停调度器
    pause() {
        this.isPaused = true;
    }
    
    // 恢复调度器
    resume() {
        this.isPaused = false;
        this.processQueue();
    }
    
    // 获取状态
    getStatus() {
        return {
            running: this.runningTasks.size,
            queued: this.taskQueue.length,
            paused: this.isPaused
        };
    }
}

2. Generator控制的任务流程

function* complexWorkflow() {
    console.log('🚀 开始复杂工作流程');
    
    // 第一阶段：数据准备
    const dataPrepResult = yield {
        type: 'parallel',
        tasks: [
            () => fetchUserData(),
            () => fetchProductData(),
            () => fetchConfigData()
        ]
    };
    console.log('✅ 数据准备完成:', dataPrepResult);
    
    // 第二阶段：数据处理
    const processedData = yield {
        type: 'sequence',
        tasks: [
            () => validateData(dataPrepResult),
            () => transformData(dataPrepResult),
            () => enrichData(dataPrepResult)
        ]
    };
    console.log('✅ 数据处理完成:', processedData);
    
    // 第三阶段：结果输出
    const finalResult = yield {
        type: 'single',
        task: () => generateReport(processedData)
    };
    console.log('✅ 报告生成完成');
    
    return finalResult;
}

// 工作流执行器
class WorkflowExecutor {
    constructor(scheduler) {
        this.scheduler = scheduler;
    }
    
    async execute(workflowGenerator) {
        const gen = workflowGenerator();
        let result = { value: undefined, done: false };
        
        while (!result.done) {
            const step = result.value;
            
            if (step.type === 'parallel') {
                // 并行执行任务
                const tasks = step.tasks.map(task => this.scheduler.addTask(task));
                result = gen.next(await Promise.all(tasks));
            } else if (step.type === 'sequence') {
                // 顺序执行任务
                const results = [];
                for (const task of step.tasks) {
                    results.push(await this.scheduler.addTask(task));
                }
                result = gen.next(results);
            } else if (step.type === 'single') {
                // 单个任务
                result = gen.next(await this.scheduler.addTask(step.task));
            } else {
                result = gen.next();
            }
        }
        
        return result.value;
    }
}

3. 模拟任务函数

// 模拟异步任务
const simulateAsyncTask = (name, duration, shouldFail = false) => {
    return () => new Promise((resolve, reject) => {
        console.log(`🕒 开始任务: ${name}`);
        setTimeout(() => {
            if (shouldFail) {
                console.log(`❌ 任务失败: ${name}`);
                reject(new Error(`${name} 执行失败`));
            } else {
                console.log(`✅ 任务完成: ${name}`);
                resolve(`${name} 的结果`);
            }
        }, duration);
    });
};

// 具体任务实现
const fetchUserData = simulateAsyncTask('获取用户数据', 1000);
const fetchProductData = simulateAsyncTask('获取产品数据', 1500);
const fetchConfigData = simulateAsyncTask('获取配置数据', 800);

const validateData = simulateAsyncTask('验证数据', 600);
const transformData = simulateAsyncTask('转换数据', 1200);
const enrichData = simulateAsyncTask('丰富数据', 900);

const generateReport = simulateAsyncTask('生成报告', 2000);

4. 完整系统演示

async function demonstrateWorkflow() {
    console.log('🎯 开始演示智能任务调度系统');
    
    // 创建调度器和执行器
    const scheduler = new TaskScheduler(2); // 最大并发数2
    const executor = new WorkflowExecutor(scheduler);
    
    try {
        const finalResult = await executor.execute(complexWorkflow);
        console.log('🎊 工作流执行完成:', finalResult);
    } catch (error) {
        console.error('💥 工作流执行失败:', error);
    }
    
    // 监控状态
    setInterval(() => {
        const status = scheduler.getStatus();
        console.log('📊 调度器状态:', status);
    }, 1000);
}

// 执行演示
demonstrateWorkflow();

五、高级应用：协程与状态管理

1. 协程通信系统

class CoroutineManager {
    constructor() {
        this.coroutines = new Map();
        this.messageQueue = [];
    }
    
    // 创建协程
    createCoroutine(id, generatorFunc) {
        const gen = generatorFunc();
        this.coroutines.set(id, {
            generator: gen,
            status: 'running'
        });
        
        this.executeCoroutine(id);
        return id;
    }
    
    // 执行协程
    async executeCoroutine(id) {
        const coroutine = this.coroutines.get(id);
        if (!coroutine) return;
        
        try {
            let result = coroutine.generator.next();
            
            while (!result.done) {
                // 处理yield的值
                if (result.value && result.value.type === 'waitForMessage') {
                    // 等待消息
                    const message = await this.waitForMessage(id, result.value.messageType);
                    result = coroutine.generator.next(message);
                } else if (result.value && result.value.type === 'sendMessage') {
                    // 发送消息
                    this.sendMessage(result.value.target, result.value.message);
                    result = coroutine.generator.next();
                } else {
                    // 其他类型的yield值
                    result = coroutine.generator.next();
                }
            }
            
            coroutine.status = 'completed';
            console.log(`协程 ${id} 执行完成`);
            
        } catch (error) {
            coroutine.status = 'error';
            console.error(`协程 ${id} 执行错误:`, error);
        }
    }
    
    // 等待消息
    waitForMessage(coroutineId, messageType) {
        return new Promise((resolve) => {
            this.messageQueue.push({
                coroutineId,
                messageType,
                resolve
            });
        });
    }
    
    // 发送消息
    sendMessage(targetCoroutineId, message) {
        const pendingMessage = this.messageQueue.find(
            msg => msg.coroutineId === targetCoroutineId && 
                   msg.messageType === message.type
        );
        
        if (pendingMessage) {
            pendingMessage.resolve(message);
            this.messageQueue = this.messageQueue.filter(msg => msg !== pendingMessage);
        }
    }
}

2. 协程示例：生产者-消费者模式

function* producerCoroutine() {
    let count = 0;
    
    while (count  setTimeout(resolve, 500));
        
        yield {
            type: 'sendMessage',
            target: 'producer',
            message: { type: 'acknowledge', data: `已处理${receivedCount}个产品` }
        };
    }
}

// 启动协程系统
const manager = new CoroutineManager();
manager.createCoroutine('producer', producerCoroutine);
manager.createCoroutine('consumer', consumerCoroutine);

六、性能优化与最佳实践

1. 内存管理

class MemoryAwareGenerator {
    constructor() {
        this.dataCache = new WeakMap();
        this.cleanupInterval = setInterval(() => {
            this.cleanup();
        }, 30000); // 30秒清理一次
    }
    
    // 使用WeakMap存储临时数据
    *memoryEfficientProcess(largeDataset) {
        for (const item of largeDataset) {
            // 处理数据，避免闭包引用
            const processed = this.processItem(item);
            yield processed;
            
            // 显式清理
            this.cleanupItem(item);
        }
    }
    
    processItem(item) {
        // 模拟数据处理
        return { ...item, processed: true };
    }
    
    cleanupItem(item) {
        // 清理不再需要的数据
        // 在实际应用中可能需要更复杂的清理逻辑
    }
    
    cleanup() {
        // 定期清理资源
        console.log('执行内存清理...');
    }
    
    destroy() {
        clearInterval(this.cleanupInterval);
    }
}

2. 错误恢复策略

function* resilientWorkflow() {
    let retryCount = 0;
    const maxRetries = 3;
    
    while (retryCount  maxRetries) {
                console.log('❌ 达到最大重试次数，放弃操作');
                throw error;
            }
            
            // 等待指数退避
            const backoffTime = Math.pow(2, retryCount) * 1000;
            yield {
                type: 'wait',
                duration: backoffTime
            };
        }
    }
}

七、总结与展望

Generator函数和异步迭代器为JavaScript带来了全新的编程范式。通过本文的实战案例，我们深入探讨了如何利用这些特性构建复杂的异步流程控制系统。

核心优势总结：

• Generator提供了可暂停执行的函数，使得复杂流程控制变得直观
• 异步迭代器使得处理异步数据流更加优雅
• 协程模式可以实现复杂的多任务协作
• 任务调度系统能够有效管理资源利用率
• 错误恢复机制提高了系统的健壮性

在实际项目中，这些技术特别适用于：

• 复杂工作流引擎的实现
• 大数据处理管道
• 实时数据流处理
• 游戏开发中的状态管理
• 微服务间的协调通信