RxJS入门（5）----编写并发程序

并发是指同时无误、有效地做一些事。为了完成这些，我们创建我们的程序充分利用时间在一个最有效的方式一起执行这些任务。例如，我们程序每天的并发包含当其他事件发生时用户接口的响应，有效处理成百上千用户的命令。

在本章节中，我们通过在浏览器中设计一个射击飞船的游戏来探索RxJS中的concurrency（并发）和pure function（纯函数）。首先，我们将介绍Observable的pipeline，链接Observable操作符并传递状态的技术。然后，我们将会展示，在不依赖额外的状态和产生副作用的情况下通过使用pipeline建立我们的程序，并在Observable内部囊括你所有的逻辑和状态。

视频游戏需要技术算计程序来维持好多状态，使用强大的Observable pipeline和伟大的RxJS的操作符，我们在没有任何额外状态的情况下编写我们的游戏。

Purity and the Observable Pipeline

一个Observable pipeline是一组链式的操作符，每一操作符需要一个Observable作为入参，并且返回一个Observable作为出参。在这本书中，我么一直在使用pipeline；使用RxJS编程它们是无处不在的，下面是一个例子：

Rx.Observable
.from(1, 2, 3, 4, 5, 6, 7, 8)
.filter(function(val) { return val % 2; })
.map(function(val) { return val * 10; });

pipeline是自包含的。所有的状态流都是从一个操作符传递给另外一个，不需要额外的变量。由于我们在编写响应式程序，我们企图在Observable pipeline之外存储状态（第五章）。这迫使我们保存设置在pipeline之外的变量，所有的这种bean-counting很容易导致bug。为了避免这些，在pipeline中的操作符 一直使用pure函数。

pure函数在给定同样的入参时一直返回同样的出参。很容易设计高并发的程序并且程序中函数不修改其他依赖函数。

Avoiding External State

在下面的例子中，我们统计偶数的个数，在每个间隔时间后创建一个Observable并且让evenTicks自增当我们接收到一个偶数的时候。

var evenTicks = 0;
function updateDistance(i) {
if (i % 2 === 0) {
evenTicks += 1;
}
return evenTicks;
}
var ticksObservable = Rx.Observable
.interval(1000)
.map(updateDistance)
ticksObservable.subscribe(function() {
console.log('Subscriber 1 - evenTicks: ' + evenTicks + ' so far');
});

如下是这个程序运行4秒之后的输出：

Subscriber 1 - evenTicks: 1 so far

Subscriber 1 - evenTicks: 1 so far

Subscriber 1 - evenTicks: 2 so far

Subscriber 1 - evenTicks: 2 so far

现在为了好玩，我们在增加另外一个订阅到ticksObservable：

var evenTicks = 0;
function updateDistance(i) {
if (i % 2 === 0) {
evenTicks += 1;
}
return evenTicks;
}
var ticksObservable = Rx.Observable
.interval(1000)
.map(updateDistance)
ticksObservable.subscribe(function() {
console.log('Subscriber 1 - evenTicks: ' + evenTicks + ' so far');
});
ticksObservable.subscribe(function() {
console.log('Subscriber 2 - evenTicks: ' + evenTicks + ' so far');
});

四秒后输入如下：

Subscriber 1 - evenTicks: 1 so far

Subscriber 2 - evenTicks: 2 so far

Subscriber 1 - evenTicks: 2 so far

Subscriber 2 - evenTicks: 2 so far

Subscriber 1 - evenTicks: 3 so far

Subscriber 2 - evenTicks: 4 so far

Subscriber 1 - evenTicks: 4 so far

Subscriber 2 - evenTicks: 4 so far

在每秒的间隔下,Subscriber2在偶数秒的时候完全完成。它使用和Subscribe1一样的evenTicks，如你猜测的原因一样，Observable pipeline为每个订阅者都调用一次，每秒都调用了两次。

共享外部的状态变量而引起的问题比这个例子更加发展。在复杂的项目中，在pipeline外部开放式的改变状态导致代码更加复杂，bug也更多。解决的办法是，尽可能的在我们pipeline内部封装更多的信息。我们将上面的代码封装起来避免外部状态如下：

