当前位置: 首页 > news >正文

Swift Combine 学习(六):自定义 Publisher 和 Subscriber

  • Swift Combine 学习(一):Combine 初印象
  • Swift Combine 学习(二):发布者 Publisher
  • Swift Combine 学习(三):Subscription和 Subscriber
  • Swift Combine 学习(四):操作符 Operator
  • Swift Combine 学习(五):Backpressure和 Scheduler
  • Swift Combine 学习(六):自定义 Publisher 和 Subscriber
  • Swift Combine 学习(七):实践应用场景举例

    文章目录

      • 引言
      • 错误处理和重试机制
      • 调试 Combine 代码
      • 自定义 Publisher 和 Subscriber
      • 结语

引言

在前面的文章中,我们已经学习了 Combine 框架的核心概念和基础组件。本文将探讨如何自定义 Publisher 和 Subscriber,以满足特定的应用需求。通过自定义这些组件,开发者可以创建更加灵活和强大的数据流处理逻辑,适应不同的应用场景。

错误处理和重试机制

Combine 提供了多种处理错误和实现重试机制方法。以下是一些常用的错误处理操作符:

  1. 使用 tryMap 进行错误检查和抛出。
  2. 使用 retry 操作符在失败时进行重试。
  3. 使用 catch 操作符处理错误并提供 fallback 值。
