当前位置：首页 > news >正文

JavaScript系列（79）--Web Worker 高级应用

news 来源：原创 2025/5/18 12:27:35

Web Worker 高级应用 🔄

Web Worker 为JavaScript提供了真正的多线程能力，让我们能够在后台线程中执行复杂的计算而不阻塞主线程。今天让我们深入探讨Web Worker的高级应用。

Web Worker 概述 🌟

💡 小知识：Web Worker 是一个独立的JavaScript执行线程，它可以在不阻塞主线程的情况下执行计算密集型任务。Worker有多种类型：专用Worker（Dedicated Worker）、共享Worker（Shared Worker）和服务Worker（Service Worker）。

Worker类型与创建 📊

// 1. 专用Worker
const dedicatedWorker = new Worker('worker.js');

// 2. 共享Worker
const sharedWorker = new SharedWorker('shared-worker.js');

// 3. 内联Worker
const workerBlob = new Blob([`
    self.onmessage = function(e) {
        const result = e.data * 2;
        self.postMessage(result);
    }
`], { type: 'application/javascript' });

const inlineWorker = new Worker(URL.createObjectURL(workerBlob));

// 4. Worker模块
const moduleWorker = new Worker('module-worker.js', {
    type: 'module'
});

// 5. Worker配置选项
const configuredWorker = new Worker('worker.js', {
    name: 'MyWorker',
    type: 'classic',
    credentials: 'same-origin'
});

通信模式实现 🔄

// 1. 基本消息传递
class WorkerMessenger {
    constructor(workerScript) {
        this.worker = new Worker(workerScript);
        this.setupMessageHandlers();
    }
    
    setupMessageHandlers() {
        this.worker.onmessage = this.handleMessage.bind(this);
        this.worker.onerror = this.handleError.bind(this);
    }
    
    handleMessage(event) {
        console.log('Received from worker:', event.data);
    }
    
    handleError(error) {
        console.error('Worker error:', error);
    }
    
    sendMessage(data) {
        this.worker.postMessage(data);
    }
    
    terminate() {
        this.worker.terminate();
    }
}

// 2. 结构化克隆传输
class StructuredCloneDemo {
    static sendComplexData(worker) {
        const data = {
            date: new Date(),
            regex: /pattern/,
            map: new Map([['key', 'value']]),
            set: new Set([1, 2, 3])
        };
        
        worker.postMessage(data);
    }
}

// 3. Transferable Objects
class TransferableDemo {
    static sendArrayBuffer(worker) {
        const buffer = new ArrayBuffer(1024 * 1024); // 1MB
        worker.postMessage(buffer, [buffer]);
        
        // buffer在传输后不可用
        console.log('Buffer size after transfer:', buffer.byteLength); // 0
    }
    
    static sendImageData(worker) {
        const canvas = document.createElement('canvas');
        const ctx = canvas.getContext('2d');
        const imageData = ctx.createImageData(100, 100);
        
        worker.postMessage(imageData, [imageData.data.buffer]);
    }
}

// 4. 共享内存通信
class SharedMemoryDemo {
    constructor() {
        this.sharedBuffer = new SharedArrayBuffer(1024);
        this.sharedArray = new Int32Array(this.sharedBuffer);
    }
    
    setupWorker() {
        const worker = new Worker('worker.js');
        worker.postMessage({ 
            buffer: this.sharedBuffer,
            command: 'initialize'
        });
        
        return worker;
    }
    
    atomicOperation() {
        // 原子操作
        Atomics.add(this.sharedArray, 0, 1);
        Atomics.store(this.sharedArray, 1, 42);
        
        const value = Atomics.load(this.sharedArray, 0);
        console.log('Current value:', value);
    }
}

性能优化模式 ⚡

// 1. Worker池实现
class WorkerPool {
    constructor(workerScript, poolSize = navigator.hardwareConcurrency) {
        this.workers = [];
        this.queue = [];
        this.activeWorkers = new Map();
        
        for (let i = 0; i < poolSize; i++) {
            const worker = new Worker(workerScript);
            worker.onmessage = this.handleWorkerMessage.bind(this, worker);
            this.workers.push(worker);
        }
    }
    
    handleWorkerMessage(worker, event) {
        const { taskId, result } = event.data;
        const resolver = this.activeWorkers.get(taskId);
        
        if (resolver) {
            resolver(result);
            this.activeWorkers.delete(taskId);
            this.processQueue();
        }
    }
    
    async executeTask(data) {
        return new Promise((resolve, reject) => {
            const task = { data, resolve, reject };
            
            if (this.hasIdleWorker()) {
                this.assignTaskToWorker(task);
            } else {
                this.queue.push(task);
            }
        });
    }
    
    hasIdleWorker() {
        return this.workers.some(worker => 
            !Array.from(this.activeWorkers.values()).includes(worker)
        );
    }
    
    assignTaskToWorker(task) {
        const worker = this.workers.find(w => 
            !Array.from(this.activeWorkers.values()).includes(w)
        );
        
        if (worker) {
            const taskId = Date.now() + Math.random();
            this.activeWorkers.set(taskId, task.resolve);
            worker.postMessage({ taskId, data: task.data });
        }
    }
    
    processQueue() {
        if (this.queue.length > 0 && this.hasIdleWorker()) {
            const task = this.queue.shift();
            this.assignTaskToWorker(task);
        }
    }
    
    terminate() {
        this.workers.forEach(worker => worker.terminate());
        this.workers = [];
        this.queue = [];
        this.activeWorkers.clear();
    }
}

// 2. 任务分片
class TaskChunking {
    static createChunks(data, chunkSize) {
        const chunks = [];
        for (let i = 0; i < data.length; i += chunkSize) {
            chunks.push(data.slice(i, i + chunkSize));
        }
        return chunks;
    }
    