function updateDistance(acc, i) {
if (i % 2 === 0) {
acc += 1;
}
return acc;
}
var ticksObservable = Rx.Observable
.interval(1000)
.scan(updateDistance, 0);
ticksObservable.subscribe(function(evenTicks) {
console.log('Subscriber 1 - evenTicks: ' + evenTicks + ' so far');
});
ticksObservable.subscribe(function(evenTicks) {
console.log('Subscriber 2 - evenTicks: ' + evenTicks + ' so far');
});

》

Subscriber 1 - evenTicks: 1 so far

Subscriber 2 - evenTicks: 1 so far

Subscriber 1 - evenTicks: 1 so far

Subscriber 2 - evenTicks: 1 so far

Subscriber 1 - evenTicks: 2 so far

Subscriber 2 - evenTicks: 2 so far

Subscriber 1 - evenTicks: 2 so far

Subscriber 2 - evenTicks: 2 so far

使用scan，我们避免了完全的外部状态。我们使用了updateDistance来累计偶数秒，而不是依赖外部的变量去维护这个值。这样，我们就不用使用每个新订阅者来增加同一个计数器了。

大部分的时候我们可以避免依赖外部的状态。通常，在程序中，我们使用缓存或者是一直保存这个值的改变。正如在下面的Space Reactive章节中看到，这些场景可以使用其他方式处理。例如，我们需要缓存值的时候，RxJS’s Subject Class这个章节会起到一些作用，当我们需要保持游戏之前的状态时，我们可以使用类似Rx.Observable.scan的方法。

Pipelines Are Efficient

第一次我链装一堆操作符成一个pipeline去改变一个sequence，我的直觉是，这不可能是有效率的。我知道在JavaScript中通过链式操作符改变数组的代价是昂贵的。在这本书中，我将告诉教你设计程序来改变sequence通过新的途径。这是极其有效的。

Observable的链式和数组的看起来很相似。两种类型中都有一些方法如：filter，map等。但是它们有截然不同的：数组的方法给每个操作的结果都创建了一个新数组，下一个操作在对它进行一次。Observable pipeline，换句话说，它不会立即创建一个Observable并对每一个元素应用所有的操作符。这样，这个Observable就会仅仅被倒腾一次，这使得链式的Observable效率高的，比较下面的两个例子：

stringArray // represents an array of 1,000 strings
.map(function(str) {
❶ return str.toUpperCase();
})
❷ .filter(function(str) {
return /^[A-Z]+$/.test(str);
})
❸ .forEach(function(str) {
console.log(str);
});

假设stringArray是一个有1000个字符串的数组，我们想转化为大写字母，并且过滤掉字母表中的字符（不包含字母）。最后我们把每一个串都在控制台中打印出来。

背后有对应如下场景发生：

1：迭代数组，并创建所有项大写的新数组。

2：迭代这个大写的数组，并创建一个1000个元素的新数组。

3：迭代这个过滤后的数组，并且在控制台打印每一个结果。

在上面的数组的转化过程中，我们迭代了数组三次，并且创建了两个新的完整的大数组。这就是最大的不同。如果你是在关心变现或者是正在处理多item的sequence，你就不应该这样编码。

下面是在同样的操作符下使用Observable：

stringObservable // represents an observable emitting 1,000 strings
.map(function(str) {
❶ return str.toUpperCase();
})
❷ .filter(function(str) {
return /^[A-Z]+$/.test(str);
})
❸ .subscribe(function(str) {
console.log(str);
});

Observable pipeline和数组链极其相似，但是他们的相似也是仅仅止步于此，在一个Observable内部，不管我们申请了多少查询和修改，都不会发生任何操作直到我们订阅。当我们像map一样链一个改变，我们正组成了一个单个的函数在每个数组的项上仅仅操作了一次。所以在上面的代码中，将会发生如下：

1：创建一个转大写的函数，它将会应用在Observable的每一个项上，并且返回一个发射这些新项的Observable，当一个observer订阅它的时候。

2：和先前的转大写的函数一起组装一个新的过滤函数，并发射这些转大写、过滤后的新项的Observable，当仅当一个observer订阅它的时候。

