• 7 Stream
  • 7.20 Two procedures on streams
  • 7.21 Defining streams implicitly
  • 7.22 Recursively defined data objects
  • 7.23 Integrator – Viewed as a signal-processing system
    • 7.23.1 Generative defining function map streams
    • 7.23.2 Defining function map streams implicitly
  • 7.24 Differential Equation
  • 7.25 Normal-Order vs. Applicative-Order

前面讲生成流，都用的cons_stream head tail，我们把这种方式叫做“生成式”的构建流的方式，每次构建一个元素，下面一个是构建流的下一个元素。接下来，我们要把流作为一个整体来看，基于流定义流，把流作为一个整体，在上面对它进行操作，书上叫做“隐式定义流”。

有两种定义流的方式：一种是用构造子，生成式地定义流，一步一步的；另外一种是把流当作整体，基于这个流进行操作，来定义流。

在看例子之前，先写两个辅助方法。

• add_stream：给两个流，把两个流中的元素逐个取出来相加，构成新的流。
• scale：对流中的每个元素都乘上一个数，做数乘。

add_stream(S1, S2) ->
    case {is_empty_stream(S1), is_empty_stream(S2)} of
        {true, true} ->
            the_empty_stream();
        {true, false} ->
            S2;
        {false, true} ->
            S1;
        {false, false} ->
            cons_stream(
                head(S1) + head(S2),
                fun() ->
                    add_stream(tail(S1), tail(S2))
                end)
    end.

scale(C, S) ->
    map_stream(fun(E) -> C * E end, S).

test_add_stream() ->
    [4, 6] = collect_stream(add_stream(integers(2), integers(3, 4))).

test_scale() ->
    [2, 4] = collect_stream(scale(2, integers(2))). 

前面讲到无限流，例如integers，都是生成式的，那些递归方法一个一个生成下面的元素。这里，我们换一个方式，把流本身当作一个参数来进行操作。这样定义出来的流叫做隐式定义流。

先定义一个全部是1的流，ones()。也就是说，这个流的第1个元素是1，第二个元素应该是个流，这个流也应该是全部为1的流，全部为1的流就是它自身。

%% Defining streams implicitly
ones() ->
    % cons_stream(1, fun() -> ones() end).
    cons_stream(1, fun ones/0).

test_ones() ->
    [1, 1, 1, 1, 1] = collect_stream_limit(5, ones()),
    test_ones_ok.

注释掉的写法是我自己写的，和老师的写法是等价的。

为什么可以这样定义？cons_stream的第二个参数是一个delay的操作。落实到Erlang语言，跟Lisp和Scheme中的定义略微有点不同。在Erlang中之所以可以这样定义，是因为函数在Erlang中有名字空间，在“ones() –>”这里定义了，在cons_stream()中就可以用。在Lisp或Scheme中，ones是作为变量放在里面的，之所以在cons_stream()中能用是因为函数体部分是被当做文本或语言结构存起来的，不对这部分内容进行解释，等到函数要被执行了，才会去解释。在Erlang中可以这样写，是因为它允许函数在自己的函数中调用自己。

再定义一个从1开始的正整数流，用add_stream/2和ones/0定义，integers()。 1 2 3 4 …… 1 1+1 1+2 1 + 3 ……

integers() ->
    cons_stream(1, 
                fun() ->
                    add_stream(integers(), ones())
                end).

test_integers() ->
    [1, 2, 3, 4, 5] = collect_stream_limit(5, integers()),
    test_integers_ok.

可以跟之前的integers_from()比对一下：

integers_from(N) ->
    cons_stream(N, fun() -> integers_from(N + 1) end).

图中“+”三角形就是add_stream，两个输入ones和integers都是实线(流)，1是虚线，作为流的初始，有了初始值以后，流才能动起来。

递归定义数据对象。

用隐式的方式定义一个无限的Fibonacci数的流。

先用生成式的方式定义：

fibs(N1, N2) ->
    cons_stream(N1,
                fun() ->
                    fibs(N2, N1 + N2)
                end).

test_fibs() ->
    [0, 1, 1, 2, 3, 5] = collect_stream_limit(6, fibs(0, 1)),
    test_fibs_ok.

