图片加载库 kean/Nuke ，源代码看看

加载一张网络图片流程，3 步:

1, 把这张网络图片，下载到本地；

这时候有了一个 Data

2，对图片 Data 解码，

即拿 Data，生成 UIImage,

3，显示图片，

把生成的 UIImage, 赋给 UIImageView 的 image

nuke 的调用

let request = ImageRequest(url: URL(string: "https://user-images.githubusercontent.com/1567433/59150453-178bbb80-8a24-11e9-94ca-fd8dff6e2a9a.jpeg")!,processors: processors)var options = ImageLoadingOptions(transition: .fadeIn(duration: 0.5))options.pipeline = pipeline//  request, 网络图片资源链接 （  对应第一步 ）//  options， 效果//  imageView， 来显示的 UIImageView， （  对应第 3 步 ）loadImage(with: request, options: options, into: imageView)

源代码部分: 从调用来看，源代码怎么走

主线，网络请求

全局方法，调用入口


public func loadImage(with request: ImageRequestConvertible,options: ImageLoadingOptions = ImageLoadingOptions.shared,into view: ImageDisplayingView,progress: ImageTask.ProgressHandler? = nil,completion: ImageTask.Completion? = nil) -> ImageTask? {// 判断主线程                  assert(Thread.isMainThread)// ImageViewController 是一个 managerlet controller = ImageViewController.controller(for: view)return controller.loadImage(with: request.asImageRequest(), options: options, progress: progress, completion: completion)
}

进入下一层，

胶水层 class ImageViewController


