ReactiveX における集計操作符の実装

数学的・集計操作符

本稿では、Observable（可観測データストリーム）に対する集計操作符について解説します。

Average（平均値）

Average操作符は、ソースデータストリームの全データの平均値を計算し、その平均値を出力します。

// C# (Rx.NET)
var source = new Subject<int>();
source.Average().Subscribe(avg =>
{
    Console.WriteLine("Average: {0}", avg);
},
() => Console.WriteLine("Average Completed"));

Console.WriteLine("OnNext(1-3)");
source.OnNext(1);
source.OnNext(2);
source.OnNext(3);
Console.WriteLine("OnCompleted()");
source.OnCompleted();

/*
出力結果:
OnNext(1-3)
OnCompleted()
Average: 2
Average Completed
*/

Concat（連結）

Concat操作符は、複数のデータストリームを順に連結します。最初のデータストリームの出力完了後、次のデータストリームの出力が開始されます。

// C# (Rx.NET) - 基本的な連結
var stream1 = Observable.Range(0, 3);
var stream2 = Observable.Range(5, 5);
stream1.Concat(stream2)
    .Subscribe(Console.WriteLine);

/*
出力:
0
1
2
5
6
7
8
9
*/

// C# (Rx.NET) - 遅延評価による連結
private static IEnumerable<IObservable<long>> GenerateSequences()
{
    Console.WriteLine("GenerateSequences() called");
    Console.WriteLine("Yield 1st sequence");
    yield return Observable.Create<long>(observer =>
    {
        Console.WriteLine("1st subscribed to");
        return Observable.Timer(TimeSpan.FromMilliseconds(500))
            .Select(i => 1L)
            .Subscribe(observer);
    });
    Console.WriteLine("Yield 2nd sequence");
    yield return Observable.Create<long>(observer =>
    {
        Console.WriteLine("2nd subscribed to");
        return Observable.Timer(TimeSpan.FromMilliseconds(300))
            .Select(i => 2L)
            .Subscribe(observer);
    });
    Thread.Sleep(1000);
    Console.WriteLine("Yield 3rd sequence");
    yield return Observable.Create<long>(observer =>
    {
        Console.WriteLine("3rd subscribed to");
        return Observable.Timer(TimeSpan.FromMilliseconds(100))
            .Select(i => 3L)
            .Subscribe(observer);
    });
    Console.WriteLine("GenerateSequences() complete");
}

GenerateSequences().Concat().Dump("Concat");

/*
出力:
GenerateSequences() called
Yield 1st sequence
1st subscribed to
Concat-->1
Yield 2nd sequence
2nd subscribed to
Concat-->2
Yield 3rd sequence
3rd subscribed to
Concat-->3
GenerateSequences() complete
Concat completed
*/

// Kotlin (RxJava)
val seq1 = Observable.range(0, 3)
val seq2 = Observable.range(10, 3)
Observable.concat(seq1, seq2)
    .subscribe { println("onNext: $it") }

/*
出力:
onNext: 0
onNext: 1
onNext: 2
onNext: 10
onNext: 11
onNext: 12
onComplete
*/

// Kotlin (RxJava) - groupByとの組み合わせ
val words = Observable.just(
    "First",
    "Second",
    "Third",
    "Fourth",
    "Fifth",
    "Sixth"
)
Observable.concat(words.groupBy { v -> v[0] })
    .subscribe { println("onNext: $it") }

/*
出力:
onNext: First
onNext: Fourth
onNext: Fifth
onNext: Second
onNext: Sixth
onNext: Third
onComplete
*/

// Swift (RxSwift)
let disposeBag = DisposeBag()
let subject1 = BehaviorSubject(value: "🍎")
let subject2 = BehaviorSubject(value: "🐶")
let subjectsSubject = BehaviorSubject(value: subject1)

subjectsSubject.asObservable()
    .concat()
    .subscribe { print($0) }
    .disposed(by: disposeBag)

subject1.onNext("🍐")
subject1.onNext("🍊")
subjectsSubject.onNext(subject2)
subject2.onNext("I would be ignored")
subject2.onNext("🐱")
subject1.onCompleted()
subject2.onNext("🐭")