作为一个流来说，无论如何都要有初始元素来激发整个过程。有了初始元素以后，后面的操作是需要延迟计算的。在生成式方式中，这个延迟计算是每做一次产生下一个元素，现在希望这个延迟计算是把整个流当做操作元素来进行操作。

用递归的方式定义：

fibs() ->
    cons_stream(0,
                fun() ->
                    cons_stream(1,
                                fun() ->
                                    add_stream(fibs(), tail(fibs()))
                                end)
                end).

test_fibs() ->
    [0, 1, 1, 2, 3, 5] = collect_stream_limit(6, fibs(0, 1)),
    [0, 1, 1, 2, 3, 5] = collect_stream_limit(6, fibs()),
    test_fibs_ok.

执行过程如下：

fibs的第一个元素0和第二个元素1是知道的，这一步得到了fibs的第三个元素1，由fibs的第一个元素和tail(fibs)的第一个元素相加得到粗体(红色)标识的“1”。

这时，fibs流后面加了一个“1”，tail(fibs)流后面也加了一个“1”，粗斜体(蓝色)标识的“1”。

fibs流的第二个元素和tail(fibs)流的第二个元素相加得到2，即fibs的第四个元素，粗体(红色)标识的“2”。

这时，fibs流后面加了一个“2”，tail(fibs)流后面也加了一个“2”，粗斜体(蓝色)标识的“2”。

fibs流的第三个元素和tail(fibs)流的第三个元素相加得到3，即fibs的第五个元素，粗体(红色)标识的“3”。

这时，fibs流后面加了一个“3”，tail(fibs)流后面也加了一个“3”，粗斜体(蓝色)标识的“3”。

……

不仅过程可以递归定义，数据也可以递归定义。fibs就是递归定义出来的数据。用这样的方式思考，会越来越模糊过程跟数据。cons_stream虽然现在是拿Pair粘合在一起的，但Pair本身只有一个公理定义，底下可以是数据也可以是过程，数据完全可以用过程表示出来，分不清什么是真正的数据、什么是真正的过程。

图中“+”三角形还是add_stream，两个输入fibs和tail(fibs)都是实线(流)，需要两个初始值0和1。

之所以这样可以运行起来，全部依赖于cons_stream中第二个参数只是一个promise，而不是真正地去做。

Integrator，积分电路、求和器、积分器。上面的图特别像数字电路。假如对某个函数y=f(x)求积分，即求函数曲线下的面积。输入S是这个函数的一个一个y值构成的流。scale是在S的基础上做一个缩放，这里的缩放就是f(x)•dt的那个dt。然后在“+”三角形(add_stream)中求和。输出也是一个流，每个点都是到这个点的面积、积分值、求和值。S0是初始值，让积分器活动起来。

先写一个求积分的函数integral。

%% Integrator
integral(S, Init, Dt) ->
    cons_stream(Init,
                fun() ->
                    add_stream(scale(Dt, S), integral(S, Init, Dt))
                end).

test_integral() ->
    [0, 1, 2, 3, 4] = collect_stream_limit(5, integral(ones(), 0, 1)),
    test_integral_ok.

integral的下一个元素就是由它本身再加上下一个f(x)•dt产生的。

接下来，给一个函数f(x)，怎么把流S构建起来，让积分器动起来？

写一个函数funmaps(Fun, Init, Dt)。输入：Fun就是函数f；Init是指从哪里开始产生流S，指定x；Dt就是delta。输出一个流。用生成式的方式定义：

funmaps(Fun, Init, Dt) ->
    cons_stream(Fun(Init),
                fun() ->
                    funmaps(Fun, Init + Dt, Dt)
                end).

test_funmaps() ->
    [0, 1, 2, 3, 4] = collect_stream_limit(5, funmaps(fun(X) -> X end, 0, 1)),
    test_funmaps_ok.

分别求函数f(x) = x从1到2的积分流S1，求函数f(x)=x * x从1到2的积分流S2，求函数f(x)=x * x从0到1的积分流S3。

test_funmaps_integral() ->
    IdFs = funmaps(fun(X) -> X end, 1, 0.001),
    S1 = integral(IdFs, 0, 0.001),
    io:format("~p~n", [nth_stream(1001, S1)]),

    SqFs = funmaps(fun(X) -> X * X end, 1, 0.001),
    S2 = integral(SqFs, 0, 0.001),
    io:format("~p~n", [nth_stream(1001, S2)]),

    SqFs2 = funmaps(fun(X) -> X * X end, 0, 0.001),
    S3 = integral(SqFs2, 0, 0.001),
    io:format("~p~n", [nth_stream(1001, S3)]).

