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

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

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

这时候有了一个 Datagit

2，对图片 Data 解码，

即拿 Data，生成 UIImage,github

3，显示图片，

把生成的 UIImage, 赋给 UIImageView 的 imageapi

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 是一个 manager
    let controller = ImageViewController.controller(for: view)
    return controller.loadImage(with: request.asImageRequest(), options: options, progress: progress, completion: completion)
}
复制代码

进入下一层，性能优化

胶水层 class ImageViewControllermarkdown

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 in
            switch 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 ImagePipelinesession

func loadImage(with request: ImageRequest,
                   isMainThreadConfined: Bool,
                   queue: DispatchQueue?,
                   observer: @escaping (ImageTask, Task<ImageResponse, Error>.Event) -> Void) -> ImageTask {
        let request = inheritOptions(request)
        // 拿相关信息，封装为ImageTask
        let task = ImageTask(taskId: nextTaskId.increment(), request: request, isMainThreadConfined: isMainThreadConfined, isDataTask: false, queue: queue)
        task.pipeline = self
        self.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 in
                guard 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] in
            guard let self = self, let task = task else { return }

            // ...
            // 先拿二进制 Data
            let data = dataCache.cachedData(for: key)
            // ...

            self.queue.async {
                
                if let data = data {
                    //  若是存在二进制数据 Data， 就解码获得 UIImage
                    self.decodeProcessedImageData(data, for: request, task: task)
                } else {
                    //  若是不存在二进制数据 Data， 就下载网络上的资源图片
                    self.loadDecompressedImage(for: request, task: task)
                }
            }
        }
        task.operation = operation
        configuration.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 in
            self.performProcessedImageFetchTask(task, request: request)
        }).publisher
    }
复制代码

进行网络请求图片

func getOriginalImage(for request: ImageRequest) -> OriginalImageTask.Publisher {
        let key = request.makeLoadKeyForOriginalImage()
        return originalImageFetchTasks.task(withKey: key, starter: { task in
            let 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 in
            let context = OriginalImageDataTaskContext(request: request)
            if self.configuration.isRateLimiterEnabled {
                // 延迟加载网络资源图片
                self.rateLimiter.execute { [weak self, weak task] in
                    guard 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] in
            guard 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 = operation
        configuration.dataCachingQueue.addOperation(operation)
    }
复制代码

网络资源加载

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

func loadImageData(for task: OriginalImageDataTask, context: OriginalImageDataTaskContext) {
        
        let operation = Operation(starter: { [weak self, weak task] finish in
            guard 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 = validate
        self.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 queue
            self.handlers[task] = handler
        }
        task.resume()
        send(task, .resumed)
        return task
    }

复制代码

接收网络图片的数据

// MARK: URLSessionDelegate
    
    
    

    func 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] in
            guard let self = self else { return }

            // ...
            // 先把图片编码为二进制 Data
            let 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 = true
        scheduleNextFlush()
    }
复制代码

延迟调度

private func scheduleNextFlush() {
        guard !isFlushScheduled else { return }
        isFlushScheduled = true
        queue.asyncAfter(deadline: .now() + flushInterval, execute: flushChangesIfNeeded)
    }
复制代码

写入磁盘，调度

private func flushChangesIfNeeded() {
        // Create a snapshot of the recently made changes
        let staging: Staging
        lock.lock()
        guard isFlushNeeded else {
            return lock.unlock()
        }
        staging = self.staging
        isFlushNeeded = false
        lock.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 in
                guard 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 += 1
        let subscriptionKey = nextSubscriptionId
        let 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 event
        case .error:
            terminate(reason: .finished)
        }

        for context in subscriptions.values {
            context.observer(event)
        }
    }



复制代码