    static async processWithWorkers(data, workerCount) {
        const chunkSize = Math.ceil(data.length / workerCount);
        const chunks = this.createChunks(data, chunkSize);
        
        const workers = chunks.map(chunk => {
            const worker = new Worker('processor.js');
            worker.postMessage(chunk);
            return worker;
        });
        
        return Promise.all(workers.map(worker => 
            new Promise(resolve => {
                worker.onmessage = e => {
                    worker.terminate();
                    resolve(e.data);
                };
            })
        ));
    }
}

错误处理与调试 🔍

// 1. 错误处理
class WorkerErrorHandler {
    constructor(worker) {
        this.worker = worker;
        this.setupErrorHandling();
    }
    
    setupErrorHandling() {
        // 错误事件
        this.worker.onerror = this.handleError.bind(this);
        
        // 未捕获的rejection
        this.worker.onmessageerror = this.handleMessageError.bind(this);
    }
    
    handleError(error) {
        console.error('Worker error:', {
            message: error.message,
            filename: error.filename,
            lineno: error.lineno,
            colno: error.colno
        });
        
        // 通知主线程
        this.worker.postMessage({
            type: 'error',
            error: {
                message: error.message,
                stack: error.error?.stack
            }
        });
    }
    
    handleMessageError(error) {
        console.error('Message error:', error);
    }
}

// 2. 调试工具
class WorkerDebugger {
    constructor(enabled = true) {
        this.enabled = enabled;
        this.logs = [];
    }
    
    log(...args) {
        if (this.enabled) {
            const log = {
                timestamp: new Date(),
                data: args
            };
            
            this.logs.push(log);
            console.log('[Worker Debug]', ...args);
        }
    }
    
    error(...args) {
        if (this.enabled) {
            const log = {
                timestamp: new Date(),
                type: 'error',
                data: args
            };
            
            this.logs.push(log);
            console.error('[Worker Error]', ...args);
        }
    }
    
    getLogHistory() {
        return this.logs;
    }
    
    clearLogs() {
        this.logs = [];
    }
}

实战应用示例 💼

// 1. 图像处理Worker
class ImageProcessingWorker {
    constructor() {
        this.worker = new Worker('image-processor.js');
        this.setupHandlers();
    }
    
    setupHandlers() {
        this.worker.onmessage = this.handleResult.bind(this);
        this.worker.onerror = this.handleError.bind(this);
    }
    
    async processImage(imageData, filters) {
        return new Promise((resolve, reject) => {
            this.worker.onmessage = e => resolve(e.data);
            this.worker.onerror = e => reject(e);
            
            this.worker.postMessage({
                imageData,
                filters
            }, [imageData.data.buffer]);
        });
    }
    
    handleResult(event) {
        const canvas = document.createElement('canvas');
        const ctx = canvas.getContext('2d');
        ctx.putImageData(event.data, 0, 0);
    }
    
    handleError(error) {
        console.error('Image processing error:', error);
    }
}

// 2. 数据分析Worker
class DataAnalysisWorker {
    constructor() {
        this.workerPool = new WorkerPool('data-analyzer.js', 4);
    }
    
    async analyzeLargeDataset(dataset) {
        const chunks = TaskChunking.createChunks(dataset, 1000);
        const results = await Promise.all(
            chunks.map(chunk => 
                this.workerPool.executeTask({
                    type: 'analyze',
                    data: chunk
                })
            )
        );
        
        return this.mergeResults(results);
    }
    
    mergeResults(results) {
        // 合并分析结果
        return results.reduce((acc, result) => ({
            sum: acc.sum + result.sum,
            average: (acc.sum + result.sum) / (acc.count + result.count),
            count: acc.count + result.count
        }), { sum: 0, average: 0, count: 0 });
    }
}

// 3. 实时数据处理
class RealTimeProcessor {
    constructor() {
        this.worker = new Worker('processor.js');
        this.sharedMemory = new SharedMemoryDemo();
    }
    
    startProcessing(dataStream) {
        dataStream.addEventListener('data', event => {
            const chunk = event.data;
            
            // 使用共享内存传输数据
            const sharedBuffer = new SharedArrayBuffer(chunk.length);
            const sharedArray = new Uint8Array(sharedBuffer);
            sharedArray.set(new Uint8Array(chunk));
            
            this.worker.postMessage({
                type: 'process',
                buffer: sharedBuffer
            });
        });
    }
}

最佳实践与注意事项 ⚠️

// 1. Worker生命周期管理
class WorkerLifecycle {
    constructor() {
        this.workers = new Set();
    }
    
    createWorker(script) {
        const worker = new Worker(script);
        this.workers.add(worker);
        return worker;
    }
    
    terminateWorker(worker) {
        worker.terminate();
        this.workers.delete(worker);
    }
    
    terminateAll() {
        this.workers.forEach(worker => worker.terminate());
        this.workers.clear();
    }
}

// 2. 资源监控
class WorkerMonitor {
    constructor() {
        this.metrics = new Map();
    }
    
    startMonitoring(worker) {
        const startTime = performance.now();
        const metric = {
            startTime,
            messageCount: 0,
            errors: 0,
            lastActive: startTime
        };
        
        this.metrics.set(worker, metric);
        
        worker.addEventListener('message', () => {
            metric.messageCount++;
            metric.lastActive = performance.now();
        });
        
        worker.addEventListener('error', () => {
            metric.errors++;
        });
    }
    
    getMetrics(worker) {
        const metric = this.metrics.get(worker);
        const now = performance.now();
        
        return {
            ...metric,
            uptime: now - metric.startTime,
            idleTime: now - metric.lastActive
        };
    }
}