3：这个Observable发射的触发器，应用我们给每元素定义的改变仅仅一次。

使用Observable，我们仅仅操作列表一次。我们仅仅应用这些操作当它真正需要的时候。例如，我们给上面的代码增加一个take操作符。

stringObservable
.map(function(str) {
return str.toUpperCase();
})
.filter(function(str) {
return /^[A-Z]+$/.test(str);
})
.take(5)
.subscribe(function(str) {
console.log(str);
});

take使得Observable仅仅发射我们指定的前面的n个项。本例子中，n是5，所以在上千个串中，我们仅仅会收到前面的5个。最酷的地方是，我们的代码永远都不会所有的项，它仅仅改变前面5个。

这个使得开发者的生活变得更加容易了。你可以大胆的休息，当操作sequence时，RxJS会尽可能多的帮你做。这种操作的方式叫做lazy evaluation（懒解决），在类似于Haskell和Miranda的函数语言中，这很常见。

RxJS’s Subject Class

一个Subject是一个既继承了Observable也继承了Observer的的类型。作为observer（观察者/订阅者），它可以订阅Observable，作为Observable（可被订阅者），它可以产生很多值让Observer来订阅。

在一些情况下，单个的Subject可以做Observable和observer的组合的功能。例如在一个数据源和一个Subject的监听者之间可以创建一个代理对象，我们可以如下使用：

var subject = new Rx.Subject();
var source = Rx.Observable.interval(300)
.map(function(v) { return 'Interval message #' + v; })
.take(5);
source.subscribe(subject);
var subscription = subject.subscribe(
function onNext(x) { console.log('onNext: ' + x); },
function onError(e) { console.log('onError: ' + e.message); },
function onCompleted() { console.log('onCompleted'); }
);
subject.onNext('Our message #1');
subject.onNext('Our message #2');
setTimeout(function() {
subject.onCompleted();
}, 1000);

》

onNext: Our message #1

onNext: Our message #2

onNext: Interval message #0

onNext: Interval message #1

onNext: Interval message #2

onCompleted

在上面的代码中，我们创建了一个新的Subject，一个source Observable（每秒发射一个整数）。之后，我们订阅这个Subject倒那个Observable上。最后，我们订阅一个observer到这个Subject自己本身。现在这个Subject就可以像一个Observable一样了。

接着，我们使这个Subject发射它自己的值（message1和message2）。在最后的结果中，我们首先获取的是它自己的值，之后才是从source代理的值。这些来自Observable的值要晚些，是由于他们是异步的，而不是像我们使用Subject自己的值一样实时地。注意到，即使我们告诉source Observable发射前面5个值，但是最后的结果仅仅展示了3个，这是由于，在一秒之后，我们调用了Subject的onCompleted函数。这种情况完成了所有订阅的通知，并且覆盖了take操作符。

这个Subject类提供了基础的创建更多的特定的Subject。事实上，Rxjs有好多有意思的：AsyncSubject, ReplaySubject,和

BehaviorSubject。

AsyncSubject

AsyncSubject发射序列的最后一个值，仅仅当序列完成时。这个值将会一直被缓存。其他订阅者这个值已经被发射过后，都会立即收到它。AsyncSubject对单个返回值的异步操作是很方便的。比如ajax请求。让我们来瞧瞧一个AsyncSubject订阅到一个range的简单例子。

var delayedRange = Rx.Observable.range(0, 5).delay(1000);
var subject = new Rx.AsyncSubject();
delayedRange.subscribe(subject);
subject.subscribe(
function onNext(item) { console.log('Value:', item); },
function onError(err) { console.log('Error:', err); },
function onCompleted() { console.log('Completed.'); }
);

在这个例子中，delayedRange在延迟一秒后将会发射0~4的 。接着，我们创建了一个新的AsyncSubject Subject并且把它订阅到 delayedRange上。输出入下：

》

Value: 4

Completed.

正如期盼，我们得到了Observable发射的最后一个值。现在让我们使用AsyncSubject在一个比较实际的场合中。我们将会去一些远程的内容：

