RxSwift - 过滤操作符系列之 1

本文最后更新于 2021年4月4日 晚上

RxSwift 中的若干个过滤功能的操作符介绍.

包括如下操作符(有些操作符可能是操作符族):

  1. ignoreElements: 应用这个操作符会生成一个 Completable, 故只有错误和完成, 所有其他元素均被忽略.

  2. elementAt: 在一个可观察序列中获取指定下标的元素.

  3. skip: 在序列中跳过若干个元素后开始取元素.

  4. skipWhile: 一直跳过元素, 直到遇到满足条件的元素才不再执行跳过.

  5. skipUntil: 在另外一个序列没有开始发射时就一直忽略元素, 直到另外一个序列发射后停止忽略.

  6. take: 在序列中只取指定个数的元素.

  7. takeWhile: 一直取元素, 直到遇到不满足指定条件的元素就终止取元素, 需要注意的是, 如果序列的第一个元素就不满足条件, 则所有的元素都不会取.

  8. takeUntil: 在另外一个序列没有发射时一直取元素, 直到另外一个序列发射就停止取.

  9. distinctUntilChanged: 在连续的元素中, 如果有若干个相邻元素相等, 则只取相邻元素中的第一个, 直到遇到不同的再取.

完整示例代码如下所示:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
/*
 含义: 应用这个操作符会生成一个 Completable, 故只有错误和完成, 所有其他元素均被忽略
 */
exampleOf(msg: "过滤操作符: ignoreElements", action: {
    let strikes = PublishSubject<String>()
    strikes.ignoreElements().subscribe(onCompleted: {
        print("完成")
    }, onError: { error in
        print(error)
    }).disposed(by: $0)
    strikes.onNext("X")
    strikes.onNext("X")
    strikes.onNext("X")
    strikes.onNext("X")
    strikes.onCompleted()
})

/*
 含义: 在一个可观察序列中获取指定下标的元素
 */
exampleOf(msg: "过滤操作符: elementAt", action: {
    let strikes = PublishSubject<String>()
    strikes.elementAt(2).subscribe(onNext: {
        print($0)
    }, onError: {
        print($0)
    }, onCompleted: {
        print("完成")
    }, onDisposed: {
        print("释放")
    }).disposed(by: $0)

    strikes.onNext("X") // 0
    strikes.onNext("Y") // 1
    strikes.onNext("Z") // 2
    strikes.onNext("1") // 3
    strikes.onCompleted()
})

/*
 含义: 在序列中跳过若干个元素后开始取元素直到完成或错误.
 */
exampleOf(msg: "过滤操作符: skip", action: {
    let flow = PublishSubject<String>()
    flow.skip(3).subscribe(onNext: {
        print($0)
    }, onError: {
        print($0)
    }, onCompleted: {
        print("完成")
    }, onDisposed: {
        print("被释放")
    }).disposed(by: $0)
    flow.onNext("X")
    flow.onNext("Y")
    flow.onNext("Z")
    flow.onNext("1")
    flow.onNext("2")
    flow.onCompleted()
})

/*
 一直跳过元素, 直到遇到满足条件的元素就不再执行跳过操作
 */
exampleOf(msg: "过滤操作符: skipWhile", action: { bag in
    let flow = PublishSubject<String>()
    flow.skipWhile({ $0.count > 2 }).subscribe(onNext: {
        print($0)
    }, onError: {
        print($0)
    }, onCompleted: {
        print("完成")
    }, onDisposed: {
        print("被释放")
    }).disposed(by: bag)

    flow.onNext("1234")
    flow.onNext("256")
    flow.onNext("377")
    flow.onNext("4")
    flow.onNext("5")
    flow.onNext("6")
    flow.onNext("7")
    flow.onCompleted()
})

/*
 在另外一个序列没有开始发射时就一直忽略元素, 直到另外一个序列发射后停止忽略.
 */
exampleOf(msg: "过滤操作符: skipUntil", action: { bag in
    let flow = PublishSubject<String>()
    let trigger = PublishSubject<Void>()

    flow.skipUntil(trigger).subscribe(onNext: {
        print($0)
    }, onError: {
        print($0)
    }, onCompleted: {
        print("完成")
    }, onDisposed: {
        print("被释放")
    }).disposed(by: bag)

    flow.onNext("X")
    flow.onNext("X")
    flow.onNext("X")
    trigger.onNext(())
    flow.onNext("Y")
    flow.onNext("Z")
    flow.onCompleted()
})

/*
 取指定个数的元素
 */
exampleOf(msg: "过滤操作符: take", action: { bag in
    Observable.from([1, 2, 3, 4, 5, 6]).take(2).subscribe(onNext: {
        print($0)
    }, onError: {
        print($0)
    }, onCompleted: {
        print("完成")
    }, onDisposed: {
        print("被释放")
    }).disposed(by: bag)
})

/*
 一直取元素, 直到遇到不满足指定条件的元素就终止取元素.
 需要注意的是, 如果序列的第一个元素就不满足条件, 则之后所有的元素都不会取.
 */
exampleOf(msg: "过滤操作符: takeWhile", action: { bag in
    Observable.of(1, 2, 3, 4, 5, 6, 7)
        .takeWhile({ $0 > 3 })
        .subscribe(onNext: {
            print($0)
        }, onError: {
            print($0)
        }, onCompleted: {
            print("完成")
        }, onDisposed: {
            print("被释放")
        }).disposed(by: bag)

    Observable.of(1, 2, 3, 4, 5, 6, 7)
        .enumerated()
        .takeWhile({ index, elem in index < 3 && elem < 2 })
        .subscribe(onNext: {
            print($0)
        }, onError: {
            print($0)
        }, onCompleted: {
            print("完成")
        }, onDisposed: {
            print("被释放")
        }).disposed(by: bag)
})

/*
 在另外一个序列没有发射时一直取元素, 直到另外一个序列发射就停止取
 */
exampleOf(msg: "过滤操作符 takeUntil", action: { bag in
    let flow = PublishSubject<String>()
    let trigger = PublishSubject<Void>()
    flow.takeUntil(trigger).subscribe(onNext: {
        print($0)
    }, onError: {
        print($0)
    }, onCompleted: {
        print("完成")
    }, onDisposed: {
        print("被释放")
    }).disposed(by: bag)
    flow.onNext("X")
    flow.onNext("X")
    flow.onNext("X")
    flow.onNext("X")
    flow.onNext("X")
    trigger.onNext(())
    flow.onNext("Y")
    flow.onNext("Z")
    flow.onNext("1")
    flow.onCompleted()
})

/*
 在连续的元素中, 如果有若干个相邻元素相等, 则只取相邻元素中的第一个, 直到遇到不同的再取.
 */
exampleOf(msg: "过滤操作符: distinctUntilChanged", action: { bag in
    let flow = PublishSubject<String>()
    flow.distinctUntilChanged().subscribe(onNext: {
        print($0)
    }, onError: {
        print($0)
    }, onCompleted: {
        print("完成")
    }, onDisposed: {
        print("被释放")
    }).disposed(by: bag)

    flow.onNext("X")
    flow.onNext("X")
    flow.onNext("Y")
    flow.onNext("X")
    flow.onNext("Z")
    flow.onNext("Z")
    flow.onNext("X")
    flow.onCompleted()
})
iOS 开发
RxSwift - 过滤操作符系列之 1
https://blog.rayy.top/2018/11/22/2019-19-rx-filter/
作者
貘鸣
发布于
2018年11月22日
许可协议