import Combine
import Foundationenum ErrorType: Error {case numberTooLarge
}var cancellables = Set<AnyCancellable>()let numbers = [2, 5, 11, 99].publisher.tryMap { number -> Int in// 检查数字是否大于10guard number <= 10 else {throw ErrorType.numberTooLarge}return number * 2}.retry(1).catch { error -> AnyPublisher<Int, Never> in// 出错就默认返回 0print("❌ 错误: \(error)")return Just(0).eraseToAnyPublisher()}.eraseToAnyPublisher()print("🍎 开始处理数字")
numbers.sink { number inprint("📍 结果: \(number)")}.store(in: &cancellables)/*输出:
🍎 开始处理数字
📍 结果: 4
📍 结果: 10
📍 结果: 4
📍 结果: 10
❌ 错误: numberTooLarge
📍 结果: 0
*/

调试 Combine 代码

响应式编程因为传统的堆栈跟踪信息不足、异步执行和线程切换、一些操作符的链式调用可能使得代码逻辑比较抽象等等原因导致其一大痛点就是出现 bug 不好排查。Swift Combine 提供了几个有用的操作符来帮助调试:

  1. 使用 print 打印所有事件。
  2. 使用 breakpoint 在特定条件下触发断点。
  3. 使用 handleEvents 在发布者生命周期的各个阶段插入自定义的调试代码。
import Combineclass CombineDebugDemo {private var cancellables = Set<AnyCancellable>()// MARK: - 使用 print 操作符追踪数据流func basicDebugDemo() {let numbers = (6...7).publisherlet printDemo = numbers.print("🔍 数据流追踪")printDemo.sink { print("Print演示完成: \($0)") }receiveValue: { print("Print值: \($0)") }.store(in: &cancellables)}// MARK: - breakpoint 条件断点func breakpointDemo() {let numbers = (7...9).publisherlet breakpointDemo = numbers.breakpoint(receiveOutput: { value inlet shouldBreak = value > 10print("⚡️ 断点检查: 值 = \(value), 是否触发 = \(shouldBreak)")return shouldBreak})breakpointDemo.sink { print("断点演示完成: \($0)") }receiveValue: { print("断点值: \($0)") }.store(in: &cancellables)}// MARK: - 使用 handleEvents 监控完整生命周期func handleEventsDemo() {let numbers = (5...7).publisherlet handleEventsDemo = numbers.handleEvents(receiveSubscription: { subscription inprint("🌟订阅开始: \(subscription)")},receiveOutput: { value inprint("⌛️准备发送值: \(value)")},receiveCompletion: { completion inprint("✅发送完成: \(completion)")},receiveCancel: {print("❌订阅取消")},receiveRequest: { demand inprint("📧收到需求: \(demand)")})handleEventsDemo.sink { print("事件处理演示完成: \($0)") }receiveValue: { print("事件处理值: \($0)") }.store(in: &cancellables)}// MARK: - 综合例子展示func comprehensiveDebugDemo() {let numbers = (16...18).publishernumbers// 1. 原始数据.print("\n1️⃣ 原始数据")// 2. 添加条件断点(可选).breakpoint(receiveOutput: { value inlet shouldBreak = value > 16print("2️⃣ 断点检查: 值 = \(value), 触发 = \(shouldBreak)")return shouldBreak})// 3. 完整的生命周期输出.handleEvents(receiveSubscription: { _ in print("3️⃣ 订阅开始") },receiveOutput: { print("3️⃣ 输出值: \($0)") },receiveCompletion: { print("3️⃣ 完成: \($0)") },receiveCancel: { print("3️⃣ 取消") },receiveRequest: { print("3️⃣ 需求: \($0)") }).sink(receiveCompletion: { print("4️⃣ 最终完成: \($0)") },receiveValue: { print("4️⃣ 最终值: \($0)") }).store(in: &cancellables)}
}let demo = CombineDebugDemo()print("\n🍎基础 print")
demo.basicDebugDemo()print("\n🍎断点演示")
demo.breakpointDemo()print("\n🍎事件处理演示")
demo.handleEventsDemo()print("\n🍎综合调试演示")
demo.comprehensiveDebugDemo()/*输出🍎基础 print
🔍 数据流追踪: receive subscription: (6...7)
🔍 数据流追踪: request unlimited
🔍 数据流追踪: receive value: (6)
Print值: 6
🔍 数据流追踪: receive value: (7)
Print值: 7
🔍 数据流追踪: receive finished
Print演示完成: finished🍎断点演示
⚡️ 断点检查: 值 = 7, 是否触发 = false
断点值: 7
⚡️ 断点检查: 值 = 8, 是否触发 = false
断点值: 8
⚡️ 断点检查: 值 = 9, 是否触发 = false
断点值: 9
断点演示完成: finished🍎事件处理演示
🌟订阅开始: 5...7
📧收到需求: unlimited
⌛️准备发送值: 5
事件处理值: 5
⌛️准备发送值: 6
事件处理值: 6
⌛️准备发送值: 7
事件处理值: 7
✅发送完成: finished
事件处理演示完成: finished🍎综合调试演示1️⃣ 原始数据: receive subscription: (16...18)
3️⃣ 订阅开始
3️⃣ 需求: unlimited1️⃣ 原始数据: request unlimited1️⃣ 原始数据: receive value: (16)
2️⃣ 断点检查: 值 = 16, 触发 = false
3️⃣ 输出值: 16
4️⃣ 最终值: 161️⃣ 原始数据: receive value: (17)
2️⃣ 断点检查: 值 = 17, 触发 = true
3️⃣ 输出值: 17
4️⃣ 最终值: 171️⃣ 原始数据: receive value: (18)
2️⃣ 断点检查: 值 = 18, 触发 = true
*/

自定义 Publisher 和 Subscriber

iOS 大部分场景下开发者无需自定义 Publisher,因为有 KVO 、 Notification 等。不过有时可能需要创建自定义的 Publisher 或 Subscriber 来满足特定需求。比如封装已经有的异步 API 、有特殊的数据传递需求、实现一些当前 Combine 操作符无法满足的功能的时候。

  • 创建一个自定义的 TimerPublisher。这个 TimerPublisher 将模拟一个计时器,每秒发布一个整数值。然后写一个自定义 TimerSubscriber 用于接收从 TimerPublisher 发布的值,并做相应的处理,例如在控制台中打印出接收到的值。

    import Combine
    import Foundation// 自定义 Publisher
    class TimerPublisher: Publisher, @unchecked Sendable {// 定义 Publisher 的输出类型和失败类型typealias Output = Inttypealias Failure = Neverprivate var counter = 0private var timer: Timer?// 使用 dictionary 存储多个订阅者及其需求private var subscribers: [UUID: (subscriber: AnySubscriber<Output, Failure>, demand: Subscribers.Demand)] = [:]deinit {stop()}// 接受 Subscriber 并建立连接func receive<S>(subscriber: S) where S : Subscriber, TimerPublisher.Failure == S.Failure, TimerPublisher.Output == S.Input {let id = UUID()// 创建一个 Subscription 并将其传递给 Subscriberlet subscription = TimerSubscription(id: id, publisher: self)subscribers[id] = (AnySubscriber(subscriber), .none)subscriber.receive(subscription: subscription)}// 开始func start() {guard timer == nil else { return }timer = Timer.scheduledTimer(withTimeInterval: 1.0, repeats: true) { [weak self] _ inself?.processValue()}}func stop() {timer?.invalidate()timer = nil// 发送完成信号给所有订阅者subscribers.values.forEach { $0.subscriber.receive(completion: .finished) }subscribers.removeAll()}// 处理订阅者的 demandfileprivate func updateDemand(for id: UUID, with newDemand: Subscribers.Demand) {if let subscriberInfo = subscribers[id] {subscribers[id] = (subscriberInfo.subscriber, subscriberInfo.demand + newDemand)}}// 取消特定订阅者fileprivate func cancelSubscription(for id: UUID) {subscribers.removeValue(forKey: id)if subscribers.isEmpty {stop()}}// 处理发送private func processValue() {counter += 1// 为每个有需求的订阅者发送值subscribers = subscribers.mapValues { subscriberInfo invar currentDemand = subscriberInfo.demand// 只在有需求时发送值if currentDemand > .none {// 直接用 receive 返回的 Demandlet newDemand = subscriberInfo.subscriber.receive(counter)currentDemand += newDemandcurrentDemand -= 1}return (subscriberInfo.subscriber, currentDemand)}}// 自定义 Subscriptionprivate class TimerSubscription: Subscription {private var id: UUIDprivate weak var publisher: TimerPublisher?init(id: UUID, publisher: TimerPublisher) {self.id = idself.publisher = publisher}// 处理 Subscriber 的请求func request(_ demand: Subscribers.Demand) {publisher?.updateDemand(for: id, with: demand)publisher?.start()}func cancel() {publisher?.cancelSubscription(for: id)}}
    }// 自定义 Subscriber
    class TimerSubscriber: Subscriber {let name: Stringinit(name: String) {self.name = name}// 指定输入、失败类型typealias Input = Inttypealias Failure = Neverfunc receive(subscription: Subscription) {print("\(name): 订阅已开始")subscription.request(.max(3)) // 限制接收3个值}func receive(_ input: Int) -> Subscribers.Demand {print("\(name): 接收到的值: \(input)")return .none // 不请求更多的值}func receive(completion: Subscribers.Completion<Never>) {print("\(name): 订阅完成")}
    }let timerPublisher = TimerPublisher()// 创建多个 subscriber
    let subscriber1 = TimerSubscriber(name: "订阅者1")
    let subscriber2 = TimerSubscriber(name: "订阅者2")// 订阅
    timerPublisher.receive(subscriber: subscriber1)
    timerPublisher.receive(subscriber: subscriber2)// 5秒后停止发布
    DispatchQueue.main.asyncAfter(deadline: .now() + 5) {timerPublisher.stop()
    }/* 输出:
    订阅者1: 订阅已开始
    订阅者2: 订阅已开始
    订阅者1: 接收到的值: 1
    订阅者2: 接收到的值: 1
    订阅者1: 接收到的值: 2
    订阅者2: 接收到的值: 2
    订阅者1: 接收到的值: 3
    订阅者2: 接收到的值: 3
    订阅者1: 订阅完成
    订阅者2: 订阅完成
    */
    

结语

自定义 Publisher 和 Subscriber 为开发者提供了更大的灵活性,能够根据具体需求扩展 Combine 框架的功能。通过掌握自定义组件的技巧,开发者可以打造出更具适应性和扩展性的应用。在下一篇文章中,将通过实际案例来展示 Combine 的贴合日常开发的简化的应用场景,帮助更好地理解和应用。

  • Swift Combine 学习(七):实践应用场景举例

相关文章:

Swift Combine 学习(六):自定义 Publisher 和 Subscriber

Swift Combine 学习&#xff08;一&#xff09;&#xff1a;Combine 初印象Swift Combine 学习&#xff08;二&#xff09;&#xff1a;发布者 PublisherSwift Combine 学习&#xff08;三&#xff09;&#xff1a;Subscription和 SubscriberSwift Combine 学习&#xff08;四&…...

Vue-router知识点汇总

import Vue from vue import Router from vue-router Vue.use(Router) import Layout from /layout export const constantRoutes [{path: /forgetpsd,name: forgetPsd,// 命名路由 &#xff0c;跳转<router-link :to"{ name: forgetPsdr, params: { userId: 123 }}&q…...

java AQS

什么是AQS AQS&#xff08;AbstractQueuedSynchronizer&#xff0c;抽象队列同步器&#xff09;是 Java 中并发控制的一种机制&#xff0c;位于 java.util.concurrent.locks 包下&#xff0c;它为构建锁、信号量等同步工具提供了一个框架。AQS 通过 队列 来管理多个线程之间的…...

L25.【LeetCode笔记】 三步问题的四种解法(含矩阵精彩解法!)

目录 1.题目 2.三种常规解法 方法1:递归做 ​编辑 方法2:改用循环做 初写的代码 提交结果 分析 修改后的代码 提交结果 for循环的其他写法 提交结果 方法3:循环数组 提交结果 3.方法4:矩阵 算法 代码实践 1.先计算矩阵n次方 2.后将矩阵n次方嵌入递推式中 提…...

sdut-C语言实验-合数分解

sdut-C语言实验-合数分解 分数 12 全屏浏览 切换布局 作者 马新娟 单位 山东理工大学 合数是指在大于1的整数中&#xff0c;除了1和本身外&#xff0c;还能被其他数整除的数。‌例如&#xff0c;4、6、8、9、10等都是合数。把一个合数分解成若干个质因数乘积的形式(即求质因…...

深入理解 pytest Fixture 方法及其应用

在 Python 自动化测试领域&#xff0c;pytest 是当之无愧的王者。提到 pytest&#xff0c;不得不说它的一大核心功能——Fixture。Fixture 的强大&#xff0c;让复杂的测试流程变得井井有条&#xff0c;让测试代码更加灵活和可复用。 那么&#xff0c;pytest 的 Fixture 究竟是…...

在Linux上获取MS(如Media Server)中的RTP流并录制为双轨PCM格式的WAV文件

在Linux上获取MS(如Media Server)中的RTP流并录制为双轨PCM格式的WAV文件 一、RTP流与WAV文件格式二、实现步骤三、伪代码示例四、C语言示例代码五、关键点说明六、总结在Linux操作系统上,从媒体服务器(如Media Server,简称MS)获取RTP(Real-time Transport Protocol)流…...

Midjourney技术浅析(八):交互与反馈

Midjourney 的用户交互与反馈通过用户输入&#xff08;User Input&#xff09;和用户反馈&#xff08;User Feedback&#xff09;机制&#xff0c;不断优化和改进图像生成的质量和用户满意度。 一、用户交互与反馈模块概述 用户交互与反馈模块的主要功能包括&#xff1a; 1.…...

【Spring MVC 核心机制】核心组件和工作流程解析

在 Web 应用开发中&#xff0c;处理用户请求的逻辑常常会涉及到路径匹配、请求分发、视图渲染等多个环节。Spring MVC 作为一款强大的 Web 框架&#xff0c;将这些复杂的操作高度抽象化&#xff0c;通过组件协作简化了开发者的工作。 无论是处理表单请求、生成动态页面&#x…...

回归问题的等量分层

目录 一、说明 二、什么是分层抽样&#xff1f; 三、那么回归又如何呢&#xff1f; 四、回归分层&#xff08;Stratification on Regression&#xff09; 一、说明 在同一个数据集中&#xff0c;我们可以看成是一个抽样体。然而&#xff0c;我们如果将这个抽样体分成两份&#…...

Unity-Mirror网络框架-从入门到精通之Basic示例

文章目录 前言Basic示例场景元素预制体元素代码逻辑BasicNetManagerPlayer逻辑SyncVars属性Server逻辑Client逻辑 PlayerUI逻辑 最后 前言 在现代游戏开发中&#xff0c;网络功能日益成为提升游戏体验的关键组成部分。Mirror是一个用于Unity的开源网络框架&#xff0c;专为多人…...

CSS 图片廊:网页设计的艺术与技巧

CSS 图片廊&#xff1a;网页设计的艺术与技巧 引言 在网页设计中&#xff0c;图片廊是一个重要的组成部分&#xff0c;它能够以视觉吸引的方式展示图片集合&#xff0c;增强用户的浏览体验。CSS&#xff08;层叠样式表&#xff09;作为网页设计的主要语言之一&#xff0c;提供…...

AI 发展的第一驱动力:人才引领变革

在科技蓬勃发展的当下&#xff0c;AI 成为了时代的焦点&#xff0c;然而其发展并非一帆风顺&#xff0c;究竟什么才是推动 AI 持续前行的关键力量呢&#xff1f; 目录 AI 发展现状剖析 期望与现实的落差 落地困境根源 人才&#xff1a;AI 发展的核心动力​编辑 技术突破的…...

[创业之路-229]:《华为闭环战略管理》-5-平衡记分卡与战略地图

目录 一、平衡记分卡 1. 财务角度&#xff1a; 2. 客户角度&#xff1a; 3. 内部运营角度&#xff1a; 4. 学习与成长角度&#xff1a; 二、BSC战略地图 1、核心内容 2、绘制目的 3、绘制方法 4、注意事项 一、平衡记分卡 平衡记分卡&#xff08;Balanced Scorecard&…...

用uniapp写一个播放视频首页页面代码

效果如下图所示 首页有导航栏&#xff0c;搜索框&#xff0c;和视频列表&#xff0c; 导航栏如下图 搜索框如下图 视频列表如下图 文件目录 视频首页页面代码如下 <template> <view class"video-home"> <!-- 搜索栏 --> <view class…...

【视觉SLAM:八、后端Ⅰ】

视觉SLAM的后端主要解决状态估计问题&#xff0c;它是优化相机轨迹和地图点的过程&#xff0c;从数学上看属于非线性优化问题。后端的目标是结合传感器数据&#xff0c;通过最优估计获取系统的状态&#xff08;包括相机位姿和场景结构&#xff09;&#xff0c;在状态估计过程中…...

PaddleOCROCR关键信息抽取训练过程

步骤1&#xff1a;python版本3.8.20 步骤2&#xff1a;下载代码&#xff0c;安装依赖 git clone https://gitee.com/PaddlePaddle/PaddleOCR.git pip uninstall opencv-python -y # 安装PaddleOCR的依赖 ! pip install -r requirements.txt # 安装关键信息抽取任务的依赖 !…...

用Python操作字节流中的Excel文档

Python能够轻松地从字节流中加载文件&#xff0c;在不依赖于外部存储的情况下直接对其进行读取、修改等复杂操作&#xff0c;并最终将更改后的文档保存回字节串中。这种能力不仅极大地提高了数据处理的灵活性&#xff0c;还确保了数据的安全性和完整性&#xff0c;尤其是在网络…...

python 桶排序(Bucket Sort)

桶排序&#xff08;Bucket Sort&#xff09; 桶排序是一种分布式排序算法&#xff0c;适用于对均匀分布的数据进行排序。它的基本思想是&#xff1a;将数据分到有限数量的桶中&#xff0c;每个桶分别排序&#xff0c;最后将所有桶中的数据合并。 桶排序的步骤&#xff1a; 划…...

Elasticsearch:探索 Elastic 向量数据库的深度应用

Elasticsearch&#xff1a;探索 Elastic 向量数据库的深度应用 一、Elasticsearch 向量数据库简介 1. Elasticsearch 向量数据库的概念 Elasticsearch 本身是一个基于 Lucene 的搜索引擎&#xff0c;提供了全文搜索和分析的功能。随着技术的发展&#xff0c;Elasticsearch 也…...

解锁数据库简洁之道:FastAPI与SQLModel实战指南

在构建现代Web应用程序时&#xff0c;与数据库的交互无疑是核心环节。虽然传统的数据库操作方式&#xff08;如直接编写SQL语句与psycopg2交互&#xff09;赋予了我们精细的控制权&#xff0c;但在面对日益复杂的业务逻辑和快速迭代的需求时&#xff0c;这种方式的开发效率和可…...

ESP32 I2S音频总线学习笔记(四): INMP441采集音频并实时播放

简介 前面两期文章我们介绍了I2S的读取和写入&#xff0c;一个是通过INMP441麦克风模块采集音频&#xff0c;一个是通过PCM5102A模块播放音频&#xff0c;那如果我们将两者结合起来&#xff0c;将麦克风采集到的音频通过PCM5102A播放&#xff0c;是不是就可以做一个扩音器了呢…...

论文浅尝 | 基于判别指令微调生成式大语言模型的知识图谱补全方法(ISWC2024)

笔记整理&#xff1a;刘治强&#xff0c;浙江大学硕士生&#xff0c;研究方向为知识图谱表示学习&#xff0c;大语言模型 论文链接&#xff1a;http://arxiv.org/abs/2407.16127 发表会议&#xff1a;ISWC 2024 1. 动机 传统的知识图谱补全&#xff08;KGC&#xff09;模型通过…...

C++八股 —— 单例模式

文章目录 1. 基本概念2. 设计要点3. 实现方式4. 详解懒汉模式 1. 基本概念 线程安全&#xff08;Thread Safety&#xff09; 线程安全是指在多线程环境下&#xff0c;某个函数、类或代码片段能够被多个线程同时调用时&#xff0c;仍能保证数据的一致性和逻辑的正确性&#xf…...

优选算法第十二讲:队列 + 宽搜 优先级队列

优选算法第十二讲&#xff1a;队列 宽搜 && 优先级队列 1.N叉树的层序遍历2.二叉树的锯齿型层序遍历3.二叉树最大宽度4.在每个树行中找最大值5.优先级队列 -- 最后一块石头的重量6.数据流中的第K大元素7.前K个高频单词8.数据流的中位数 1.N叉树的层序遍历 2.二叉树的锯…...

Hive 存储格式深度解析:从 TextFile 到 ORC,如何选对数据存储方案?

在大数据处理领域&#xff0c;Hive 作为 Hadoop 生态中重要的数据仓库工具&#xff0c;其存储格式的选择直接影响数据存储成本、查询效率和计算资源消耗。面对 TextFile、SequenceFile、Parquet、RCFile、ORC 等多种存储格式&#xff0c;很多开发者常常陷入选择困境。本文将从底…...

【VLNs篇】07:NavRL—在动态环境中学习安全飞行

项目内容论文标题NavRL: 在动态环境中学习安全飞行 (NavRL: Learning Safe Flight in Dynamic Environments)核心问题解决无人机在包含静态和动态障碍物的复杂环境中进行安全、高效自主导航的挑战&#xff0c;克服传统方法和现有强化学习方法的局限性。核心算法基于近端策略优化…...

【 java 虚拟机知识 第一篇 】

目录 1.内存模型 1.1.JVM内存模型的介绍 1.2.堆和栈的区别 1.3.栈的存储细节 1.4.堆的部分 1.5.程序计数器的作用 1.6.方法区的内容 1.7.字符串池 1.8.引用类型 1.9.内存泄漏与内存溢出 1.10.会出现内存溢出的结构 1.内存模型 1.1.JVM内存模型的介绍 内存模型主要分…...

认识CMake并使用CMake构建自己的第一个项目

1.CMake的作用和优势 跨平台支持&#xff1a;CMake支持多种操作系统和编译器&#xff0c;使用同一份构建配置可以在不同的环境中使用 简化配置&#xff1a;通过CMakeLists.txt文件&#xff0c;用户可以定义项目结构、依赖项、编译选项等&#xff0c;无需手动编写复杂的构建脚本…...

微服务通信安全:深入解析mTLS的原理与实践

&#x1f525;「炎码工坊」技术弹药已装填&#xff01; 点击关注 → 解锁工业级干货【工具实测|项目避坑|源码燃烧指南】 一、引言&#xff1a;微服务时代的通信安全挑战 随着云原生和微服务架构的普及&#xff0c;服务间的通信安全成为系统设计的核心议题。传统的单体架构中&…...