JavaScript Generator函数与协程编程：异步流程控制的革命性解决方案

一、Generator函数基础概念

1.1 什么是Generator函数？

Generator函数是ES6引入的一种特殊函数，它可以在执行过程中多次暂停和恢复，通过yield关键字控制执行流程。这种特性使得Generator成为处理异步操作和复杂迭代逻辑的强大工具。

1.2 Generator的核心特性

• 可暂停执行：函数执行可以在yield语句处暂停
• 状态保持：暂停时函数内部状态被完整保存
• 双向通信：可以通过next()方法向函数内部传递值
• 迭代器接口：天然实现迭代器协议

二、语法详解与执行机制

2.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.2 执行流程详解

function* executionFlow() {
    console.log('步骤1');
    const result1 = yield '暂停点1';
    console.log('步骤2，收到外部值:', result1);
    const result2 = yield '暂停点2';
    console.log('步骤3，收到外部值:', result2);
    return '完成';
}

const flow = executionFlow();

// 第一次调用next()，执行到第一个yield
const step1 = flow.next(); 
// 输出: 步骤1
// step1: { value: '暂停点1', done: false }

// 第二次调用next()，传入值给result1
const step2 = flow.next('外部数据1');
// 输出: 步骤2，收到外部值: 外部数据1
// step2: { value: '暂停点2', done: false }

// 第三次调用next()，传入值给result2
const step3 = flow.next('外部数据2');
// 输出: 步骤3，收到外部值: 外部数据2
// step3: { value: '完成', done: true }
            

三、迭代器协议深度解析

3.1 自定义迭代器实现

// 使用Generator实现自定义迭代器
function* rangeGenerator(start, end, step = 1) {
    let current = start;
    while (current <= end) {
        yield current;
        current += step;
    }
}

// 使用for...of循环迭代
for (let num of rangeGenerator(1, 5)) {
    console.log(num); // 依次输出: 1, 2, 3, 4, 5
}

// 手动控制迭代
const range = rangeGenerator(10, 15, 2);
console.log(range.next().value); // 10
console.log(range.next().value); // 12
console.log(range.next().value); // 14
console.log(range.next().done);  // true
            

3.2 无限序列生成

// 无限斐波那契数列生成器
function* fibonacciGenerator() {
    let [prev, curr] = [0, 1];
    while (true) {
        yield curr;
        [prev, curr] = [curr, prev + curr];
    }
}

const fibonacci = fibonacciGenerator();

// 生成前10个斐波那契数
for (let i = 0; i < 10; i++) {
    console.log(fibonacci.next().value);
}
// 输出: 1, 1, 2, 3, 5, 8, 13, 21, 34, 55