function getProducts(url) {
var subject;
❶ return Rx.Observable.create(function(observer) {
if (!subject) {
subject = new Rx.AsyncSubject();
❷ Rx.DOM.get(url).subscribe(subject);
}
❸ return subject.subscribe(observer);
});
}
❹ var products = getProducts('/products');
// Will trigger request and receive the response when read
❺ products.subscribe(
function onNext(result) { console.log('Result 1:', result.response); },
function onError(error) { console.log('ERROR', error); }
);
// Will receive the result immediately because it's cached
❻ setTimeout(function() {
products.subscribe(
function onNext(result) { console.log('Result 2:', result.response); },
function onError(error) { console.log('ERROR', error); }
);
}, 5000);

上面的代码中，当getProduct使用url别调用时，它返回一个observer（发射那个http get请求的结果）。下面就是它的拆解：

1：getProduct返回一个Observable sequence，我们在这里创建它。

2：如果我们没有创建一个AsyncSubject，我们创建一个把它订阅到Rx.Observalbe.get(url)返回的Observable上。

3：我们把observer订阅到AsyncSubject上，每次当一个observer订阅到那个Observable上，它都会实际上被订阅到那个AsyncSubject上，AsyncSubject扮演了获取url的Observable和Observer之间的代理。

4：我们创建了通过url“product.json”获取的那个Observable，并且把它在procucts变量中存储。

5：这是第一个订阅，它将开始url数据的获取，当结果获取完成后并打印出来。

6：这是第二个订阅，它将在第一个订阅完成5秒后开始。由于早前的url已经获取了，所以没有必要再次发起网络请求，它将会立即获取请求的结果，因为在AsyncSubject Subject中已近存在这个值。

这很有趣，当我们使用AsyncSubject订阅到一个Rx.DOM.Request.get Observable。由于AsyncSubject缓存最后一个结果，任何序列的订阅都会没有网络请求地立即接受到这个结果。可以使用AsyncSubject当我们期望一个单个结果并且保存它的时候。

Does That Mean AsyncSubject Acts Like a Promise?

AsyncSubject代表了异步操作的结果，你可以使用它作为promise的替代者。内部的区别是：promise仅处理一个值，而AsyncSubject处理序列中的所有值仅是发射并缓存最后一个。

可以如此容易地模拟promise体现了Rxjs的灵活性。（其实没有AsyncSubject，Rxjs也可以比较容易地使用Observable模拟promise）

BehaviorSubject

当一个订阅者订阅到一个BehaviorSubject时，它收到最后发射的一个值和之后所有的序列值。BehaviorSubject需要我们提供一个开始值，以便所有的订阅者都将接收到当它们订阅到BehaviorSubject时的值。

假设我们想获取一个远程的文件并且打印它的内容在一个html页面上。但是在获取等待内容时，我们想用占位字符。我们可以使用BehaviorSubject如下：

var subject = new Rx.BehaviorSubject('Waiting for content');
subject.subscribe(
function(result) {
document.body.textContent = result.response || result;
},
function(err) {
document.body.textContent = 'There was an error retrieving content';
}
);
Rx.DOM.get('/remote/content').subscribe(subject);

上述的代码中，我们使用占位符初始化一个新的BehaviorSubject，然后我们订阅它，并且通过依赖结果使用onNext和onError来改变HTML body的值。

现在HTML body包含我们的占位符字符，它会一直保持直到Subject发射了一个新的值。最后，我们请求我们需要的资源，并订阅Subject到正在监测结果的observer上。

BehaviorSubject保证至少会发射一个值，这是由于我们在它的构造函数中提供了一个初始的默认值。一旦BehaviorSubject完成了，它不会再发射任何值，并释放缓存值使用的内存。

ReplaySubject

ReplaySubject缓存它的值并再发射它们到之后订阅它们的任何observer上。不像AsyncSubject，这个序列不需要completed来发生。



ReplaySubject很有用，它确保observer能获取Observable从开始发送的所有值。它使我们不写凌乱的代码来缓存之前的值，使我们远离不干净的并发相关的bug。

当然，为了达到这个目的，ReplaySubject在内存中缓存了所有的值。为了阻止它使用太多的内存，我们可以限制存储数据的buffer大小或者窗口时间，也可以传递特定的参数给构造函数。