需要注意nth_stream第一个参数的取值。如果Dt为0.001，nth_stream第一个参数为1001，注意不是1000而是1001，要多一个，否则计算结果会不在误差范围之内。如果Dt为0.01，nth_stream第一个参数为101。

运行结果：

从数学角度可以验证结果的正确性。

对于函数f(x)=x，如下图所示：

函数f(x)=x从1到2的积分即图中粗体梯形的面积：(1+2)*1/2 = 1.5。

对于函数f(x)=x*x，根据牛顿-莱布尼兹公式(Newton-Leibniz formula)，一个连续函数在区间[a, b]上的定积分等于它的任意一个原函数在区间[a, b]上的增量。

对于f(x)=xx来说，F(x)=1/3 * xx*x， 求1到2的积分：F(2) – F(1) = 8/3 – 1/3 = 7/3 ≈ 2.333333 求0到1的积分：F(1) – F(0) = 1/3 – 0 = 1/3 ≈ 0.333333

体会一下用“流”的方式思考问题。

前面的funmaps是用生成式的方式定义的，如何用隐式的方式定义？或者说如何基于流操作把funmaps定义出来？进一步来看，我们想定义的funmaps流是一个什么样的流。这个流跟积分器有没有什么关系。假设积分器已经是一个组件了，用积分器怎么写出funmaps。

funmaps产生一个f(x)的流。假设我们已经有了一个我们需要的x流，对x流做一个f映射。x流的特点是：x、x+dt、x+dt+dt ……，是一个累加。积分器/求和器也是累加。如果想用积分器构建出x流的话，积分器的输入是ones流，经过scale(dt)以后是恒定的dt流，如果再加上funmaps的Init——x的初始值，积分器的输出就是x流。在这个x流上map上f，就得到了f(x)流，再把这个f(x)流作为下一个积分器的输入。

integrator(Fun, Init, Dt) ->
    Ones = ones(),
    S = map_stream(Fun, integral(Ones, Init, Dt)),
    integral(S, 0, Dt).

test_integrator() ->
    S1 = integrator(fun(X) -> X end, 1, 0.001),
    io:format("~p~n", [nth_stream(1001, S1)]),

    S2 = integrator(fun(X) -> X * X end, 1, 0.001),
    io:format("~p~n", [nth_stream(1001, S2)]),

    S3 = integrator(fun(X) -> X * X end, 0, 0.001),
    io:format("~p~n", [nth_stream(1001, S3)]).

运行结果好久才出来，但结果和上一小节中的结果是一样的。

把dt改大一点，从0.001改为0.01，nth_stream的第一个参数从1001改为101，算得很快。

test_integrator() ->
    S1 = integrator(fun(X) -> X end, 1, 0.01),
    io:format("~p~n", [nth_stream(101, S1)]),

    S2 = integrator(fun(X) -> X * X end, 1, 0.01),
    io:format("~p~n", [nth_stream(101, S2)]),

    S3 = integrator(fun(X) -> X * X end, 0, 0.01),
    io:format("~p~n", [nth_stream(101, S3)]).

运行结果：

接下来看一下流编程中的delay可能会带来一些什么样的问题。

微分方程。求解微分方程也可以用流的思路来逼近。

有一个函数y，它的微分dys，经过积分以后得到它。而且它本身具备这样的性质，它自己平方以后得到它的微分。square就是一个map操作。初始值y0。

根据上面的图，定义出dys和ys这两个流。

%% Differential Equation
ys() ->
    integral(dys(), 1, 0.001).
    
dys() ->
    map_stream(fun(E) -> E * E end, ys()).

这样定义，代码能编译过，但运行时会出现如下问题，内存耗尽了。

编译没问题，跟定义递归函数没有问题是一样的，在Erlang中定义递归函数或函数时，函数名称就已经发挥作用了，就能用了，和变量名称不一样。

但在运行环境中，一调用ys()，就开始integral(dys(), …)，integral是一个函数，对dys要进行求值，在dys的求值过程中，map_stream函数的第二个参数ys要被求值。这就出问题了。这个问题怎么解决呢？

