Hello, 异步函数#

func hello() async -> String {
    try? await Task.sleep(nanoseconds: 1000_000_000)
    return "hello"
}

Kotlin:

suspend fun hello(): String {
    delay(1.seconds)
    return "hello"
}

结构化并发#

let task = Task {
    await hello()
}
task.cancel()

Kotlin:

val scope = CoroutineScope(EmptyCoroutineContext)
scope.launch(Dispatchers.Main) {
    hello()
}
scope.cancel()

public init(priority: TaskPriority? = nil, operation: @escaping @Sendable () async -> Success)

Task(priority: .high) {
    ...
}

high, medium, low,
userInitiated, utility, background,

withUnsafeCurrentTask { task in
    task?.isCancelled
    task?.priority
    task?.cancel()
    ...
}

Task {
    Task {
        ...
    }
}

Task {
    Task.detached {
        ...
    }
}

public final class CancelBag {

    private var cancels = [() -> Void]()

    public func on(cancel: @escaping () -> Void) {
        self.cancels.append(cancel)
    }

    private func cancel() {
        cancels.forEach { cancel in cancel() }
    }
    
    deinit {
        self.cancel()
    }
}

extension Task {
    func cancel(by bag: CancelBag) {
        bag.on(cancel: { cancel() })
    }
}

let bag = CancelBag()

Task {
    ...
}.cancel(by: bag)

public final class CancelBag {

    private var cancels = [() -> Void]()

    public func on(cancel: @escaping () -> Void) {
        self.cancels.append(cancel)
    }

    public func cancel() {
        cancels.forEach { cancel in cancel() }
        cancels.removeAll()
    }
}

async let#

async let hello = hello()
async let world = world()
print("\(await hello) \(await world)")

Kotlin:

val hello = async { hello() }
val world = async { world() }
println("${hello.await()} ${world.await()}")

同时执行多个async函数. 比如发送网络请求时, 如果第三个请求的参数依赖前两个请求的结果:

async let x = reqX()
async let y = reqY()
let z = await reqZ(await x, await y)

可以让x, y请求同时发出, 都得到响应后, 再发出z请求

for await#

func number() async -> Int {
    .random(in: 1...100)
}

await withTaskGroup(of: Int.self) { group in
    for _ in 1...100 {
        group.addTask { await number() }
    }
    
    for await i in group {
        print(i)
    }
}

Kotlin:

suspend fun number() = Random.nextInt(0..100)

val numbers = produce {
    coroutineScope {
        repeat(100) {
            launch { send(number()) }
        }
    }
}

for (i in numbers) {
    println(i)
}

for await用于迭代实现了AsyncSequence协议的类型, 如TaskGroup和下文将要提到的AsyncStream.

withTaskGroup(of:)效果类似于corountineScope, 可以添加一系列子任务并等待这些任务完成.

如果子Task会调用throws, 可以使用withThrowingTaskGroup:

try await withThrowingTaskGroup(of: Int.self) { group in
    group.addTask { throw SomeError() }

    for try await i in group {
	...
    }
}

await all#

await withTaskGroup(of: Int.self) { group in
    for _ in 1...100 {
        group.addTask { await number() }
    }
    
    await group.reduce(into: []) { result, transition in result.append(transition) }
}

(1..100)
    .map { async { number() } }
    .awaitAll()

收集子任务结果.

AsyncStream#

let stream = AsyncStream<Int> { continuation in
    for i in 1...100 {
        continuation.yield(i)
    }
    continuation.finish()
}

for await i in stream {
    print(i)
}

let stream = AsyncStream<Int> {
    try? await Task.sleep(nanoseconds: 1000_000_000)
    return .random(in: 1...100)
}

for await i in stream {
    print(i)
}

Kotlin:

flow {
    (1..10).forEach { emit(it) }
}.collect {
    println(it)
}

flow {
    while (true) {
        delay(1.seconds)
        emit(Random.nextInt(1..10))
    }
}.collect {
    println(it)
}

这里AsyncStream和flow对比比较直观, 但其行为更像ReceiveChannel, 因为flow是冷的.

Actor#

actor Counter {
    var count: Int = 0
    
    func inc() {
        count += 1
    }
}

let couter = Counter()
await couter.inc()
let count = await couter.count

Kotlin:

sealed interface CounterMsg

object IncCount : CounterMsg
class GetCount(val response: CompletableDeferred<Int> = CompletableDeferred()) : CounterMsg

fun CoroutineScope.counterActor() = actor<CounterMsg> {
    var count = 0
    for (msg in channel) {
        when (msg) {
            is GetCount -> msg.response.complete(count)
            is IncCount -> ++count
        }
    }
}

suspend fun SendChannel<CounterMsg>.count() = GetCount().also { send(it) }.response.await()

val counter = counterActor()

counter.send(IncCount)
val count = counter.count()

Swift在语言层面支持了actor, 因此使用actor要比Kotlin方便许多.

从外部调用actor中的函数、可变属性和计算属性需要使用await. 如果确定不存在数据竞争, 可以用nonisolated修饰:

nonisolated func foo() -> String {
    ...
}

这样就可以像调用普通函数一样从外部调用.

@globalActor
actor BusActor: GlobalActor {
    public static let shared = BusActor()
}

@BusActor func foo() {
    ...
}

@BusActor var bar = 0

@BusActor class Bus {}

@globalActor final public actor MainActor : GlobalActor

@MainActor
func update() {
    ...
}

public protocol Sendable {
}

class Foo: @unchecked Sendable {
}

从Callback迁移#

func request(_ completion: (String) -> Void) {
    ...
}

func request() async -> String {
    await withCheckedContinuation { continuation in
        request { response in
            continuation.resume(returning: response)
        }
    }
}

Kotlin:

fun request(callback: (String) -> Unit) {
    ...
}

suspend fun request(): String = suspendCoroutine { continuation ->
    request {
        continuation.resume(it)
    }
}

如果闭包内调用了throws, 可以使用withCheckedThrowingContinuation(function:_:)

如果可以确保调用且仅调用一次resume, 可以使用unsafe版本:

• withUnsafeContinuation(_:)

• withUnsafeThrowingContinuation(_:)