// 惰性求值：只在需要时计算
function* primeGenerator() {
    function isPrime(num) {
        for (let i = 2; i  1;
    }
    
    let num = 2;
    while (true) {
        if (isPrime(num)) {
            yield num;
        }
        num++;
    }
}

const primes = primeGenerator();
console.log(primes.next().value); // 2
console.log(primes.next().value); // 3
console.log(primes.next().value); // 5
            

四、异步编程实战应用

4.1 基于Generator的异步流程控制

// 模拟异步操作
function asyncOperation(data, delay) {
    return new Promise(resolve => {
        setTimeout(() => {
            console.log('异步操作完成:', data);
            resolve(data * 2);
        }, delay);
    });
}

// Generator异步流程控制函数
function asyncGenerator(generatorFunc) {
    const generator = generatorFunc();
    
    function handle(result) {
        if (result.done) return Promise.resolve(result.value);
        
        return Promise.resolve(result.value)
            .then(res => handle(generator.next(res)))
            .catch(err => handle(generator.throw(err)));
    }
    
    return handle(generator.next());
}

// 使用Generator管理异步流程
function* complexAsyncFlow() {
    try {
        console.log('开始复杂异步流程');
        
        const result1 = yield asyncOperation(10, 1000);
        console.log('第一阶段结果:', result1);
        
        const result2 = yield asyncOperation(result1, 500);
        console.log('第二阶段结果:', result2);
        
        const result3 = yield asyncOperation(result2, 800);
        console.log('最终结果:', result3);
        
        return result3;
    } catch (error) {
        console.error('流程执行失败:', error);
        throw error;
    }
}

// 执行异步流程
asyncGenerator(complexAsyncFlow)
    .then(finalResult => {
        console.log('异步流程执行完成:', finalResult);
    })
    .catch(error => {
        console.error('执行出错:', error);
    });
            

4.2 错误处理机制

function* errorHandlingDemo() {
    try {
        const data = yield Promise.resolve('正常数据');
        console.log('收到数据:', data);
        
        // 模拟可能出错的操作
        const riskyData = yield riskyOperation();
        console.log('风险操作结果:', riskyData);
        
    } catch (error) {
        console.error('Generator内部捕获错误:', error.message);
        // 可以进行错误恢复或重试逻辑
        yield errorRecovery();
    }
}

function riskyOperation() {
    return Math.random() > 0.5 
        ? Promise.resolve('成功') 
        : Promise.reject(new Error('随机失败'));
}

function errorRecovery() {
    return Promise.resolve('错误恢复后的数据');
}

// 增强的异步执行器
function advancedAsyncRunner(generatorFunc) {
    return new Promise((resolve, reject) => {
        const generator = generatorFunc();
        
        function step(nextFn) {
            let next;
            try {
                next = nextFn();
            } catch (error) {
                return reject(error);
            }
            
            if (next.done) {
                return resolve(next.value);
            }
            
            Promise.resolve(next.value)
                .then(
                    value => step(() => generator.next(value)),
                    error => step(() => generator.throw(error))
                );
        }
        
        step(() => generator.next());
    });
}

// 使用增强执行器
advancedAsyncRunner(errorHandlingDemo)
    .then(result => console.log('执行完成:', result))
    .catch(error => console.error('执行失败:', error));
            

五、协程编程模式实现

5.1 协程调度器实现

class CoroutineScheduler {
    constructor() {
        this.coroutines = new Map();
        this.isRunning = false;
    }
    
    // 添加协程任务
    addCoroutine(name, generatorFunc, ...args) {
        const coroutine = {
            generator: generatorFunc(...args),
            state: 'ready',
            priority: 0
        };
        this.coroutines.set(name, coroutine);
        return this;
    }
    
    // 启动调度器
    start() {
        if (this.isRunning) return;
        this.isRunning = true;
        this.schedule();
    }
    
    // 协程调度算法
    async schedule() {
        while (this.isRunning && this.coroutines.size > 0) {
            for (const [name, coroutine] of this.coroutines) {
                if (coroutine.state === 'ready') {
                    coroutine.state = 'running';
                    
                    try {
                        const result = coroutine.generator.next();
                        
                        if (result.done) {
                            console.log(`协程 ${name} 执行完成`);
                            this.coroutines.delete(name);
                        } else {
                            // 处理yield的值
                            if (result.value instanceof Promise) {
                                coroutine.state = 'waiting';
                                result.value
                                    .then(() => {
                                        coroutine.state = 'ready';
                                        this.schedule();
                                    })
                                    .catch(error => {
                                        console.error(`协程 ${name} 出错:`, error);
                                        this.coroutines.delete(name);
                                    });
                            } else {
                                coroutine.state = 'ready';
                            }
                        }
                    } catch (error) {
                        console.error(`协程 ${name} 执行异常:`, error);
                        this.coroutines.delete(name);
                    }
                }
            }
            
            // 避免阻塞事件循环
            if (Array.from(this.coroutines.values()).some(c => c.state === 'ready')) {
                await new Promise(resolve => setTimeout(resolve, 0));
            } else {
                await new Promise(resolve => setTimeout(resolve, 10));
            }
        }
        
        this.isRunning = false;
        console.log('所有协程执行完成');
    }
}

// 使用协程调度器
const scheduler = new CoroutineScheduler();

// 定义多个协程任务
function* workerCoroutine(name, workload) {
    for (let i = 1; i  setTimeout(resolve, 100)); // 模拟异步工作
    }
    return `${name} 工作完成`;
}

function* monitorCoroutine() {
    while (true) {
        console.log('监控协程: 系统运行正常');
        yield new Promise(resolve => setTimeout(resolve, 1000));
    }
}

// 添加协程任务
scheduler.addCoroutine('worker1', workerCoroutine, '工作协程1', 5);
scheduler.addCoroutine('worker2', workerCoroutine, '工作协程2', 3);
scheduler.addCoroutine('monitor', monitorCoroutine);

// 启动协程调度
scheduler.start();
            

5.2 生产者-消费者模式

class Channel {
    constructor() {
        this.consumers = [];
        this.messages = [];
    }
    
    // 发送消息
    send(message) {
        return new Promise(resolve => {
            if (this.consumers.length > 0) {
                const consumer = this.consumers.shift();
                consumer({ value: message, resolve });
            } else {
                this.messages.push({ message, resolve });
            }
        });
    }
    
    // 接收消息
    receive() {
        return new Promise(resolve => {
            if (this.messages.length > 0) {
                const { message, resolve: sendResolve } = this.messages.shift();
                sendResolve();
                resolve(message);
            } else {
                this.consumers.push(resolve);
            }
        });
    }
}

