关于后端:百度小程序包流式下载安装优化

45次阅读

共计 7259 个字符，预计需要花费 19 分钟才能阅读完成。

导读：文本介绍了百度小程序包下载链路的一种优化伎俩 —— 流式下载安装。首先引出原有计划的可优化点，接着探讨了优化计划是如何更充沛地利用了网络 IO、本地 IO、CPU 计算资源，最初介绍了代码层面的实现原理。

全文 3608 字，预计浏览工夫 10 分钟

一、问题背景

小程序安装过程中波及安装包的网络下载、保留文件、签名校验、解压解密等多个步骤，原有计划中各个步骤相互依赖串行执行，装置过程耗费时长为各个步骤之总和。

然而装置过程中各个阶段所竞争的资源不同，其中竞争网络 IO 资源的下载阶段耗时最长，且该阶段本地 IO 和 CPU 计算资源绝对最闲。

实践上能够突破装置过程中各个步骤间的依赖，实现在读取网络下载流的同时，将签名校验、解密解压等步骤同时执行，尽量在网络下载阶段充分利用零碎中的本地 IO 和 CPU 计算资源，以此缩小网络下载之后各个步骤所耗费的额定工夫。

二、解决方案

原有计划时序如下图所示，各个阶段竞争的忙碌关系剖析如下：

  • 下载安装包：网络 IO 最忙，CPU 计算较闲，本地 IO 较忙
  • 校验安装包：无需网络 IO，CPU 计算 (计算签名) 最忙，本地 IO (读文件)较忙
  • 提取包文件：无需网络 IO，CPU 计算 (解密、解压) 最忙，本地 IO (读写文件)最忙

对于性能开销，已知网络 IO 耗时远高于本地 IO 和 CPU 计算。

联合以上剖析可得，在读取下载流的同时能够并行实现流式解压和校验文件等解决，从而实现进步小程序下载安装阶段性能的指标，产出流式装置计划如下：

流式下载安装计划如上图所示，实现流式下载安装性能须要实现下载流 (response.body) 接入和解决流管线 (PipeLine) 派发这两个问题。

职责设计上 MultiPipe 是一个解决流的根底工具，能够将接入的一个输出流，同时泵到不同线程中的执行管线上，实现输出流的一分多，是一种相似进气歧管的结构，如下图所示：

MultiPipe 在工作过程中，会结构出若干消费者解决管线(PipeLine)，由一个 executor 异步执行，同时将输出流中的数据源源不断的泵给各条 PipeLine，直到输出流完结以及所有 PipeLine 全副执行处理完毕。

以下是 MultiPipe 的用例，例如输出通道是 okhttp3.ResponseBody#source 的返回后果，即网络申请响应体的二进制流，两个消费者，别离实现签名校验和解压解密的动作。

ReadableByteChannel srcChannel = ... // 例如 okhttp3.ResponseBody#source 办法的返回后果 ExecutorService threadPool = Executors.newFixedThreadPool(2); // 可选参数 MultiPipe multiplePipe = new MultiPipe(new Consumer<ReadableByteChannel>() {@Override    public void accept(ReadableByteChannel source) {// 对整个流做 md5，进行签名校验，CPU 忙}}, new Consumer<ReadableByteChannel>() {    @Override    public void accept(ReadableByteChannel source) {// 对整个流进行解密、解压、写入磁盘，CPU 和 IO 忙}}) {@Override    protected ExecutorService onCreateExecutor(int consumerSize) {return threadPool;}};// 每次能从网络流中读取的最大字节数，对应 okio.Segment#SIZEmultiplePipe.setTmpBufferCapacity(MultiPipe.TMP_BUFFER_CAPACITY); // 开始传输 multiplePipe.connect(srcChannel);

小结：小程序调起场景的包下载是一种高优的工作，通过优化解决主包 response.body 的办法，在读取网络流时，对每次读到的字节数组进行拷贝，再通过 Pipe 传输给各个消费者(签名校验、解密解压)，从而将串行的包下载到本地、签名校验、解密解压的耗时，降为读网络流、签名校验或解密解压的三者之间的最大耗时。

收益：通过流式下载安装优化，线上包下载时长升高了 21%。

三、实现剖析

MultiPipe 是一种能够将一个输出通道，分为多个输入通道的工具类，示例代码如下：

/**
 * 能够将一个输出通道，分为多个输入通道的管道
 */
public class MultiPipe {/** 长期缓存的大小 {@see okio.Segment#SIZE} */
    public static final int TMP_BUFFER_CAPACITY = 8 * 1024;
    /** 消费者列表 */
    private final List<Consumer<ReadableByteChannel>> mConsumerList;
    /** 长期缓存的大小 {@see okio.Segment#SIZE} */
    private int mTmpBufferCapacity = TMP_BUFFER_CAPACITY;
    
