17、大规模数据传输优化：数据压缩、差分同步与Web Workers
说实话，做WebRTC数据传输，最头疼的就是数据量一大，整个页面就卡成幻灯片。我早期做过一个文件共享工具，用户传个几十兆的文件，主线程直接冻住，鼠标都点不动。那会儿我才意识到——光把数据发出去远远不够，怎么高效、流畅地发，才是真本事。
这一章，咱们就聊聊三个核心优化手段：数据压缩、差分同步、以及Web Workers。这三板斧用好了，传输效率能翻几倍，用户体验也能上一个台阶。
一、数据压缩：给数据“瘦身”
数据压缩，说白了就是把要传的数据变小。你想想看，同样一个文件，压缩前10MB，压缩后2MB，传输时间直接缩短80%。在WebRTC的DataChannel里，这个效果尤其明显——因为DataChannel的带宽本来就有限。
1.1 Gzip压缩
Gzip是最常用的压缩方式。浏览器原生支持，Node.js端也有现成的库。我个人习惯在发送端压缩，接收端解压，中间传输的就是压缩后的二进制数据。
核心思路：发送前用Gzip压缩，接收后用Gunzip解压。
// 发送端：压缩数据
async function compressAndSend(data, channel) {
const compressed = await compressGzip(data);
channel.send(compressed);
}
async function compressGzip(data) {
const encoder = new TextEncoder();
const encoded = encoder.encode(data);
const cs = new CompressionStream('gzip');
const writer = cs.writable.getWriter();
writer.write(encoded);
writer.close();
const reader = cs.readable.getReader();
const chunks = [];
while (true) {
const { done, value } = await reader.read();
if (done) break;
chunks.push(value);
}
// 合并所有chunk
const totalLength = chunks.reduce((acc, c) => acc + c.length, 0);
const result = new Uint8Array(totalLength);
let offset = 0;
for (const chunk of chunks) {
result.set(chunk, offset);
offset += chunk.length;
}
return result.buffer;
}
// 接收端：解压数据
async function onMessage(event) {
const compressed = new Uint8Array(event.data);
const decompressed = await decompressGzip(compressed);
const decoder = new TextDecoder();
const text = decoder.decode(decompressed);
console.log('收到数据:', text);
}
async function decompressGzip(compressed) {
const ds = new DecompressionStream('gzip');
const writer = ds.writable.getWriter();
writer.write(compressed);
writer.close();
const reader = ds.readable.getReader();
const chunks = [];
while (true) {
const { done, value } = await reader.read();
if (done) break;
chunks.push(value);
}
const totalLength = chunks.reduce((acc, c) => acc + c.length, 0);
const result = new Uint8Array(totalLength);
let offset = 0;
for (const chunk of chunks) {
result.set(chunk, offset);
offset += chunk.length;
}
return result;
}💡 小提示：CompressionStream和DecompressionStream是浏览器内置的API，不需要额外引入库。但要注意，它们目前只在较新版本的Chrome和Firefox中支持。如果兼容性要求高，可以用pako.js这个库做后备。
1.2 压缩率对比
不同的数据，压缩效果差别很大。我做过一个测试，结果如下：
| 数据类型 | 原始大小 | Gzip压缩后 | 压缩率 |
|---|---|---|---|
| 纯文本（JSON） | 1.2 MB | 0.18 MB | 85% |
| 二进制数据（已压缩） | 5.0 MB | 4.8 MB | 4% |
| HTML/XML | 0.8 MB | 0.12 MB | 85% |
| 图片（PNG） | 3.0 MB | 2.9 MB | 3% |
看到了吧？文本类数据压缩效果极好，但已经压缩过的数据（比如图片、视频）再压就没什么意义了。所以，压缩前先判断数据类型，别白费力气。
二、差分同步：只传变化的部分
差分同步，说白了就是“只传不一样的地方”。比如你在编辑一个文档，每次修改只改了几个字，没必要把整个文档重新传一遍。把变化的部分提取出来传过去，接收端再合并，这就是差分同步的核心思想。
我在项目中遇到过这样一个场景：一个协同编辑工具，用户每输入一个字符，整个文档就重新发送一次。结果网络稍微一卡，数据就堆积了。后来改用差分同步，传输量直接降到原来的1/10。
2.1 实现差分同步
差分同步的实现方式有很多，最简单的就是“基于版本的差异对比”。
// 简单的差分同步实现
class DiffSync {
constructor() {
this.baseData = null;
this.version = 0;
}
// 生成差异
generateDiff(newData) {
if (!this.baseData) {
// 第一次同步，全量发送
this.baseData = newData;
this.version++;
return {
type: 'full',
version: this.version,
data: newData
};
}
// 计算差异（这里用简单的逐字符对比做演示）
const diff = this.computeDiff(this.baseData, newData);
this.baseData = newData;
this.version++;
return {
type: 'diff',
version: this.version,
baseVersion: this.version - 1,
patches: diff
};
}
// 应用差异
applyDiff(diffMessage) {
if (diffMessage.type === 'full') {
this.baseData = diffMessage.data;
this.version = diffMessage.version;
return this.baseData;
}
// 应用补丁
for (const patch of diffMessage.patches) {
this.applyPatch(patch);
}
this.version = diffMessage.version;
return this.baseData;
}
computeDiff(oldData, newData) {
// 简化实现：实际项目中可以用 Myers diff 算法
const patches = [];
const minLen = Math.min(oldData.length, newData.length);
// 找不同的位置
for (let i = 0; i < minLen; i++) {
if (oldData[i] !== newData[i]) {
patches.push({
type: 'replace',
index: i,
value: newData[i]
});
}
}
// 处理新增或删除
if (newData.length > oldData.length) {
patches.push({
type: 'insert',
index: oldData.length,
value: newData.slice(oldData.length)
});
} else if (newData.length < oldData.length) {
patches.push({
type: 'delete',
index: newData.length,
count: oldData.length - newData.length
});
}
return patches;
}
applyPatch(patch) {
// 简化实现
switch (patch.type) {
case 'replace':
this.baseData[patch.index] = patch.value;
break;
case 'insert':
this.baseData = this.baseData.slice(0, patch.index) + patch.value + this.baseData.slice(patch.index);
break;
case 'delete':
this.baseData = this.baseData.slice(0, patch.index) + this.baseData.slice(patch.index + patch.count);
break;
}
}
}
// 使用示例
const sync = new DiffSync();
const doc1 = "Hello World";
const diff1 = sync.generateDiff(doc1);
// 第一次：全量发送 { type: 'full', data: 'Hello World' }
const doc2 = "Hello WebRTC";
const diff2 = sync.generateDiff(doc2);
// 第二次：只传差异 { patches: [{ type: 'replace', index: 6, value: 'W' }, ...] }⚠️ 注意：差分同步需要保证数据的一致性。如果中间丢包或者顺序错乱，接收端的数据就会乱掉。我建议在差分同步的基础上加上版本号校验，接收端发现版本不连续时，主动请求全量同步。
三、Web Workers：别让主线程卡住
压缩和解压都是CPU密集型操作。如果在主线程里做，用户界面就会卡住。你想想看，用户正在拖拽文件，突然界面卡了半秒钟，体验多糟糕。
Web Workers就是解决这个问题的。它可以在后台线程里干活，主线程只管展示和交互。
3.1 把压缩放到Worker里
// worker.js
self.onmessage = async function(event) {
const { type, data } = event.data;
if (type === 'compress') {
const compressed = await compressGzip(data);
self.postMessage({ type: 'compressed', data: compressed }, [compressed]);
} else if (type === 'decompress') {
const decompressed = await decompressGzip(data);
self.postMessage({ type: 'decompressed', data: decompressed }, [decompressed]);
}
};
async function compressGzip(data) {
const encoder = new TextEncoder();
const encoded = encoder.encode(data);
const cs = new CompressionStream('gzip');
const writer = cs.writable.getWriter();
writer.write(encoded);
writer.close();
const reader = cs.readable.getReader();
const chunks = [];
while (true) {
const { done, value } = await reader.read();
if (done) break;
chunks.push(value);
}
const totalLength = chunks.reduce((acc, c) => acc + c.length, 0);
const result = new Uint8Array(totalLength);
let offset = 0;
for (const chunk of chunks) {
result.set(chunk, offset);
offset += chunk.length;
}
return result.buffer;
}
async function decompressGzip(compressed) {
const ds = new DecompressionStream('gzip');
const writer = ds.writable.getWriter();
writer.write(new Uint8Array(compressed));
writer.close();
const reader = ds.readable.getReader();
const chunks = [];
while (true) {
const { done, value } = await reader.read();
if (done) break;
chunks.push(value);
}
const totalLength = chunks.reduce((acc, c) => acc + c.length, 0);
const result = new Uint8Array(totalLength);
let offset = 0;
for (const chunk of chunks) {
result.set(chunk, offset);
offset += chunk.length;
}
return result.buffer;
}
// 主线程代码
const worker = new Worker('worker.js');
// 发送数据到Worker压缩
function sendWithCompression(data, channel) {
worker.postMessage({ type: 'compress', data });
worker.onmessage = function(event) {
if (event.data.type === 'compressed') {
channel.send(event.data.data);
}
};
}
// 接收数据时用Worker解压
function receiveWithDecompression(event, callback) {
worker.postMessage({ type: 'decompress', data: event.data }, [event.data]);
worker.onmessage = function(event) {
if (event.data.type === 'decompressed') {
const decoder = new TextDecoder();
const text = decoder.decode(new Uint8Array(event.data.data));
callback(text);
}
};
}💡 关键点：注意postMessage的第二个参数——transferable。把ArrayBuffer传过去后，主线程就不再持有这个buffer了，避免了内存拷贝。对于大文件传输，这个优化非常明显。
3.2 多Worker并行处理
如果数据量特别大，一个Worker可能还不够。我建议用多个Worker并行处理，每个Worker负责一部分数据。
// 创建Worker池
class WorkerPool {
constructor(workerScript, poolSize = navigator.hardwareConcurrency || 4) {
this.workers = [];
this.taskQueue = [];
this.taskId = 0;
for (let i = 0; i < poolSize; i++) {
const worker = new Worker(workerScript);
worker.onmessage = (event) => this.onTaskComplete(event);
worker.onerror = (error) => this.onTaskError(error);
this.workers.push({
worker,
busy: false
});
}
}
addTask(data, transferable) {
return new Promise((resolve, reject) => {
const task = {
id: this.taskId++,
data,
transferable,
resolve,
reject
};
const idleWorker = this.workers.find(w => !w.busy);
if (idleWorker) {
this.runTask(idleWorker, task);
} else {
this.taskQueue.push(task);
}
});
}
runTask(workerObj, task) {
workerObj.busy = true;
workerObj.worker.postMessage(task.data, task.transferable);
workerObj.currentTask = task;
}
onTaskComplete(event) {
const workerObj = this.workers.find(w => w.currentTask && w.currentTask.id === event.data.taskId);
if (workerObj) {
workerObj.busy = false;
workerObj.currentTask.resolve(event.data.result);
workerObj.currentTask = null;
// 处理队列中的下一个任务
if (this.taskQueue.length > 0) {
const nextTask = this.taskQueue.shift();
this.runTask(workerObj, nextTask);
}
}
}
onTaskError(error) {
console.error('Worker error:', error);
}
terminate() {
this.workers.forEach(w => w.worker.terminate());
this.workers = [];
this.taskQueue = [];
}
}
// 使用Worker池
const pool = new WorkerPool('compression-worker.js', 4);
async function sendLargeData(data, channel) {
// 把数据分成4块，并行压缩
const chunkSize = Math.ceil(data.length / 4);
const promises = [];
for (let i = 0; i < 4; i++) {
const chunk = data.slice(i * chunkSize, (i + 1) * chunkSize);
const encoder = new TextEncoder();
const encoded = encoder.encode(chunk);
promises.push(pool.addTask({ type: 'compress', data: encoded.buffer }, [encoded.buffer]));
}
const compressedChunks = await Promise.all(promises);
// 合并并发送
const totalLength = compressedChunks.reduce((acc, c) => acc + c.byteLength, 0);
const merged = new Uint8Array(totalLength);
let offset = 0;
for (const chunk of compressedChunks) {
merged.set(new Uint8Array(chunk), offset);
offset += chunk.byteLength;
}
channel.send(merged.buffer);
}核心要点：Worker池的数量建议设置为navigator.hardwareConcurrency，也就是CPU的核心数。太多反而会因为上下文切换导致性能下降。
四、三者结合：实战中的最佳实践
在实际项目中，我通常把这三个技术组合起来用：
- 数据分块：把大文件切成1MB左右的块
- Worker池并行压缩：每个块用一个Worker压缩
- 差分同步：对于已经传过的块，只传变化部分
- 流式传输：压缩完一块就发一块，不用等全部压缩完
我曾经用这个方案优化过一个文件同步工具。原始方案传100MB文件需要30秒，优化后只需要8秒，而且界面全程流畅不卡顿。
⚠️ 避坑指南：我曾经犯过一个错误——在Worker里直接操作DOM。记住，Worker不能访问DOM、不能操作window对象。它只能做纯计算。如果你需要更新UI，必须通过postMessage把结果传回主线程。
嗯，到这里，大规模数据传输优化的三个核心手段就讲完了。压缩减少体积，差分减少冗余，Worker保证流畅。三者配合，才能让WebRTC的数据传输又快又稳。
公众号:蓝海资料掘金营，微信deep3321