// 生产者协程
function* producer(channel, items) {
    for (const item of items) {
        console.log(`生产者: 发送 ${item}`);
        yield channel.send(item);
        console.log(`生产者: ${item} 已发送`);
    }
    yield channel.send(null); // 发送结束信号
}

// 消费者协程
function* consumer(channel, name) {
    while (true) {
        const item = yield channel.receive();
        if (item === null) {
            console.log(`消费者 ${name}: 收到结束信号`);
            break;
        }
        console.log(`消费者 ${name}: 处理 ${item}`);
        yield new Promise(resolve => setTimeout(resolve, 500)); // 模拟处理时间
    }
}

// 运行生产者-消费者示例
async function runProducerConsumer() {
    const channel = new Channel();
    const items = ['苹果', '香蕉', '橙子', '葡萄', '芒果'];
    
    const producerGen = producer(channel, items);
    const consumer1Gen = consumer(channel, '消费者1');
    const consumer2Gen = consumer(channel, '消费者2');
    
    // 简单的协程执行器
    async function runCoroutine(gen) {
        let result = gen.next();
        while (!result.done) {
            await result.value;
            result = gen.next();
        }
    }
    
    // 并行执行生产者和消费者
    await Promise.all([
        runCoroutine(producerGen),
        runCoroutine(consumer1Gen),
        runCoroutine(consumer2Gen)
    ]);
    
    console.log('生产者-消费者模式执行完成');
}

runProducerConsumer();
            

六、高级特性与性能优化

6.1 Generator组合与委托

// Generator委托 yield*
function* combinedGenerator() {
    yield '开始组合生成器';
    
    // 委托给另一个Generator
    yield* numberGenerator(1, 3);
    
    yield '中间分隔';
    
    // 委托给数组
    yield* ['A', 'B', 'C'];
    
    yield '结束组合生成器';
}

function* numberGenerator(start, end) {
    for (let i = start; i <= end; i++) {
        yield `数字: ${i}`;
    }
}

const combined = combinedGenerator();
for (const value of combined) {
    console.log(value);
}
// 输出: 开始组合生成器, 数字: 1, 数字: 2, 数字: 3, 中间分隔, A, B, C, 结束组合生成器

// 递归Generator
function* recursiveTraversal(node) {
    if (!node) return;
    
    yield node.value;
    
    if (node.children) {
        for (const child of node.children) {
            yield* recursiveTraversal(child);
        }
    }
}

const tree = {
    value: '根节点',
    children: [
        {
            value: '子节点1',
            children: [
                { value: '孙节点1-1' },
                { value: '孙节点1-2' }
            ]
        },
        {
            value: '子节点2'
        }
    ]
};

console.log('树结构遍历:');
for (const nodeValue of recursiveTraversal(tree)) {
    console.log(nodeValue);
}
            

6.2 性能优化策略

// 惰性计算优化
function* lazyFilter(iterable, predicate) {
    for (const item of iterable) {
        if (predicate(item)) {
            yield item;
        }
    }
}

function* lazyMap(iterable, mapper) {
    for (const item of iterable) {
        yield mapper(item);
    }
}

// 使用惰性操作处理大数据集
const bigData = Array.from({ length: 1000000 }, (_, i) => i);

// 传统方式：立即执行所有操作
const traditionalResult = bigData
    .filter(x => x % 2 === 0)
    .map(x => x * 2)
    .slice(0, 10);

// 惰性方式：按需计算
const lazyResult = [];
const lazyStream = lazyMap(
    lazyFilter(bigData, x => x % 2 === 0),
    x => x * 2
);

for (const value of lazyStream) {
    lazyResult.push(value);
    if (lazyResult.length >= 10) break;
}

console.log('传统方式结果:', traditionalResult);
console.log('惰性方式结果:', lazyResult);

// 内存优化：流式处理
function* streamProcessor(stream, processor) {
    let buffer = [];
    const bufferSize = 1000;
    
    for (const item of stream) {
        buffer.push(processor(item));
        
        if (buffer.length >= bufferSize) {
            yield buffer;
            buffer = [];
        }
    }
    
    if (buffer.length > 0) {
        yield buffer;
    }
}

// 模拟数据流
function* dataStream() {
    let i = 0;
    while (i < 10000) {
        yield { id: i++, data: Math.random() };
        if (Math.random()  ({
        ...item,
        processed: true
    }))) {
        console.log(`处理批次: ${batch.length} 条数据`);
        // 这里可以批量存储或发送数据
        await new Promise(resolve => setTimeout(resolve, 10));
    }
    console.log('流式处理完成');
}

processLargeData();
            

七、生产环境应用案例

7.1 分页数据流处理