    /**
     * 构造方法
     *
     * @param consumers 消费者列表
     */
    @SafeVarargs
    public MultiPipe(Consumer<ReadableByteChannel>... consumers) {mConsumerList = Arrays.asList(consumers);
    }
    
    // 设置示意每次最多传输多少字节，例如 8 * 1024
    // public final void setTmpBufferCapacity(int maxBytes)
    
    // connect 办法及其依赖的办法：//   transfer、createPipeLineList、launchPipeLineList 办法
    
    // 可供使用方重写的办法：//   setHasPipeBuffer、onStart、onCreateExecutor、onException、//   onTransferComplete、onUpdateProgress、onFinish
}

3.1 创立管线列表 (PipeLineList) 并连贯输出通道(ReadableChannel)

应用方通过 connect 实现所有工作，在该办法中，首先依据构造方法中的消费者列表，创立管线列表和 latch，再通过线程池启动各个管线，使各个工作开始工作。

(1) 依据消费者数量创立对应的管线(PipeLine)，连贯输出通道。latch 的作用是确保所有消费者工作都完结后，才视为执行实现，要害语句为 latch.await();

/** * 连贯输出流 * * @param source 输出流 */public final void connect(ReadableByteChannel source) {onStart(source); // 回调 - 开始        // 创立管线列表    List<PipeLine> pipeLineList = createPipeLineList();    // 依据消费者数量，创立 latch    CountDownLatch latch = new CountDownLatch(pipeLineList.size());     // 让连贯各个管线的工作开始工作    ExecutorService executorService = launchPipeLineList(pipeLineList, latch);     try {transfer(source, pipeLineList); // 开始传输        onTransferComplete(latch); // 回调 - 传输实现(期待敞开，默认期待 latch)    } catch (IOException e) {onException(e); // 回调 - 异样解决    } finally {onFinish(source, executorService); // 回调 - 完结    }}
// 当开始连贯时，回调给应用方 // protected void onStart(ReadableByteChannel source)
/** * 能够由应用方重写，传输实现，解决 Latch，能够抉择始终期待，也能够设置为超时机制 * * @param latch CountDownLatch */protected void onTransferComplete(CountDownLatch latch) {try {        latch.await();    } catch (InterruptedException ignored) {}}
/** * 能够抉择是否敞开线程池 * * @param source          输出 * @param executorService 线程池 */protected void onFinish(ReadableByteChannel source,                         ExecutorService executorService) {closeChannel(source);    executorService.shutdown();}

(2) 依据消费者数量创立管线列表

/* 创立管道列表 @return 管道列表 */private List<PipeLine> createPipeLineList() { final List<PipeLine> pipeLineList = new ArrayList<>(mConsumerList.size()); for (Consumer<ReadableByteChannel> consumer : mConsumerList) {pipeLineList.add(new PipeLine(consumer, hasPipeBuffer())); } return pipeLineList;}

(3) 通过线程池启动每个管线

其中线程池能够设置为已有线程池。如果为了防止已有线程池被敞开，则须要重写 onFinish 办法，移除 executorService.shutdown(); 语句。

/** * 调起管线列表，返回执行者实例 * * @param pipeLineList 管线列表 * @param latch        用于确保所有工作一起实现 * @return 执行者实例 */private ExecutorService launchPipeLineList(List<PipeLine> pipeLineList,                                            CountDownLatch latch) {ExecutorService executorService = onCreateExecutor(pipeLineList.size());    for (PipeLine pipeLine : pipeLineList) {pipeLine.setLaunch(latch);        executorService.submit(pipeLine);    }    return executorService;}/** * 由应用方决定如何创立线程池，例如应用已有的线程池 * * @param consumerSize 消费者个数 * @return 可用的线程池实例 */protected ExecutorService onCreateExecutor(int consumerSize) {return Executors.newFixedThreadPool(consumerSize);}

3.2 将每次读到的内容传输给各个管线(PipeLine)

在读取输出通道时，将每次读到的字节缓冲区，传输给各个消费者。
通过 ByteBuffer 接管每次能够读到的内容，再遍历消费者列表，将 ByteBuffer 中的内容写入管线的 sink 通道。
最初，将各消费者管线的 sink 通道敞开。

通过 onUpdateProgress 办法能够回调以后进度给应用方。

/** * 传输 * * @param source       输出流 / 输出通道 * @param pipeLineList 管线列表 */private void transfer(ReadableByteChannel source,                       List<PipeLine> pipeLineList) throws IOException {long writeBytes = 0; // 累计写出的字节数    onUpdateProgress(writeBytes); // 告诉应用方以后的传输进度    try {final ByteBuffer buf = ByteBuffer.allocate(mTmpBufferCapacity);        long reads;        while ((reads = source.read(buf)) != -1) {buf.flip(); // 开始读取 buf 中的内容            for (PipeLine pipeLine : pipeLineList) {if (pipeLine.mSink.isOpen() && pipeLine.mSource.isOpen()) {buf.rewind(); // 重读 Buffer 中的所有数据                    pipeLine.mSink.write(buf); // 向管线中传输内容                }            }            buf.clear();            writeBytes += reads;            onUpdateProgress(writeBytes); // 告诉应用方以后的传输进度        }    } finally {for (PipeLine pipeLine : pipeLineList) {closeChannel(pipeLine.mSink); // 须要敞开，否则会陷入阻塞        }    }}

‍

3.3 管线 (PipeLine) 的实现

每个消费者工作对应一个管线(PipeLine)，在传输网络流的过程中，将每次读到的字节缓冲写入每个管线的 sink 通道，再将管线的 source 通道提供给消费者工作。

其中管线的实现能够基于 java.nio.channels.Pipe，也能够应用带缓冲区的 okio.Pipe。
带缓冲区会减少传输耗时，但能够躲避极其状况下生产过慢导致读取速度变慢的问题：例如消费者解码耗时过长，导致 TCP 误判网络不好，而频繁超时重传。

/** * 管线，也作为工作工作将流导给消费者 */private static class PipeLine implements Runnable {/** 管线消费者 */    final transient Consumer<ReadableByteChannel> mConsumer;    /** pipe.source */    final transient ReadableByteChannel mSource;    /** pipe.sink */    final transient WritableByteChannel mSink;    /** 用来做线程同步 */    transient CountDownLatch mLatch;        /**     * 构造方法     *     * @param hasBuffer 是否带缓冲区     * @param consumer  消费者     */    public PipeLine(Consumer<ReadableByteChannel> consumer,                     boolean hasBuffer) {mConsumer = consumer;        if (hasBuffer) {// 带缓冲区的 Pipe            okio.Pipe okioPipe = new okio.Pipe(getPipeMaxBufferBytes());            mSink = okio.Okio.buffer(okioPipe.sink());            mSource = okio.Okio.buffer(okioPipe.source());        } else {// 无缓冲区的 Pipe            try {                java.nio.channels.Pipe pipe = java.nio.channels.Pipe.open();                mSource = pipe.source();                mSink = pipe.sink();            } catch (IOException e) {throw new IllegalStateException(e);            }        }    }        @Override    public void run() {        try {            mConsumer.accept(mSource);        } finally {closeChannel(mSink);            closeChannel(mSource);            if (mLatch != null) {mLatch.countDown();            }        }    }}

其中 getPipeMaxBufferBytes 办法能够参考以下实现，基于以后可用内存思考，返回一个平安的缓冲区容量。

 /** 可用内存的比例 */
  private static final float FACTOR = 0.75F;
    
  /**
   * 返回缓冲区的最大容量
   *
   * @return 基于理论内存思考，防止 OOM 的可用内存字节数
   */
  private static long getPipeMaxBufferBytes() {Runtime r = Runtime.getRuntime();
      long available = r.maxMemory() - r.totalMemory() + r.freeMemory();
      return (long) (available * FACTOR);
  }

四、总结与瞻望

总结：本文形容了百度小程序包下载链路中的一种优化伎俩，即流式下载安装计划。全文以充分利用网络 IO、本地 IO、CPU 资源的指标为线索，剖析了原有计划的可优化点，并将优化计划与原有计划进行比照，最初对其外围实现进行了代码剖析，介绍了一种反对将一个输出通道，分为多个输入通道的实现形式。

瞻望：流式下载安装计划须要解压整个小程序包，展望未来，摸索一种渐进式的资源加载计划，能够利用这种共享流的形式，尽可能使低优先级的资源下载异步化。

举荐浏览：

日志中台不重不丢实现浅谈

百度 ToB 垂类账号权限平台的设计与实际

视觉 Transformer 中的输出可视化办法

深刻了解 WKWebView (渲染篇) —— DOM 树的构建

深刻了解 WKWebView（入门篇）—— WebKit 源码调试与剖析

正文完
后端
发表至： 后端
2022-04-20
 0
关于后端:12张图一次性搞懂高性能并发容器ConcurrentLinkedQueue
关于后端:基于Java的心理服务平台的设计与实现第十二稿中期检查表ppt开题任
关于后端:Python入门系列七开发常说的累与对象
关于后端:ffmpeg实战音视频合成案例