/*
出力:
next(🍎)
next(🍐)
next(🍊)
next(🐱)
next(🐭)
*/

Count（要素数）

Count操作符は、ソースデータストリームの要素数をカウントし、その値を出力します。

// C# (Rx.NET)
var source = new Subject<int>();
source.Count()
    .Subscribe(
        i => Console.WriteLine("Count OnNext({0})", i),
        () => Console.WriteLine("Count OnCompleted()"));

Console.WriteLine("OnNext(1)");
source.OnNext(1);
Console.WriteLine("OnNext(10)");
source.OnNext(10);
Console.WriteLine("OnNext(100)");
source.OnNext(100);
Console.WriteLine("OnCompleted()");
source.OnCompleted();

/*
出力:
OnNext(1)
OnNext(10)
OnNext(100)
OnCompleted()
Count OnNext(3)
Count OnCompleted()
*/

// Kotlin (RxJava)
val values = Observable.range(0, 3)
values.subscribe { println("Values: onNext: $it") }
values.count()
    .subscribe { println("Count: onSuccess: $it") }

/*
出力:
Values: onNext: 0
Values: onNext: 1
Values: onNext: 2
Values: onComplete
Count: onSuccess: 3
*/

Max（最大値）

Max操作符は、ソースデータストリームの全データから最大値を取得し、その値を出力します。MaxByを使用すると、指定したキー関数に基づいた最大値を取得できます。

// C# (Rx.NET) - 基本的なMax
var source = new Subject<int>();
source.Max().Subscribe(max =>
{
    Console.WriteLine("Max: {0}", max);
},
() => Console.WriteLine("Max Completed"));

Console.WriteLine("OnNext(1-3)");
source.OnNext(1);
source.OnNext(2);
source.OnNext(3);
Console.WriteLine("OnCompleted()");
source.OnCompleted();

/*
出力:
OnNext(1-3)
OnCompleted()
Max: 3
Max Completed
*/

// C# (Rx.NET) - MaxBy
var source = new Subject<Tuple<int, int>>();
source.MaxBy(t => t.Item1).Subscribe(maxList =>
{
    foreach (var item in maxList)
    {
        Console.WriteLine("MaxBy: {0}", item);
    }
},
() => Console.WriteLine("MaxBy Completed"));

Console.WriteLine("OnNext");
source.OnNext(Tuple.Create(1, 1));
source.OnNext(Tuple.Create(1, 2));
source.OnNext(Tuple.Create(3, 1));
Console.WriteLine("OnCompleted()");
source.OnCompleted();

/*
出力:
OnNext
OnCompleted()
MaxBy (3, 1)
MaxBy Completed
*/

Min（最小値）

Min操作符は、ソースデータストリームの全データから最小値を取得し、その値を出力します。MinByを使用すると、指定したキー関数に基づいた最小値を取得できます。

// C# (Rx.NET) - 基本的なMin
var source = new Subject<int>();
source.Min().Subscribe(min =>
{
    Console.WriteLine("Min: {0}", min);
},
() => Console.WriteLine("Min Completed"));

Console.WriteLine("OnNext(1-3)");
source.OnNext(1);
source.OnNext(2);
source.OnNext(3);
Console.WriteLine("OnCompleted()");
source.OnCompleted();

/*
出力:
OnNext(1-3)
OnCompleted()
Min: 1
Min Completed
*/

// C# (Rx.NET) - MinBy
var source = new Subject<Tuple<int, int>>();
source.MinBy(t => t.Item1).Subscribe(minList =>
{
    foreach (var item in minList)
    {
        Console.WriteLine("MinBy: {0}", item);
    }
},
() => Console.WriteLine("MinBy Completed"));

Console.WriteLine("OnNext");
source.OnNext(Tuple.Create(1, 1));
source.OnNext(Tuple.Create(1, 2));
source.OnNext(Tuple.Create(3, 1));
Console.WriteLine("OnCompleted()");
source.OnCompleted();

/*
出力:
OnNext
OnCompleted()
MinBy (1, 1)
MinBy (1, 2)
MinBy Completed
*/

Reduce / Aggregate / Collect

