Understanding and Using Schedulers in Swift Combine

The article explores how to manage task execution in Combine using Schedulers. It assumes readers are familiar with Combine basics and have some knowledge of GCD, threads, and RunLoop.

It outlines the following topics:

What is a Scheduler?

Combine's default scheduling mechanism

Operators for switching Schedulers

In‑depth examples of Scheduler behavior

Types of Schedulers provided by Apple and their usage

What is a Scheduler

A Scheduler is a protocol that defines when and where a closure is executed. "Where" refers to a RunLoop, DispatchQueue, or an operator queue, while "when" refers to the virtual time of the Combine event stream. The concrete Scheduler implementation (e.g., DispatchQueue, OperationQueue) determines the actual thread, but the Scheduler itself does not directly reference threads.

Combine's Default Scheduling Mechanism

When no Scheduler is specified, Combine forwards events on the thread where the upstream Publisher emits them. The following example demonstrates this behavior:

var cancellables = Set
()

let intSubject = PassthroughSubject
()

intSubject.sink(receiveValue: { value in
    print(value)
    print(Thread.current)
}).store(in: &cancellables)

intSubject.send(1)

DispatchQueue.global().async {
    intSubject.send(2)
}

let queue = OperationQueue()
queue.maxConcurrentOperationCount = 1
queue.underlyingQueue = DispatchQueue(label: "com.donnywals.queue")

queue.addOperation {
    intSubject.send(3)
}

Output shows that the sink receives values on three different threads, reflecting the thread that performed each .send() call.

Operators for Switching Schedulers

Two main operators control Scheduler switching:

subscribe(on:) – influences the upstream Publisher and any operators before it.

receive(on:) – influences downstream operators after its position.

receive(on:)

It only affects operators that appear after it in the chain:

publisher
    .operatorA
    .receive(on:)
    .operatorB
    .sink

Thus, operatorB and the sink run on the specified Scheduler, while operatorA runs on the original context.

subscribe(on:)

This operator controls the execution of the Publisher, its subscription, and any upstream operators. Its effect depends on the Publisher’s internal implementation; if the Publisher explicitly chooses a thread, the Scheduler may be ignored.

In‑Depth Example

The article provides a custom Publisher ( ExpensiveComputation ) and demonstrates how subscribe(on:) and receive(on:) affect where the computation and the final sink run.

final class ComputationSubscription
: Subscription {
    private let duration: TimeInterval
    private let sendCompletion: () -> Void
    private let sendValue: (Output) -> Subscribers.Demand
    private let finalValue: Output
    private var cancelled = false
    // ... initializer and request/cancel implementation ...
}

extension Publishers {
    public struct ExpensiveComputation: Publisher {
        public typealias Output = String
        public typealias Failure = Never
        public let duration: TimeInterval
        public init(duration: TimeInterval) { self.duration = duration }
        public func receive
(subscriber: S) where S : Subscriber, Failure == S.Failure, Output == S.Input {
            // ... create subscription and send values after a delay ...
        }
    }
}

let computationPublisher = Publishers.ExpensiveComputation(duration: 3)
let queue = DispatchQueue(label: "serial queue")
let subscription = computationPublisher
    .subscribe(on: queue)
    .map { val -> String in
        print("Thread.current.number=\(Thread.current.number)")
        return val
    }
    .receive(on: DispatchQueue.main)
    .sink { value in
        let thread = Thread.current.number
        print("Received computation result on thread \(thread): '\(value)'")
    }

The output shows that the heavy computation runs on a background thread, while the final sink runs on the main thread.

System Publishers

Examples with PassthroughSubject and CurrentValueSubject illustrate how the thread that calls .send() determines where the downstream receives values, regardless of subscribe(on:) unless the Scheduler is explicitly used inside the Publisher.

PassthroughSubject Example

let intSubject = PassthroughSubject
()

intSubject
    .subscribe(on: DispatchQueue.global())
    .sink { value in
        print(Thread.current)
    }
    .store(in: &cancellables)

intSubject.send(1)
intSubject.send(2)
intSubject.send(3)

Even though subscribe(on:) switches to a background queue, the sink still runs on the main thread because the subject sends values from the main thread.

CurrentValueSubject Example

let intSubject = CurrentValueSubject
(0)

intSubject
    .subscribe(on: DispatchQueue.global())
    .map { num in
        print("num=\(num), map: \(Thread.current)")
        return num + 1
    }
    .receive(on: DispatchQueue.global())
    .sink { value in
        print("num=\(value), sink: \(Thread.current)")
    }
    .store(in: &cancellables)

sleep(5)
intSubject.send(1)
intSubject.send(2)

The initial value is emitted on the thread dictated by subscribe(on:) , while subsequent values follow the thread that performs .send() , and the final sink runs on the Scheduler specified by receive(on:) .

Scheduler Types Provided by Apple

ImmediateScheduler – Executes work on the current thread.

RunLoop – Associates with a thread’s RunLoop; useful for timers.

DispatchQueue – Serial or concurrent queues; most commonly used for background work.

OperationQueue – Higher‑level queue that can limit concurrency.

ImmediateScheduler Example

var subscription = Set
()
let source = Timer.publish(every: 1.0, on: .main, in: .common).autoconnect().scan(0) { counter, _ in counter + 1 }

source
    .map { value -> Int in
        print("1: \(Thread.current)")
        return value
    }
    .receive(on: ImmediateScheduler.shared)
    .map { value -> Int in
        print("2: \(Thread.current)")
        return value
    }
    .eraseToAnyPublisher()
    .sink { _ in }
    .store(in: &subscription)

Both maps run on the main thread because ImmediateScheduler executes on the current thread.

RunLoop Example

source
    .receive(on: DispatchQueue.global())
    .map { value -> Int in
        print("1: \(Thread.current)")
        return value
    }
    .receive(on: RunLoop.current)
    .map { value -> Int in
        print("2: \(Thread.current)")
        return value
    }
    .eraseToAnyPublisher()
    .sink { _ in }
    .store(in: &subscription)

The first map runs on a background thread, while the second map runs on the main thread’s RunLoop.

DispatchQueue Example

let serialQueue = DispatchQueue(label: "Serial queue")
let sourceQueue = DispatchQueue.main
let source = PassthroughSubject
()

source
    .map { _ in
        print("1: \(Thread.current)")
    }
    .receive(on: serialQueue, options: DispatchQueue.SchedulerOptions(qos: .userInteractive))
    .map { _ in
        print("2: \(Thread.current)")
    }
    .eraseToAnyPublisher()
    .sink { _ in }
    .store(in: &subscriptions)

DispatchQueue does not guarantee a fixed thread; the same queue may execute on different threads each time.

OperationQueue Example

let queue = OperationQueue()
(1...10).publisher
    .receive(on: queue)
    .sink { value in
        print("Received \(value) on thread \(Thread.current)")
    }

OperationQueue may run on multiple threads; setting maxConcurrentOperationCount = 1 makes it behave like a serial queue.

Conclusion

When performing time‑consuming or resource‑intensive work in a Combine pipeline, appropriate use of Schedulers prevents blocking the main thread. Schedulers provide a logical abstraction for choosing execution contexts (main vs. background) without binding to a specific thread. Among the available implementations, DispatchQueue is generally the preferred choice for iOS development.

References

https://developer.apple.com/documentation/combine/scheduler

https://www.vadimbulavin.com/understanding-schedulers-in-swift-combine-framework/

Combine Asynchronous Programming with Swift – raywenderlich.com

Practical Combine – practicalcombine.com