Swift 线程安全数组

有并发的地方就存在线程安全问题,尤其是对于 Swift 这种还没有内置并发支持的语言来说线程安全问题更为突出。下面我们通过常见的数组操作来分析其中存在的线程问题,以及如何实现一个线程安全数组。

问题所在

因为无法确定执行顺序,所以并发导致的问题一般都很难模拟和测试。不过我们可以通过下面这段代码来模拟一个并发情形下导致的数据竞争问题。

var array = [Int]()
 
DispatchQueue.concurrentPerform(iterations: 1000) { index in
    let last = array.last ?? 0
    array.append(last + 1)
}

这段代码中我们对数组 array 进行了 1000 次并发修改操作,虽然有些夸张但是它能很好的揭示一些并发环境下数组写操作存在的一些问题。因为对于值类型来说 Swift 采用的是 Copy On Write 机制,所以在进行 Copy On Write 处理是可能数组已经被另一个写操作给修改了。这就造成了数组中元素和数据的丢失现象,如下:

Unsafe loop count: 988.
Unsafe loop count: 991.
Unsafe loop count: 986.
Unsafe loop count: 995.

串行队列

这应该是大家都能想到的一种最常见处理方式。 由于串行队列每次都只能运行一个进程,所以即使有多个数组写操作进程我们也能确保资源的互斥访问。这样数组是从设计的并发进程安全的。

let queue = DispatchQueue(label: "SafeArrayQueue")
 
queue.async() {
  // 写操作
}
 
queue.sync() {
  // 读操作
}

由于写操作并不需要返回操作结果,所有这里可以使用异步的方式进行。而对于读操作来说则必须采用同步的方式实时返回操作结果。但是串行队列有一个最为明显的缺陷:多个读操作之间也是互斥的。很显然这种方式太过粗暴存在明显的性能问题,毕竟读操作的频率直觉上是要高过写操作的。

并发队列

采用并发队列我们就可以很好的解决上面提到的多个读操作的性能问题,不过随之而来的就是写操作的数据竞争。这与我们在学习操作系统是的 读者-作者 问题本质上是一类问题,我们可以通过共享互斥锁来解决写操作的数据竞争问题。对于 iOS 来说它就是 GCD 中的写栏栅 barrier 机制。

let queue = DispatchQueue(label: "SafeArrayQueue", attributes: .concurrent)
 
queue.async(flags: .barrier) {
  // 写操作
}
 
queue.sync() {
  // 读操作
}

上面代码中我们对异步的写操作设置了 barrier 标示,这意味着在执行异步操作代码的时候队列不能执行其他代码。而对于同步的读操作来说,由于是并发队列同时读取数据并不会存在任何性能问题。

实践

/// A thread-safe array.
public class SafeArray<Element> {
    fileprivate let queue = DispatchQueue(label: "Com.BigNerdCoding.SafeArray", attributes: .concurrent)
    fileprivate var array = [Element]()
}
 
// MARK: - Properties
public extension SafeArray {
 
    var first: Element? {
        var result: Element?
        queue.sync { result = self.array.first }
        return result
    }
 
    var last: Element? {
        var result: Element?
        queue.sync { result = self.array.last }
        return result
    }
 
    var count: Int {
        var result = 0
        queue.sync { result = self.array.count }
        return result
    }

    var isEmpty: Bool {
        var result = false
        queue.sync { result = self.array.isEmpty }
        return result
    }
    
    var description: String {
        var result = ""
        queue.sync { result = self.array.description }
        return result
    }
}
 
// MARK: - 读操作
public extension SafeArray {
    func first(where predicate: (Element) -> Bool) -> Element? {
        var result: Element?
        queue.sync { result = self.array.first(where: predicate) }
        return result
    }
    
    func filter(_ isIncluded: (Element) -> Bool) -> [Element] {
        var result = [Element]()
        queue.sync { result = self.array.filter(isIncluded) }
        return result
    }
    
    func index(where predicate: (Element) -> Bool) -> Int? {
        var result: Int?
        queue.sync { result = self.array.index(where: predicate) }
        return result
    }
    
    func sorted(by areInIncreasingOrder: (Element, Element) -> Bool) -> [Element] {
        var result = [Element]()
        queue.sync { result = self.array.sorted(by: areInIncreasingOrder) }
        return result
    }
    
    func flatMap<ElementOfResult>(_ transform: (Element) -> ElementOfResult?) -> [ElementOfResult] {
        var result = [ElementOfResult]()
        queue.sync { result = self.array.flatMap(transform) }
        return result
    }
 
    func forEach(_ body: (Element) -> Void) {
        queue.sync { self.array.forEach(body) }
    }
    
    func contains(where predicate: (Element) -> Bool) -> Bool {
        var result = false
        queue.sync { result = self.array.contains(where: predicate) }
        return result
    }
}
 
// MARK: - 写操作
public extension SafeArray {
 
    func append( _ element: Element) {
        queue.async(flags: .barrier) {
            self.array.append(element)
        }
    }
 
    func append( _ elements: [Element]) {
        queue.async(flags: .barrier) {
            self.array += elements
        }
    }
 
    func insert( _ element: Element, at index: Int) {
        queue.async(flags: .barrier) {
            self.array.insert(element, at: index)
        }
    }
 
    func remove(at index: Int, completion: ((Element) -> Void)? = nil) {
        queue.async(flags: .barrier) {
            let element = self.array.remove(at: index)
            
            DispatchQueue.main.async {
                completion?(element)
            }
        }
    }
    
    func remove(where predicate: @escaping (Element) -> Bool, completion: ((Element) -> Void)? = nil) {
        queue.async(flags: .barrier) {
            guard let index = self.array.index(where: predicate) else { return }
            let element = self.array.remove(at: index)
            
            DispatchQueue.main.async {
                completion?(element)
            }
        }
    }
 