ReplaySubject构造函数的第一个参数是一个数字，它代表着我们想buffer（缓存）多少值。

var subject = new Rx.ReplaySubject(2); // Buffer size of 2
subject.onNext(1);
subject.onNext(2);
subject.onNext(3);
subject.subscribe(function(n) {
console.log('Received value:', n);
});

》

Received value: 2

Received value: 3

第二个参数是一个数字，它代表着我们需要缓存存在的单位毫秒数。

var subject = new Rx.ReplaySubject(null, 200); // Buffer size of 200ms
setTimeout(function() { subject.onNext(1); }, 100);
setTimeout(function() { subject.onNext(2); }, 200);
setTimeout(function() { subject.onNext(3); }, 300);
setTimeout(function() {
subject.subscribe(function(n) {
console.log('Received value:', n);
});
subject.onNext(4);
}, 350);

》

Received value: 2

Received value: 3

Received value: 4

- 在上面的例子中，我们创建了一个基于时间的buffer，而不是值的数量。我们的ReplaySubject缓存值，直到200毫秒前释放。我们发射了三个值每隔100毫秒，350毫秒之后，我们订阅到了一个observer并且发射了另外一个值。发生订阅的这个时候，它缓存了2和3，这是由于1发射在250秒之前，所以1不会被缓存了。

- Subject是一个能节省你好多时间的强有力的工具。它提供了好多好的解决途径如缓存和重复等。由于在它的核心，它既是Observable也是observer，所以你不需要学起其他的新东西。

Spaceship Reactive!

为了展示我们可以保持程序的纯粹性，我们建立了一个video game，在里面，我们的英雄无休止和敌人的大波的飞船战斗。我们将极大的使用Observable pipeline，我将指出试图存贮在pipeline之外的状态，并如何避免它。

viedo games众所周知地需要保存一些外部的状态，如分数， 人物的屏幕坐标，计时器等等。我们的目标是建立一个完整的没有依赖任何外部状态变量的游戏。

在我们的游戏中，游戏者将会使用鼠标水平的移动飞船，并且使用鼠标点解或者敲击空格键来射击。我们的游戏有四个角色：背景中移动的星；游戏者的飞船；敌人；来自游戏者或者敌人的射击。

如下图所示：



在屏幕射击或中，红色的三角形代表我们的飞船，绿色的是敌人，哪些极小的三角形是射击的子弹。

让我们开始搭台子，如实时候我们的HTML文件：

<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>Spaceship Reactive!</title>
<script src="../rx.all-4.0.0.js"></script>
<style>
html, body {
margin: 0;
padding: 0;
}
</style>
</head>
<body>
<script src="spaceship.js"></script>
</body>
</html>

这仅仅是一个加载js文件的HTML文件，在余下的本章节中，我们会完成。在那个js文件中，我们以设置一个需要重绘我们游戏的canvas为开始。

var canvas = document.createElement('canvas');
var ctx = canvas.getContext("2d");
document.body.appendChild(canvas);
canvas.width = window.innerWidth;
canvas.height = window.innerHeight;

在这个地方，我们可以开始描述我们的组件了，首先绘制布满星星的背景。

Creating the Star Field

第一件事我们为游戏设置是空间上的星星。我们创建向下滚动的星星来创建空间旅游的感觉。我们使用range操作符生成哪些星星。

var SPEED = 40;
var STAR_NUMBER = 250;
var StarStream = Rx.Observable.range(1, STAR_NUMBER)
.map(function() {
return {
x: parseInt(Math.random() * canvas.width),
y: parseInt(Math.random() * canvas.height),
size: Math.random() * 3 + 1
};
})

每一颗星都代表着一个包含着随机坐标和大小在1~4的对象。这将会产生250颗星的流。

我们需要让这些星星一直移动。一种途径是每隔若干毫秒让这些星星的y坐标增加。我们可以使用toArray把星星流的Observable变为数组，这之后发射包含range产生的所有对象的数组。接着，我们使用那个数组，它使用flatMap操作符的Observable转化为每若干毫秒产生一个值。使用map我们可以增加数组每个原始项的y坐标。