func loadImage(with request: ImageRequest,options: ImageLoadingOptions,progress progressHandler: ImageTask.ProgressHandler? = nil,completion: ImageTask.Completion? = nil) -> ImageTask? {// 重置状态cancelOutstandingTask()// ...let pipeline = options.pipeline ?? ImagePipeline.shared// ...// 通用的两级缓存，先查内存有无图片// ...// 先放占位图task = pipeline.loadImageB(with: request, isMainThreadConfined: true, queue: .main) { [weak self] task, event inswitch event {case .progress:// ...// 过程处理case let .value(response, isCompleted):if isCompleted {// 完成了，展示图片self?.handle(result: .success(response), fromMemCache: false, options: options)completion?(.success(response))} else {if options.isProgressiveRenderingEnabled {self?.handle(partialImage: response, options: options)}progressHandler?(response, task.completedUnitCount, task.totalUnitCount)}case let .error(error):// ...// 错误处理}}return task}

主要做事情的，

是图片下载与解码管道 class ImagePipeline


func loadImage(with request: ImageRequest,isMainThreadConfined: Bool,queue: DispatchQueue?,observer: @escaping (ImageTask, Task<ImageResponse, Error>.Event) -> Void) -> ImageTask {let request = inheritOptions(request)// 拿相关信息，封装为ImageTasklet task = ImageTask(taskId: nextTaskId.increment(), request: request, isMainThreadConfined: isMainThreadConfined, isDataTask: false, queue: queue)task.pipeline = selfself.queue.async {// 开启图片任务// observer 是上面的 event 事件代码self.startImageTask(task, observer: observer)}return task}

开启图片任务


func startImageTask(_ task: ImageTask, observer: @escaping (ImageTask, Task<ImageResponse, Error>.Event) -> Void) {// ...// 获取待解码的图片tasks[task] = getDecompressedImage(for: task.request).subscribe(priority: task._priority) { [weak self, weak task] event inguard let self = self, let task = task else { return }// ...// 事件完成，重置状态if event.isCompleted {self.tasks[task] = nil}// 拿到过程数据，传递给上一步(task.queue ?? self.configuration.callbackQueue).async {guard !task.isCancelled else { return }if case let .progress(progress) = event {task.setProgress(progress)}observer(task, event)}}}

获取图片


func getDecompressedImage(for request: ImageRequest) -> DecompressedImageTask.Publisher {let key = request.makeLoadKeyForFinalImage()return decompressedImageFetchTasks.task(withKey: key, starter: { task in// 实际获取图片的方法，其他都是简单封装self.performDecompressedImageFetchTask(task, request: request)}).publisher}

实际获取图片的方法


func performDecompressedImageFetchTask(_ task: DecompressedImageTask, request: ImageRequest) {// 图片两级缓存// 先看内存中的图片if let image = cachedImage(for: request) {let response = ImageResponse(container: image)if image.isPreview {task.send(value: response)} else {return task.send(value: response, isCompleted: true)}}guard let dataCache = configuration.dataCache, configuration.dataCacheOptions.storedItems.contains(.finalImage), request.cachePolicy != .reloadIgnoringCachedData else {// 下载网络上的资源图片return loadDecompressedImage(for: request, task: task)}// 先看磁盘中的图片let key = cacheKey(for: request, item: .finalImage)let operation = BlockOperation { [weak self, weak task] inguard let self = self, let task = task else { return }// ...// 先拿二进制 Datalet data = dataCache.cachedData(for: key)// ...self.queue.async {if let data = data {//  如果存在二进制数据 Data， 就解码得到 UIImageself.decodeProcessedImageData(data, for: request, task: task)} else {//  如果不存在二进制数据 Data， 就下载网络上的资源图片self.loadDecompressedImage(for: request, task: task)}}}task.operation = operationconfiguration.dataCachingQueue.addOperation(operation)}

I , 下载网络上的资源图片

func loadDecompressedImage(for request: ImageRequest, task: DecompressedImageTask) {// 拿网络请求 request, 去获取图片task.dependency = getProcessedImage(for: request).subscribe(task) { [weak self] image, isCompleted, task in// 拿到最终的图片数据，先编码 encode,存一份到磁盘self?.storeDecompressedImageInDataCache(image, request: request)// 把 Data 数据解压为 UIImage，存一份到内存// 发送完成事件，给回调代码消费self?.decompressProcessedImage(image, isCompleted: isCompleted, for: request, task: task)}}

继续网络请求


func getProcessedImage(for request: ImageRequest) -> ProcessedImageTask.Publisher {guard !request.processors.isEmpty else {// 没有滤镜处理，直接下载return getOriginalImage(for: request) // No processing needed}// 下载，并进行滤镜处理let key = request.makeLoadKeyForFinalImage()return processedImageFetchTasks.task(withKey: key, starter: { task inself.performProcessedImageFetchTask(task, request: request)}).publisher}

进行网络请求图片


func getOriginalImage(for request: ImageRequest) -> OriginalImageTask.Publisher {let key = request.makeLoadKeyForOriginalImage()return originalImageFetchTasks.task(withKey: key, starter: { task inlet context = OriginalImageTaskContext(request: request)// 发起网络请求任务task.dependency = self.getOriginalImageData(for: request).subscribe(task) { [weak self] value, isCompleted, task in// 解码图片，提供回调self?.decodeData(value.0, urlResponse: value.1, isCompleted: isCompleted, task: task, context: context)}}).publisher}

发起网络请求任务


unc getOriginalImageData(for request: ImageRequest) -> OriginalImageDataTask.Publisher {let key = request.makeLoadKeyForOriginalImage()return originalImageDataFetchTasks.task(withKey: key, starter: { task inlet context = OriginalImageDataTaskContext(request: request)if self.configuration.isRateLimiterEnabled {// 延迟加载网络资源图片self.rateLimiter.execute { [weak self, weak task] inguard let self = self, let task = task, !task.isDisposed else {return false}self.performOriginalImageDataTask(task, context: context)return true}} else {// 直接加载，网络资源图片self.performOriginalImageDataTask(task, context: context)}}).publisher}

直接加载，网络资源图片

通过 OperationQueue，限制磁盘缓存的线程并发数目


func performOriginalImageDataTask(_ task: OriginalImageDataTask, context: OriginalImageDataTaskContext) {// 能用磁盘图片，就用磁盘图片// ...let key = cacheKey(for: context.request, item: .originalImageData)let operation = BlockOperation { [weak self, weak task] inguard let self = self, let task = task else { return }let log = Log(self.log, "Read Cached Image Data")log.signpost(.begin)// 不停检查内存图片let data = cache.cachedData(for: key)log.signpost(.end)self.queue.async {if let data = data {task.send(value: (data, nil), isCompleted: true)} else {// 网络资源加载self.loadImageData(for: task, context: context)}}}task.operation = operationconfiguration.dataCachingQueue.addOperation(operation)}

网络资源加载

通过 OperationQueue，限制下载的线程并发数目


func loadImageData(for task: OriginalImageDataTask, context: OriginalImageDataTaskContext) {let operation = Operation(starter: { [weak self, weak task] finish inguard let self = self, let task = task else {return finish()}self.queue.async {self.loadImageData(for: task, context: context, finish: finish)}})configuration.dataLoadingQueue.addOperation(operation)task.operation = operation}

一层又一层的下载图片


func loadImageData(for task: OriginalImageDataTask, context: OriginalImageDataTaskContext, finish: @escaping () -> Void) {// ...// 任务取消的时机var urlRequest = context.request.urlRequest// 做断点下载// ...let dataTask = configuration.dataLoader.loadData(with: urlRequest,didReceiveData: { [weak self, weak task] data, response in// 接收数据的过程guard let self = self, let task = task else { return }self.queue.async {self.imageDataLoadingTask(task, context: context, didReceiveData: data, response: response, log: log)}},completion: { [weak self, weak task] error in// 完成下载finish() // Finish the operation!guard let self = self, let task = task else { return }self.queue.async {log.signpost(.end, "Finished with size \(Log.bytes(context.data.count))")self.imageDataLoadingTask(task, context: context, didFinishLoadingDataWithError: error)}})task.onCancelled = { [weak self] in// 取消下载任务guard let self = self else { return }log.signpost(.end, "Cancelled")dataTask.cancel()finish() // Finish the operation!self.tryToSaveResumableData(for: context)}}

下载图片的最后一环

进入 DataLoader 这个类

调用内部类 _DataLoader

public func loadDataZ(with request: URLRequest,didReceiveData: @escaping (Data, URLResponse) -> Void,completion: @escaping (Swift.Error?) -> Void) -> Cancellable {return impl.loadData(with: request, session: session, didReceiveData: didReceiveData, completion: completion)}

DataLoader 这个类, 初始化的时候，就设置好了网络代理

public init(configuration: URLSessionConfiguration = DataLoader.defaultConfiguration,validate: @escaping (URLResponse) -> Swift.Error? = DataLoader.validate) {let queue = OperationQueue()queue.maxConcurrentOperationCount = 1// 设置网络代理，给内部类 `_DataLoader`self.session = URLSession(configuration: configuration, delegate: impl, delegateQueue: queue)self.impl.validate = validateself.impl.observer = self}

内部类 _DataLoader 具体请求网络

请求网络图片


/// Loads data with the given request.func loadData(with request: URLRequest,session: URLSession,didReceiveData: @escaping (Data, URLResponse) -> Void,completion: @escaping (Error?) -> Void) -> Cancellable {let task = session.dataTask(with: request)let handler = _Handler(didReceiveData: didReceiveData, completion: completion)session.delegateQueue.addOperation { // `URLSession` is configured to use this same queueself.handlers[task] = handler}task.resume()send(task, .resumed)return task}

接收网络图片的数据

    // MARK: URLSessionDelegatefunc urlSession(_ session: URLSession,dataTask: URLSessionDataTask,didReceive response: URLResponse,completionHandler: @escaping (URLSession.ResponseDisposition) -> Void) {send(dataTask, .receivedResponse(response: response))guard let handler = handlers[dataTask] else {completionHandler(.cancel)return}if let error = validate(response) {handler.completion(error)completionHandler(.cancel)return}completionHandler(.allow)}

支线，下载网络上的资源图片，再存一份最终的处理后的 Data 到磁盘

从上面的方法 I , 下载网络上的资源图片 继续

保存最终的图片二进制数据，到磁盘


func storeDecompressedImageInDataCache(_ response: ImageResponse, request: ImageRequest) {//...// 如果不用处理，就没有最终版图片，就算了let context = ImageEncodingContext(request: request, image: response.image, urlResponse: response.urlResponse)let encoder = configuration.makeImageEncoder(context)configuration.imageEncodingQueue.addOperation { [weak self] inguard let self = self else { return }// ...// 先把图片编码为二进制 Datalet encodedData = encoder.encode(response.container, context: context)// ...guard let data = encodedData else { return }let key = self.cacheKey(for: request, item: .finalImage)// 把二进制 Data 写入磁盘dataCache.storeData(data, for: key)}}

把二进制 Data 写入磁盘

进入 DataCache 这个类

public func storeData(_ data: Data, for key: Key) {// 执行操作stage { // 定义行为staging.add(data: data, for: key) }}

执行操作

用了一个线程锁， NSLock


private func stage(_ change: () -> Void) {lock.lock()change()setNeedsFlushChanges()lock.unlock()}

进入磁盘刷新

改标记

private func setNeedsFlushChanges() {guard !isFlushNeeded else { return }isFlushNeeded = truescheduleNextFlush()}

延迟调度


private func scheduleNextFlush() {guard !isFlushScheduled else { return }isFlushScheduled = truequeue.asyncAfter(deadline: .now() + flushInterval, execute: flushChangesIfNeeded)}

写入磁盘，调度

private func flushChangesIfNeeded() {// Create a snapshot of the recently made changeslet staging: Staginglock.lock()guard isFlushNeeded else {return lock.unlock()}staging = self.stagingisFlushNeeded = falselock.unlock()// 写入磁盘performChanges(for: staging)// ...// 调度机制}

写入磁盘,

用了一个自动对象释放池

关心结果，不在意过程，

过程中处理的对象内存较大，autoreleasepool

private func performChanges(for staging: Staging) {autoreleasepool {if let change = staging.changeRemoveAll {perform(change)}for change in staging.changes.values {perform(io: change)}}}

写入磁盘, 关键代码不多，

线程管理，和性能优化上，绕来绕去

通过标记延时操作，尽可能同一时间段，写入磁盘

不会频繁触发写入

关键代码，就创建文件夹，写入数据


private func perform(io change: Staging.Change) {guard let url = url(for: change.key) else {return}switch change.type {case let .add(data):do {// 写入数据try data.write(to: url)} catch let error as NSError {// 写入数据失败 guard error.code == CocoaError.fileNoSuchFile.rawValue && error.domain == CocoaError.errorDomain else { return }// 创建文件夹try? FileManager.default.createDirectory(at: self.path, withIntermediateDirectories: true, attributes: nil)// 写入数据try? data.write(to: url) // re-create a directory and try again}case .remove:try? FileManager.default.removeItem(at: url)}}

代码特色

自动释放池，`autoreleasepool`

上面的文件 IO ，用了一次

图片解码，也用到了

图片解码，耗 CPU, 流程多 ( 颜色模式转换、采样、分块、离散余弦变换 …)

我们只关心处理好的图片

func decode(_ data: Data, urlResponse: URLResponse?, isCompleted: Bool) -> ImageResponse? {func _decode() -> ImageContainer? {if isCompleted {return decode(data)} else {return decodePartiallyDownloadedData(data)}}guard let container = autoreleasepool(invoking: _decode) else {return nil}ImageDecompression.setDecompressionNeeded(true, for: container.image)return ImageResponse(container: container, urlResponse: urlResponse)}

订阅和发布，Subscriber 和 Publisher

苹果官方的框架 Combine ，有 Publisher`

ReactiveCocoa 和 RxSwift 专门搞这个的

Nuke 通过匿名函数 Block 和简单的结构体，封装实现

Task 这个类，包含一个类 Publisher

类 Publisher 包含一个任务，可被订阅，

订阅之后，有 3 种行为，下载失败，下载成功有值，

下载过程中，得到部分数据

final class Task<Value, Error>{struct Publisher {let task: Task// 订阅后，有三种行为// 订阅 Publisher， 就得传入行为// 传入存在事件的时候，需要执行的匿名函数行为func subscribe<NewValue>(_ task: Task<NewValue, Error>, onValue: @escaping (Value, Bool, Task<NewValue, Error>) -> Void) -> TaskSubscription? {return subscribe { [weak task] event inguard let task = task else { return }switch event {case let .value(value, isCompleted):onValue(value, isCompleted, task)case let .progress(progress):task.send(progress: progress)case let .error(error):task.send(error: error)}}}}}

Publisher 结构体的订阅, 走任务的订阅


func subscribeX(priority: TaskPriority = .normal, _ observer: @escaping (Event) -> Void) -> TaskSubscription? {task.subscribe(priority: priority, observer)}

Task 类中的订阅，是建立任务和行为之间的对应关系，并存为属性

private func subscribe(priority: TaskPriority = .normal, _ observer: @escaping (Event) -> Void) -> TaskSubscription? {// ...// 建立任务索引nextSubscriptionId += 1let subscriptionKey = nextSubscriptionIdlet subscription = TaskSubscription(task: self, key: subscriptionKey)//  建立任务和行为之间的对应关系，并存为属性subscriptions[subscriptionKey] = Subscription(observer: observer, priority: priority)//...return subscription}

建立订阅之后，需要的时候，

可以发布事件，去执行

Task 类中

func send(value: Value, isCompleted: Bool = false) {send(event: .value(value, isCompleted: isCompleted))}func send(error: Error) {send(event: .error(error))}func send(progress: TaskProgress) {send(event: .progress(progress))}private func send(event: Event) {guard !isDisposed else { return }switch event {case let .value(_, isCompleted):if isCompleted {terminate(reason: .finished)}case .progress:break // Simply send the eventcase .error:terminate(reason: .finished)}for context in subscriptions.values {context.observer(event)}}

示例 repo