其实，我们一直在解决这个问题。我们在做cons_stream的时候，有了一个初始值以后，使得流本身在名义上可用了以后，才用头算后面的元素，其实后面的元素都是被延迟来计算的。我们一直都在解决类似的问题，否则不可能在流还没有完整定义的时候，就直接对这个流整体来操作。

所以，解决这个问题，我们也需要手工地加delay，放一个promise。

integral_delay(DelayS, Init, Dt) ->
    cons_stream(Init,
                fun() ->
                    S = DelayS(),
                    add_stream(scale(Dt, S), integral(S, Init, Dt))
                end).

ys() ->
    integral_delay(fun() -> dys() end, 1, 0.001).

dys() ->
    map_stream(fun(E) -> E * E end, ys()).

想让cons_stream去解耦表面上看到的顺序和真正执行的顺序，增加表达力的话，增加一个Promise (DelayS) 进去，使得我能够更好地构建程序。

OO把时间引入进来，赋值把时间引入进来，我们要非常显式地考虑时间，但时间如果都考虑进去，状态不断地在变化的话，程序的组合性很差，可理解性也比较差，也不优雅。所以希望一点一点地放弃时间，这里的放弃是指把时间作为整体来考虑而不是单独去考虑。

如果有一种语言，它所有的计算都是Promise，直到真正需要它，即碰到原生算子或真地需要获取它的值，计算才会发生。如果这些都在语言中内置支持，这是一种正则序的语言。

我们平常用的语言一般都是应用序的。在调用函数的时候，先求解参数，全部求解完了以后，取出函数体，把参数的值填到函数体中，再继续求解。

以求factorial为例。下面的程序是我们经常写的一个迭代的过程。如果用替换模型来解释，每次fact()参数值都在变，都保存在状态中。

fact(N) -> fact(N, 1).

fact(0, R) -> R;
fact(N, R) -> fact(N -1, R * N).

把上面的程序稍微再写清楚一点。

fact(N, R) when N > 0 ->
    fact(N - 1, R * N);
fact(N, R) when N =:= 0 ->
    R.

在比较N > 0 和 N =:= 0时是碰到了原生算子，需要计算。但在没有需要真正去获取结果的时候，在正则序中，所有的乘积是被累积的。

正则序语言的关键是把时间抛弃、放在自己内部考虑。当你把程序编写得极致优雅的时候，理解力可能会比较弱一些。

Erlang是应用序语言，没办法表达正则序。老师自己写了一个小语言。首先，定义了变量x，是0。再定义一个函数id，参数是n，把n设置到x中，再返回n，所以是identity函数。再定义一个函数inc，参数是a，把a加1并返回。最后定义一个函数y，先调用(id 3)，返回的值再inc，最后返回计算结果。

如果在正则序中，在问y是几的时候，才会真正去计算。

直接写x，表示要求x的值：x -----> 0

再写y，表示要求y的值：y -----> 4

再写x，表示要求x的值：x -----> 3

这里不是要求大家这样写程序，只是举个例子，当我们把时间全部放弃以后，就会发现当我问它是几的时候，可能没有办法回答。在程序的组合性上可能会更高，但可能不那么容易理解，调试也会更不容易。

也有正则序语言的坚持者，但他们首先要解决dragging-tail的问题。

这里引出正则序和应用序是想说明，OO是OO看待问题解决问题的视角，流处理是流处理看待问题解决问题的视角，一个是把时间当作一个单独的维度考虑，随着时间变化，一个是把时间和其他维度合起来，放弃时间，作为一个整体考虑。在有了组合性的同时，可能会带来程序不太好理解的问题。但这种方式的拥趸会说，程序优雅了肯定要付出代价，而且这个代价不算大。

其实纯数学函数真的非常好，组合性非常好，而且重要的是并行起来非常容易。如果经常从流的角度或方式思考问题的话，视角会不同。但想要我们构建的软件能够实用的话，肯定要跟现实世界有连接，如果仅仅在连接的地方付出一点代价，那么，无论是从系统复杂性还是其他方面来说，我们都是可以控制的，程序可以做得非常好。但如果我们不换这种视角，在每个地方都把时间单独拎出来考虑的话，会引入其他更多的复杂性。