var SPEED = 40;
var STAR_NUMBER = 250;
var StarStream = Rx.Observable.range(1, STAR_NUMBER)
.map(function() {
return {
x: parseInt(Math.random() * canvas.width),
y: parseInt(Math.random() * canvas.height),
size: Math.random() * 3 + 1
};
})
.toArray()
.flatMap(function(starArray) {
return Rx.Observable.interval(SPEED).map(function() {
starArray.forEach(function(star) {
if (star.y >= canvas.height) {
star.y = 0; // Reset star to top of the screen
}
star.y += 3; // Move star
});
return starArray;
});
})

在map内部，我们检查是否星星的y坐标是否已经超出屏幕，超出重置为0。我们可以一直使用同样的星星数组来改变每个星星对象的坐标。

现在我们需要一个小小的帮助函数给给我们canvas上的星星数组添上颜色：

function paintStars(stars) {
ctx.fillStyle = '#000000';
ctx.fillRect(0, 0, canvas.width, canvas.height);
ctx.fillStyle = '#ffffff';
stars.forEach(function(star) {
ctx.fillRect(star.x, star.y, star.size, star.size);
});
}

paintStars绘制了一个黑色的背景并在上面绘制星星。唯一留下的事就是需要移动星星，办法是订阅到那个Observable上在结果数组上调用paintStars。下面就是最终的代码：

function paintStars(stars) {
ctx.fillStyle = '#000000';
ctx.fillRect(0, 0, canvas.width, canvas.height);
ctx.fillStyle = '#ffffff';
stars.forEach(function(star) {
ctx.fillRect(star.x, star.y, star.size, star.size);
});
}
var SPEED = 40;
var STAR_NUMBER = 250;
var StarStream = Rx.Observable.range(1, STAR_NUMBER)
.map(function() {
return {
x: parseInt(Math.random() * canvas.width),
y: parseInt(Math.random() * canvas.height),
size: Math.random() * 3 + 1
};
})
.toArray()
.flatMap(function(starArray) {
return Rx.Observable.interval(SPEED).map(function() {
starArray.forEach(function(star) {
if (star.y >= canvas.height) {
star.y = 0; // Reset star to top of the screen
}
star.y += 3; // Move star
});
return starArray;
});
})
.subscribe(function(starArray) {
paintStars(starArray);
});

现在我们已经设置了这个台子，是时候让我们的英雄出现了。

var HERO_Y = canvas.height - 30;
var mouseMove = Rx.Observable.fromEvent(canvas, 'mousemove');
var SpaceShip = mouseMove
.map(function(event) {
return {
x: event.clientX,
y: HERO_Y
};
})
.startWith({
x: canvas.width / 2,
y: HERO_Y
});

注意到我使用了startWith()，它会设置Observable的第一个值，我把它设置我屏幕中间的位置。没有startWith，我们的Observable只有当游戏者移动鼠标的时候才会发射值。

让我们在屏幕上渲染我们的hero，在这个游戏中，所有的角色都是三角形（我的图形设计没有给人留下深刻的映像），因此我们将定义一个帮助函数来绘制canvas上给定坐标、大小、颜色的三角形，代码如下：

function drawTriangle(x, y, width, color, direction) {
ctx.fillStyle = color;
ctx.beginPath();
ctx.moveTo(x - width, y);
ctx.lineTo(x, direction === 'up' ? y - width : y + width);
ctx.lineTo(x + width, y);
ctx.lineTo(x - width,y);
ctx.fill();
}

我们也定义paintSpaceShip，它使用如下的帮助函数：

function paintSpaceShip(x, y) {
drawTriangle(x, y, 20, '#ff0000', 'up');
}

但是我们现在遇到了一个难题，如果我们订阅到SpaceShip Observable并调用drawTriangle，我们的飞船仅仅是在我们移动鼠标的时候是瞬间可见的。这是有由于starStream每秒被canvas更新下好多次，清除掉了我们的飞船当不移动鼠标的时候。并且由于starStream没有直接拥有spaceship（飞船），我们没办法在starStream订阅的时候渲染spaceship。starStream可以保存飞船最后的一个坐标，之后我们需要改变我们不需要外部变量的规则么，怎么办？

通常，Rxjs有一个非常方便的操作符来解决这种问题。

