c# MoreLinq 之 Aggregate

Posted 2021-04-03 DotNet技术说

前言

本系列是对MoreLinq库的学习与总结，分析各个Api的实现方式及用法，也为能写出更高效的Linq打下基础。

Aggregate 对序列应用累加器

• 原生方法

//1. 
public static TSource Aggregate<TSource>(this IEnumerable<TSource> source, Func<TSource, TSource, TSource> func);
//2. 
public static TAccumulate Aggregate<TSource, TAccumulate>(this IEnumerable<TSource> source, TAccumulate seed, Func<TAccumulate, TSource, TAccumulate> func);
//3. 
public static TResult Aggregate<TSource, TAccumulate, TResult>(this IEnumerable<TSource> source, TAccumulate seed, Func<TAccumulate, TSource, TAccumulate> func, Func<TAccumulate, TResult> resultSelector);

对原生方法的应用

var nums = Enumerable.Range(0, 2);
string.Join(",", nums).Dump();//0,1
nums.Aggregate((x, y) => x + y).Dump();//重载1,=> 1
nums.Aggregate(1, (x, y) => x + y).Dump();//重载2, =>2
nums.Aggregate(1, (x, y) => x + y, r => "=" + r).Dump("accumulate:3");//重载3,=>=2

• 重载1是直接从把第一个值作为初始值，源码：

using System.Collections.Generic;
public static TSource Aggregate<TSource>(this IEnumerable<TSource> source, Func<TSource, TSource, TSource> func)
{
    //省略null值判断
    using IEnumerator<TSource> enumerator = source.GetEnumerator();
    if (!enumerator.MoveNext())
    {
        ThrowHelper.ThrowNoElementsException();
    }
    TSource val = enumerator.Current;
    while (enumerator.MoveNext())
    {
        val = func(val, enumerator.Current);
    }
    return val;
}

• 重载2的初始值则是用的传入的值

using System.Collections.Generic;
public static TAccumulate Aggregate<TSource, TAccumulate>(this IEnumerable<TSource> source, TAccumulate seed, Func<TAccumulate, TSource, TAccumulate> func)
{
    //省略null值判断
    TAccumulate val = seed;
    foreach (TSource item in source)
    {
        val = func(val, item);
    }
    return val;
}

• 重载3使用在重载2的基础上对结果执行传入的方法

using System.Collections.Generic;
public static TResult Aggregate<TSource, TAccumulate, TResult>(this IEnumerable<TSource> source, TAccumulate seed, Func<TAccumulate, TSource, TAccumulate> func, Func<TAccumulate, TResult> resultSelector)
{
    //省略null值判断
    TAccumulate val = seed;
    foreach (TSource item in source)
    {
        val = func(val, item);
    }
    return resultSelector(val);
}

• MoreLinq中定义：
  同步数据流方法

namespace MoreLinq
{
    using System;
    using System.Collections.Generic;

    partial class MoreEnumerable
    {
        /// <summary>
        /// Applies two accumulators sequentially in a single pass over a
        /// sequence.
        /// </summary>
        /// <typeparam name="T">The type of elements in <paramref name="source"/>.</typeparam>
        /// <typeparam name="TAccumulate1">The type of first accumulator value.</typeparam>
        /// <typeparam name="TAccumulate2">The type of second accumulator value.</typeparam>
        /// <typeparam name="TResult">The type of the accumulated result.</typeparam>
        /// <param name="source">The source sequence</param>
        /// <param name="seed1">The seed value for the first accumulator.</param>
        /// <param name="accumulator1">The first accumulator.</param>
        /// <param name="seed2">The seed value for the second accumulator.</param>
        /// <param name="accumulator2">The second accumulator.</param>
        /// <param name="resultSelector">
        /// A function that projects a single result given the result of each
        /// accumulator.</param>
        /// <returns>The value returned by <paramref name="resultSelector"/>.</returns>
        /// <remarks>
        /// This operator executes immediately.
        /// </remarks>