Reduce操作符（Rx.NETではAggregate）は、二項演算子とオプションの初期値を受け取ります。初期値が存在する場合、最初の要素と初期値に対して演算子を適用し、その結果を次の要素に対して適用していくことで最終結果を導出します。初期値がない場合は、最初の2要素から開始されます。Collect操作符は似ていますが、初期値コールバックを使用して可変初期値を生成します。

// C# (Rx.NET) - 初期値なし
var source = new Subject<int>();
source.Aggregate((acc, val) =>
{
    Console.WriteLine("log({0}, {1})", acc, val);
    return acc + val;
})
.Subscribe(
    result => Console.WriteLine("Aggregate OnNext({0})", result),
    () => Console.WriteLine("Aggregate OnCompleted()"));

Console.WriteLine("OnNext(1)");
source.OnNext(1);
Console.WriteLine("OnNext(10)");
source.OnNext(10);
Console.WriteLine("OnNext(100)");
source.OnNext(100);
Console.WriteLine("OnCompleted()");
source.OnCompleted();

/*
出力:
OnNext(1)
OnNext(10)
log(1, 10)
OnNext(100)
log(11, 100)
OnCompleted()
Aggregate OnNext(111)
Aggregate OnCompleted()
*/

// C# (Rx.NET) - 初期値あり
var source = new Subject<int>();
source.Aggregate(5, (acc, val) =>
{
    Console.WriteLine("log({0}, {1})", acc, val);
    return acc + val;
})
.Subscribe(
    result => Console.WriteLine("Aggregate OnNext({0})", result),
    () => Console.WriteLine("Aggregate OnCompleted()"));

Console.WriteLine("OnNext(1)");
source.OnNext(1);
Console.WriteLine("OnNext(10)");
source.OnNext(10);
Console.WriteLine("OnNext(100)");
source.OnNext(100);
Console.WriteLine("OnCompleted()");
source.OnCompleted();

/*
出力:
OnNext(1)
log(5, 1)
OnNext(10)
log(6, 10)
OnNext(100)
log(16, 100)
OnCompleted()
Aggregate OnNext(116)
Aggregate OnCompleted()
*/

// Kotlin (RxJava) - 合計と最小値
val values = Observable.range(0, 5)
values.reduce { i1, i2 -> i1 + i2 }
    .subscribe { println("Sum: onSuccess: $it") }
values.reduce { i1, i2 -> if (i1 > i2) i2 else i1 }
    .subscribe { println("Min: onSuccess: $it") }

/*
出力:
Sum: onSuccess: 10
Min: onSuccess: 0
*/

// Kotlin (RxJava) - Countとしての利用
val text = Observable.just("Rx", "is", "easy")
text.reduce(0) { acc, next -> acc + 1 }
    .subscribe { println("Count: onSuccess: $it") }

/*
出力:
Count: onSuccess: 3
*/

// Kotlin (RxJava) - Collect
val numbers = Observable.range(10, 5)
numbers.collect(
    { ArrayList<Int>() },
    { acc, value -> acc.add(value) })
    .subscribe { println("Result: onSuccess: $it") }

/*
出力:
Result: onSuccess: [10, 11, 12, 13, 14]
*/

// Swift (RxSwift)
let disposeBag = DisposeBag()
Observable.of(10, 100, 1000)
    .reduce(1, accumulator: +)
    .subscribe(onNext: { print($0) })
    .disposed(by: disposeBag)

/*
出力:
1111
*/

Sum（合計）

Sum操作符は、ソースデータストリームの全データの合計値を計算し、その値を出力します。

// C# (Rx.NET)
var source = new Subject<int>();
source.Sum()
    .Subscribe(
        result => Console.WriteLine("Sum OnNext({0})", result),
        () => Console.WriteLine("Sum OnCompleted()"));

Console.WriteLine("OnNext(1)");
source.OnNext(1);
Console.WriteLine("OnNext(10)");
source.OnNext(10);
Console.WriteLine("OnNext(100)");
source.OnNext(100);
Console.WriteLine("OnCompleted()");
source.OnCompleted();

/*
出力:
OnNext(1)
OnNext(10)
OnNext(100)
OnCompleted()
Sum OnNext(111)
Sum OnCompleted()
*/

タグ: ReactiveX rxjava rxnet RxSwift observable

7月14日 21:51 投稿

ホットタグ

©2026 異端開発室