Rx.Observable.combineLatest是一个很方便的操作符。他需要两个及以上的Observable并发送每个Observable上一次最后发射的结果（无论什么时候它发射一个新值）。starStream发射一个新项（星星的数组）很频繁，我们可以取消starStream订阅并使用combineLatest去组合starStream和SpaceShip而成Observable，一旦他们中的任何一个发射了新项就会立即更新：

function renderScene(actors) {
paintStars(actors.stars);
paintSpaceShip(actors.spaceship.x, actors.spaceship.y);
}
var Game = Rx.Observable
.combineLatest(
StarStream, SpaceShip,
function(stars, spaceship) {
return { stars: stars, spaceship: spaceship };
});
Game.subscribe(renderScene);

现在我们使用renderScene绘制屏幕上的所有事，所以你可以去掉StarStream中的如下订阅代码：

.subscribe(function(starArray) {
paintStars(starArray);
});

至此，我们就能绘制所有的星星背景和飞船每当Observable发射新值的时候。我们现有有了飞行的飞船，同时我们也能使用鼠标移动它。这一点代码还不赖！但是在这个宽阔的太空中我们的飞船太寂寞了。如何给他一些同伴？

Generating Enemies

如果没有敌人需要操心的话，这将会是一个无聊的游戏。所以我们来创建一个无穷的敌人。我们想每秒半创建一个新的敌人，但不能覆盖我们的英雄。让我们一起看看敌人的Observable：

var ENEMY_FREQ = 1500;
var Enemies = Rx.Observable.interval(ENEMY_FREQ)
.scan(function(enemyArray) {
var enemy = {
x: parseInt(Math.random() * canvas.width),
y: -30,
};
enemyArray.push(enemy);
return enemyArray;
}, []);
var Game = Rx.Observable
.combineLatest(
StarStream, SpaceShip, Enemies,
function(stars, spaceship, enemies) {
return {
stars: stars,
spaceship: spaceship,
enemies: enemies
};
});
Game.subscribe(renderScene);

为了创建敌人，我们使用interval操作符每1500毫秒运行一次，之后，我们使用scan操作符创建敌人的数组。

我们可以简单的看下前面在Observable中聚合这一节中scan操作符的的描述。scan聚合Observable发射的每一个结果并且立即发射每一个结果。在敌人的Observable中我们以一个空的数组作为scan的第一个参数开始，同时，我们每次迭代都保存一个object里面。这个对象包含屏幕可见之外的一个随机的x坐标和一个固定的y坐标。这样Enemies将会每隔1500毫秒发射数组所有的现存的敌人。

仅仅遗留的重绘敌人是一个在canvas上绘制的帮助函数。这个函数也更新敌人数组中每一个项的坐标：

// Helper function to get a random integer
function getRandomInt(min, max) {
return Math.floor(Math.random() * (max - min + 1)) + min;
}
function paintEnemies(enemies) {
enemies.forEach(function(enemy) {
enemy.y += 5;
enemy.x += getRandomInt(-15, 15);
drawTriangle(enemy.x, enemy.y, 20, '#00ff00', 'down');
});
}

在paintEnemies中，我们已将看到随机改变x坐标一边一边敌人的移动是无规则的。现在我们需要更新renderScene函数来包含paintEnemies的调用。

你可能发现了到目前为止的当你玩这个游戏时的一个奇怪的影响：当你移动鼠标时，敌人跑的比你快。这有可能是一个比较漂亮的特点。但我们确实不需要，你能猜猜这是什么导致的么？

如果你猜到 了它和paintEnemies函数相关， 你是对的。combineLatest渲染了我们的场景无论哪些Observable产生了什么值。如果我们不移动鼠标，最快的发射者应该是starStream，因为它有一个40毫秒的interval（敌人的Observable是1500毫秒），当我们移动鼠标，SpaceShip将不会比starStream快（你的鼠标发射坐标多次每秒），并且paintEnemies将会执行这些多次，导致了敌人的坐标更快。

为了解决些问题和相似的情景，我们需要标准化游戏的速度，以便没有Observable能发射的速度快过我们为游戏选定的速度。

当然，你也猜到了，Rxjs有这样的一个操作符。