// 分页API数据获取器
function* paginatedDataFetcher(apiUrl, pageSize = 100) {
    let page = 1;
    let hasMore = true;
    
    while (hasMore) {
        console.log(`获取第 ${page} 页数据`);
        
        const url = `${apiUrl}?page=${page}&size=${pageSize}`;
        const response = yield fetch(url);
        const data = yield response.json();
        
        if (data.items && data.items.length > 0) {
            yield data.items; // 返回当前页数据
            page++;
            hasMore = data.hasMore;
        } else {
            hasMore = false;
        }
        
        // 添加延迟避免请求过于频繁
        if (hasMore) {
            yield new Promise(resolve => setTimeout(resolve, 100));
        }
    }
    
    console.log('所有分页数据获取完成');
}

// 数据处理器
async function processPaginatedData(apiUrl, processor) {
    const fetcher = paginatedDataFetcher(apiUrl);
    let result = fetcher.next();
    
    while (!result.done) {
        const value = await result.value;
        
        if (Array.isArray(value)) {
            // 处理数据批次
            await processor(value);
        }
        
        result = fetcher.next();
    }
}

// 使用示例
async function exampleUsage() {
    const mockProcessor = async (dataBatch) => {
        console.log(`处理 ${dataBatch.length} 条数据`);
        // 模拟数据处理
        await new Promise(resolve => setTimeout(resolve, 50));
    };
    
    // 在实际使用中替换为真实的API地址
    await processPaginatedData('/api/data', mockProcessor);
}

exampleUsage();
            

7.2 复杂工作流引擎

class WorkflowEngine {
    constructor() {
        this.steps = new Map();
        this.context = {};
    }
    
    addStep(name, stepGenerator) {
        this.steps.set(name, stepGenerator);
        return this;
    }
    
    setContext(context) {
        this.context = { ...this.context, ...context };
        return this;
    }
    
    async execute(workflowName, initialContext = {}) {
        this.setContext(initialContext);
        const workflow = this.steps.get(workflowName);
        
        if (!workflow) {
            throw new Error(`工作流 ${workflowName} 未定义`);
        }
        
        const generator = workflow.call(this, this.context);
        let result = generator.next();
        
        while (!result.done) {
            const value = await result.value;
            
            if (value && typeof value === 'object' && value.type === 'UPDATE_CONTEXT') {
                this.setContext(value.payload);
                result = generator.next();
            } else {
                result = generator.next(value);
            }
        }
        
        return result.value;
    }
}

// 定义工作流步骤
function* orderProcessingWorkflow(context) {
    console.log('开始订单处理工作流');
    
    // 步骤1: 验证订单
    const validationResult = yield validateOrder(context.order);
    if (!validationResult.valid) {
        throw new Error(`订单验证失败: ${validationResult.error}`);
    }
    
    yield { type: 'UPDATE_CONTEXT', payload: { validation: validationResult } };
    
    // 步骤2: 库存检查
    const inventoryCheck = yield checkInventory(context.order.items);
    if (!inventoryCheck.available) {
        throw new Error('库存不足');
    }
    
    // 步骤3: 支付处理
    const paymentResult = yield processPayment(context.order, context.paymentMethod);
    if (!paymentResult.success) {
        throw new Error(`支付失败: ${paymentResult.error}`);
    }
    
    // 步骤4: 发货处理
    const shippingResult = yield arrangeShipping(context.order);
    
    console.log('订单处理完成');
    return {
        success: true,
        orderId: context.order.id,
        trackingNumber: shippingResult.trackingNumber
    };
}

// 模拟业务函数
async function validateOrder(order) {
    await new Promise(resolve => setTimeout(resolve, 100));
    return { valid: true };
}

async function checkInventory(items) {
    await new Promise(resolve => setTimeout(resolve, 150));
    return { available: true };
}

async function processPayment(order, paymentMethod) {
    await new Promise(resolve => setTimeout(resolve, 200));
    return { success: true, transactionId: 'txn_12345' };
}

async function arrangeShipping(order) {
    await new Promise(resolve => setTimeout(resolve, 100));
    return { trackingNumber: 'TRK_67890' };
}

// 使用工作流引擎
async function runWorkflowExample() {
    const engine = new WorkflowEngine();
    engine.addStep('orderProcessing', orderProcessingWorkflow);
    
    const order = {
        id: 'ORD_001',
        items: [{ productId: 'PROD_1', quantity: 2 }],
        total: 99.99
    };
    
    try {
        const result = await engine.execute('orderProcessing', {
            order,
            paymentMethod: 'credit_card'
        });
        
        console.log('工作流执行结果:', result);
    } catch (error) {
        console.error('工作流执行失败:', error.message);
    }
}

runWorkflowExample();