    func removeAll(completion: (([Element]) -> Void)? = nil) {
        queue.async(flags: .barrier) {
            let elements = self.array
            self.array.removeAll()
            
            DispatchQueue.main.async {
                completion?(elements)
            }
        }
    }
}
 
public extension SafeArray {

    subscript(index: Int) -> Element? {
        get {
            var result: Element?
            
            queue.sync {
                guard self.array.startIndex..<self.array.endIndex ~= index else { return }
                result = self.array[index]
            }
            
            return result
        }
        set {
            guard let newValue = newValue else { return }
            
            queue.async(flags: .barrier) {
                self.array[index] = newValue
            }
        }
    }
}
 
 
// MARK: - Equatable
public extension SafeArray where Element: Equatable {

    func contains(_ element: Element) -> Bool {
        var result = false
        queue.sync { result = self.array.contains(element) }
        return result
    }
}
 
// MARK: - 自定义操作符
public extension SynchronizedArray {
 
    static func +=(left: inout SynchronizedArray, right: Element) {
        left.append(right)
    }
 
    static func +=(left: inout SynchronizedArray, right: [Element]) {
        left.append(right)
    }
}

通过 filePrivate 属性 arrayqueueSafeArray 成功的实现了大多数数组常用功能,更为关键的是该类型并发安全:所有的写操作都通过 barrier 方式的异步进行,而读操作则与内置 Array 没有什么区别。

需要注意的是:我们使用同样的方式可以实现并发安全的 Dictionary 类似:SynchronizedDictionary。

接下来,我们可以对传统的非并发安全数组和 SafeArray 进行以下比较:

import Foundation
import PlaygroundSupport
 
// Thread-unsafe array
do {
    var array = [Int]()
    var iterations = 1000
    let start = Date().timeIntervalSince1970
 
    DispatchQueue.concurrentPerform(iterations: iterations) { index in
        let last = array.last ?? 0
        array.append(last + 1)
 
        DispatchQueue.global().sync {
            iterations -= 1
        
            // Final loop
            guard iterations <= 0 else { return }
            let message = String(format: "Unsafe loop took %.3f seconds, count: %d.",
                Date().timeIntervalSince1970 - start,
                array.count)
            print(message)
        }
    }
}
 
// Thread-safe array
do {
    var array = SafeArray<Int>()
    var iterations = 1000
    let start = Date().timeIntervalSince1970
 
    DispatchQueue.concurrentPerform(iterations: iterations) { index in
        let last = array.last ?? 0
        array.append(last + 1)
 
        DispatchQueue.global().sync {
            iterations -= 1
        
            // Final loop
            guard iterations <= 0 else { return }
            let message = String(format: "Safe loop took %.3f seconds, count: %d.",
                Date().timeIntervalSince1970 - start,
                array.count)
            print(message)
        }
    }
}
 
PlaygroundPage.current.needsIndefiniteExecution = true

得到的输出可能如下:

Unsafe loop took 1.031 seconds, count: 989.
Safe loop took 1.363 seconds, count: 1000.

虽然由于使用了 GCD 机制导致速度慢了 30% 左右并且使用了更多的内存,但是与之对应的是我们实现了一个并发安全的数组类型。

原文地址

原创声明,本文系作者授权云+社区发表,未经许可,不得转载。

如有侵权,请联系 yunjia_community@tencent.com 删除。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏互联网杂技

SpringBoot ( 八 ) :RabbitMQ 详解

RabbitMQ 即一个消息队列,主要是用来实现应用程序的异步和解耦,同时也能起到消息缓冲,消息分发的作用。

752
来自专栏纯洁的微笑

springboot(八):RabbitMQ详解

RabbitMQ 即一个消息队列,主要是用来实现应用程序的异步和解耦,同时也能起到消息缓冲,消息分发的作用。 消息中间件在互联网公司的使用中越来越多,刚才还看到...

3204
来自专栏DT乱“码”

Memcached使用实例

package com.memcached.util; import java.io.BufferedWriter; import java.io.FileW...

1926
来自专栏Java进阶

TCP IP 里的网络请求 到底是什么样的?

35211
来自专栏风中追风

分布式基础__TCP IP 里的网络请求 到底是什么样的?

当我输入 一个url 在浏览器的地址栏按下回车 这个以前也写过博客了:一次URL输入域名按下回车到底发生了什么?

38216
来自专栏拭心的安卓进阶之路

Android 进阶12:进程通信之 Socket (顺便回顾 TCP UDP)

不要害怕困难,这是你进步的机会! 前面几篇文章我们介绍了 AIDL 、Binder、Messenger 以及 ContentProvider 实现进程通信的方式...

2487
来自专栏彭湖湾的编程世界

【react】关于react框架使用的一些细节要点的思考

( _(:3 」∠)_给云友们提个建议,无论是API文档还是书籍,一定要多看几遍!特别是隔一段时间后,会有意想不到的收获的) 这篇文章主要是写关于学习react...

1928
来自专栏运维

yum安装rabbitmq3.6.11与erlange20配置及优化

AMQP,即Advanced Message Queuing Protocol,高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。消息中间件...

491
来自专栏扎心了老铁

python分布式环境下的限流器

项目中用到了限流,受限于一些实现方式上的东西,手撕了一个简单的服务端限流器。 服务端限流和客户端限流的区别,简单来说就是: 1)服务端限流 对接口请求进行限流,...

4248
来自专栏三杯水

RabbitMQ3.6.11部署配置及优化

AMQP,即Advanced Message Queuing Protocol,高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。消息中间件...

791

扫码关注云+社区