        public static TResult Aggregate<T, TAccumulate1, TAccumulate2, TResult>(
            this IEnumerable<T> source,
            TAccumulate1 seed1, Func<TAccumulate1, T, TAccumulate1> accumulator1,
            TAccumulate2 seed2, Func<TAccumulate2, T, TAccumulate2> accumulator2,
            Func<TAccumulate1, TAccumulate2, TResult> resultSelector)
        {
            if (source == null) throw new ArgumentNullException(nameof(source));
            if (accumulator1 == null) throw new ArgumentNullException(nameof(accumulator1));
            if (accumulator2 == null) throw new ArgumentNullException(nameof(accumulator2));
            if (resultSelector == null) throw new ArgumentNullException(nameof(resultSelector));

            var a1 = seed1;
            var a2 = seed2;

            foreach (var item in source)
            {
                a1 = accumulator1(a1, item);
                a2 = accumulator2(a2, item);
            }

            return resultSelector(a1, a2);
        }
    }
}

分析：
该方法是对两个初始值和两个累加器的扩展，且sead1只应用accumulator1，同理sead2只应用accumulator2，然后对结果应用resultSelector(a1,a2)方法。

异步数据流方法

namespace MoreLinq.Experimental
{
    using System;
    using System.Collections.Generic;
    using Reactive;

    partial class ExperimentalEnumerable
    {
        /// <summary>
        /// Applies two accumulator queries sequentially in a single
        /// pass over a sequence.
        /// </summary>
        /// <typeparam name="T">The type of elements in <paramref name="source"/>.</typeparam>
        /// <typeparam name="TResult1">The type of the result of the first accumulator.</typeparam>
        /// <typeparam name="TResult2">The type of the result of the second accumulator.</typeparam>
        /// <typeparam name="TResult">The type of the accumulated result.</typeparam>
        /// <param name="source">The source sequence</param>
        /// <param name="accumulator1">The first accumulator.</param>
        /// <param name="accumulator2">The second accumulator.</param>
        /// <param name="resultSelector">
        /// A function that projects a single result given the result of each
        /// accumulator.</param>
        /// <returns>The value returned by <paramref name="resultSelector"/>.</returns>
        /// <exception cref="InvalidOperationException">
        /// An <see cref="IObservable{T}"/> returned by an accumulator function
        /// produced zero or more than a single aggregate result.
        /// </exception>
        /// <remarks>
        /// <para>This operator executes immediately.</para>
        /// <para>
        /// Each accumulator argument is a function that receives an
        /// <see cref="IObservable{T}"/>, which when subscribed to, produces the
        /// items in the <paramref name="source"/> sequence and in original
        /// order; the function must then return an <see cref="IObservable{T}"/>
        /// that produces a single aggregate on completion (when
        /// <see cref="IObserver{T}.OnCompleted"/> is called. An error is raised
        /// at run-time if the <see cref="IObserver{T}"/> returned by an
        /// accumulator function produces no result or produces more than a
        /// single result.
        /// </para>
        /// </remarks>

        public static TResult Aggregate<T, TResult1, TResult2, TResult>(
            this IEnumerable<T> source,
            Func<IObservable<T>, IObservable<TResult1>> accumulator1,
            Func<IObservable<T>, IObservable<TResult2>> accumulator2,
            Func<TResult1, TResult2, TResult> resultSelector)
        {
            if (source == null) throw new ArgumentNullException(nameof(source));
            if (accumulator1 == null) throw new ArgumentNullException(nameof(accumulator1));
            if (accumulator2 == null) throw new ArgumentNullException(nameof(accumulator2));
            if (resultSelector == null) throw new ArgumentNullException(nameof(resultSelector));

            var r1 = new (bool, TResult1)[1];
            var r2 = new (bool, TResult2)[1];

            var subject = new Subject<T>();

            using (SubscribeSingle(accumulator1, subject, r1, nameof(accumulator1)))
            using (SubscribeSingle(accumulator2, subject, r2, nameof(accumulator2)))
            {
                foreach (var item in source)
                    subject.OnNext(item);

                subject.OnCompleted();
            }

            return resultSelector
            (
                GetAggregateResult(r1[0], nameof(accumulator1)),
                GetAggregateResult(r2[0], nameof(accumulator2))
            );
        }
    }
}

分析：
分别订阅流数据，将执行结果应用resultSelector方法。

测试：

void AppendTwoAccumulatorTest()
{
    var a = new[] { 1, 2 };
    var res = a.Aggregate(1, (x, y) => x + y, 2, (x, y) => x * y, (x, y) => x * y);
    //(1+1+2)*(2+1+2)
    res.Dump();//16
}

分析：
应用了两个累加器。