在研究kafka过程中需要用到kafka的管理平台， 但这个管理平台用sbt进行编译，并且使用了 Play Framework框架进行开发，因此查阅了资料， 发现了下面很好的相关教程

从文章中获取 部分信息如下， 全部信息请到 原文的论坛处获取：http://www.scala-china.net/discuz/forum.php?mod=viewthread&tid=2376&extra=page%3D2

在博客前面已经介绍过两个用于开发Web应用的框架，采用PHP开发的Yii 开发教程 和基于纯Java的Vaadin Web开发教程 ，此外还有基于Ruby的Ruby on Rail 和基于.Net 的 ASP.NET MVC4 ，其中Ruby on Rail 和 ASP.NET MVC 教程已经非常多，也不准备再重复这些教程。

本 篇开始介绍Play Framework ，它也是一个用来开发Web应用的开发框架，可以使用Java和Scala，但本教程将使用Scala作为开发语言，Scala语言比较新，但它已经成为 我最喜欢的语言之一，如果你喜欢使用Java，PHP，Perl，C#或者JavaScript那么你一定也会喜欢上Scala, 它集中了这些语言的优点，同时除了支持面对象的编程外，它还是一种函数式编程语言(Functional programming)，支持扩展，你可以根据需要扩展Scala语言，这是一般的开发语言所不支持的。

Play Framework 基于Akka (并行应用开发包，这个开发包的一个显著特定是基于Actor模型，并行运算是这个某些的基本特点，可以避免由于一般多线程开发不正确使用同步或互锁造成 的程序死锁等问题）。采用MVC模型（非常类似于ASP.NET MVC4），缺省支持REST和JSON，支持编译CoffeeScript和LESS代码。 支持数据库编程，并且可以和Slick（类似于LINQ的数据库接口）。开发可以使用Eclipse ，Intellij IDEA或是Sbt(Typesafe 自带的编译工具）。

本教程采用Eclipse 开发环境配合sbt 工具开发。

下面简单介绍一下Play Framework开发环境的搭建，（Windows开发环境）。

• 首先需要安装Java SDK ,建议安装最新版Java SDK ，64位系统安装64位Java SDK。
• 安装Scala编译工具sbt.
• 安装Scala开发环境
• 安装Scala IDE for Eclipse
• 下载解压Play Framework ，目前版本为2.1.3

安装好之后，注意把这些命令行所在的目录添加的系统环境变量PATH中，然后到到命令行中执行

play new myFirstPlayApp

在提示选择语言（Java或Scala）选择 Scala

然后进入 新创建myFirstPlayApp ，运行 play run ，如果编译成功，将显示下面类似的消息，表示新创建的Web应用运行在端口 9000

打开浏览器 输入地址http://localhost:9000,如果显示如下页面，将表示开发环境已经就绪，后面文章将介绍这个例子。

相信Scala 语言会逐渐普及开来，Play Framework 也会得到广泛的应用，目前使用Play Framework的一些著名网站有 Linked In等

注：本博客的一个新的别名　http://www.scala-playframework.com ，之后可能会作为专门介绍Scala和Play Framework的中文技术论坛。

Play应用标准的目录结构

app                      → 应用的源码
└ assets                → 可以编译的Asset源码
└ stylesheets        → 通常为Less代码
└ javascripts        → 通常为CoffeeScipts代码
└ controllers           → 应用的controllers类定义
└ models                → 应用的Model类定义
└ views                 → UI模板
conf                     → 配置文件
└ application.conf      → 主应用配置文件
└ routes                → 路由定义
public                   → 公开的Asset定义
└ stylesheets           → CSS文件
└ javascripts           → Javascript文件
└ images                → Image文件
project                  → sbt配置文件（用来编译项目）
└ build.properties      → sbt 项目属性定义
└ Build.scala           → 应用编译脚本定义
└ plugins.sbt           → sbt插件定义
lib                      → 应用使用的库文件（不由sbt管理的其它库）
logs                     → 日志目录
└ application.log       → 缺省的应用日志文件
target                   → 编译生成的文件目录
└ scala-2.10.0
└ cache
└ classes            → 编译过的class文件
└ classes_managed    → Managed class 文件 (模板等, ...)
└ resource_managed   → Managed 资源(less, ...)
└ src_managed        → 生成的源码(模板, ...)
test                     → 单元测试或功能测试代码目录

app目录

app目录包含了所有可运行代码或资源:Scala代码，UI模板和Less代码，CoffeeScript代码， app通常包含下面三个子目录：对于与MVC的三个部分：

• app/controllers
• app/models
• app/views

当然你可以添加你自己的目录，比如app/utils。此外还有一个可选的标准目录名assets
app/assets ，用来存放可供编译的LESS或CoffeeScript代码

public目录

public目录存放了可以供网页直接访问的资源，比如CSS，图像，JavaScript文件。

conf目录

该目录存放应用的配置文件，主要有下面两种配置：

• application.conf 应用主配置文件，定义了应用的一些配置参数
• routes ,路由配置文件

lib目录

主要用来添加一些第三方库文件，这些库主要指一些没有通过sbt或编译脚本定义的库文件。直接把jar文件添加到这个目录下，编译时会直接添加到应用的classpath.

project目录

定义sbt编译项目的定义，包含下面三个文件：

• plugins.sbt 定义项目使用的插件定义
• Build.scala 定义项目编译脚本.
• build.properties 指定编译sbt的版本等.

target目录

包含了编译后生成的一下文件。

如果你使用GIT做为版本管理工具，下面为典型的 .gitingore 文件：
logs
project/project
project/target
target
tmp
dist
.cache
即使你使用IntelliJ IDEA或是Eclipse IDE来开发Play应用，了解Play Console的使用还是非常有用的，Play Console 为一基于sbt的命令行工具，可以用来管理Play应用的开发生命周期。

启动Play Console

使用Play Console前，注意把Play所在目录添加到系统目录中，这样，启动Play Console 非常简单，只需在Play应用所在目录在命令行输入 “play”

play

获得Play 命令帮助

你可以输入help play 获得关于Play命令使用的基本方法和命令：
常见的命令如下：
1    play                         Enter the play console
2    help                         Displays this help message or prints detailed help on requested commands (run 'help <command').
3    about                        Displays basic information about sbt and the build.
4    tasks                        Lists the tasks defined for the current project.
5    settings                     Lists the settings defined for the current project.
6    reload                       (Re)loads the project in the current directory
7    projects                     Lists the names of available projects or temporarily adds/removes extra builds to the sessio.
8    project                      Displays the current project or changes to the provided `project`.
9    set                          Evaluates a Setting and applies it to the current project.
10    session                      Manipulates session settings.  For details, run 'help session'.
11    inspect                      Prints the value for 'key', the defining scope, delegates, related definitions, and dependenies.
12    ; <command> (; <command>)*   Runs the provided semicolon-separated commands.
13    ~ <command>                  Executes the specified command whenever source files change.
14    last                         Displays output from a previous command or the output from a specific task.
15    last-grep                    Shows lines from the last output for 'key' that match 'pattern'.
16    exit                         Terminates the build.
17    show <key>                   Displays the result of evaluating the setting or task associated with 'key'.

在开发模式下启动Play应用

使 用run 命令，可以开发模式下启动Play应用，此时一些调整信息可以在屏幕上显示出来，缺省情况下，Play Web应用使用端口9000，如果你修改代码，Play Console会检测到代码变化，在下个请求时会自动重新编译代码或资源，如要必要，还可以重新启动Play应用。如果出现编译错误，那么在浏览器中会显 示错误信息：

如果要停止Web服务，可以按Ctrl+D ，此外还可以通过任务管理器，杀死对应的Java进程来停止Web服务。

编译

Compile 命令可以编译Play 源码或是资源（如CoffeeScript代码，LESS代码，Scala代码等），如果出现编译错误，屏幕上显示对应的错误提示：

调试命令

如 果你使用Eclipse IDE,你可以使用Eclipse来调试代码，此时需要使用 play debug 启动一个JPDA调试端口，缺省调试端口为 9999,此后可以通过配置Eclipse远程调试指向这个端口，就可以使用Eclipse和其它Java 调试器来调试代码，如果使用IntelliJ IDEA IDE可以自动完成这些配置。

强制清理编译结果

如果你觉得编译的结果出现意想不到的机会，那么你的sbt的缓存可能出现问题，此时可以通过 clean-all 命令强制清除之前的编译结果，然后使用compile 重新编译生成新的编译内容。

此外play Console 基于 sbt 工具，因此你也可以使用sbt 的一些功能 ，比如 ~compile ,~run, ~test等

Distributed RPC(分布式RPC)

下面是 storm的原文（http://storm.apache.org/documentation/Distributed-RPC.html）

The idea behind distributed RPC (DRPC) is to parallelize the computation of really intense functions on the fly using Storm. The Storm topology takes in as input a stream of function arguments, and it emits an output stream of the results for each of those function calls.
DRPC是为了真正实现在storm上高速计算的。Storm topology获取输入方法参数流，发射每个方法调用的结果数据流。

DRPC is not so much a feature of Storm as it is a pattern expressed from Storm’s primitives of streams, spouts, bolts, and topologies. DRPC could have been packaged as a separate library from Storm, but it’s so useful that it’s bundled with Storm.
DRPC并不是Storm的一个特性，不像Storm的streams，spouts，bolts和topology。DRPC可以被打包到一个独立于Storm的包，但是它对Storm是很有用的。

分布式RPC（distributed RPC，DRPC）用于对Storm上大量的函数调用进行并行计算过程。对于每一次函数调用，Storm集群上运行的拓扑接收调用函数的参数信息作为输入流，并将计算结果作为输出流发射出去。

DRPC本身算不上Storm的特性，它是通过Storm的基本元素：streams，spouts，bolts，topologies而衍生的一个模式。DRPC可以单独作为一个独立于Storm的库发布，但由于其重要性还是和Storm捆绑在了一起。

总体概述

Distributed RPC is coordinated by a “DRPC server” (Storm comes packaged with an implementation of this). The DRPC server coordinates receiving an RPC request, sending the request to the Storm topology, receiving the results from the Storm topology, and sending the results back to the waiting client. From a client’s perspective, a distributed RPC call looks just like a regular RPC call. For example, here’s how a client would compute the results for the “reach” function with the argument “http://twitter.com”:
分布式RPC是需要”DRPC server”配合的（storm有一个实现）。DRPC server协调接收RPC请求，发送请求到Storm topology, 接收Storm topology的结果，发送结果给等待的客户端。从客户端来看，一个分布式RPC调用就像一个regular RPC调用。例如这有一个客户端使用参数”http://twitter.com“计算”reach”方法结果：

DRPCClient client = new DRPCClient("drpc-host", 3772);
String result = client.execute("reach", "http://twitter.com");

The distributed RPC workflow looks like this:
DRPC工作流程如下：

The distributed RPC workflow looks like this:

A client sends the DRPC

A client sends the DRPC server the name of the function to execute and the arguments to that function. The topology implementing that function uses a DRPCSpout to receive a function invocation stream from the DRPC server. Each function invocation is tagged with a unique id by the DRPC server. The topology then computes the result and at the end of the topology a bolt called ReturnResults connects to the DRPC server and gives it the result for the function invocation id. The DRPC server then uses the id to match up that result with which client is waiting, unblocks the waiting client, and sends it the result.
一个客户端发送DRPC server方法名和方法执行参数。topology实现该方法，用DRPCSpout接收方法调用数据流从DRPC server。每个方法调用会被DRPC server被打上唯一id标识。topology计算结果，在topology的结尾bolt上调用ReturnResults链接DRPC server，给它该方法调用id的结果。DRPC server用这个id匹配等待的客户端，解锁该等待客户端，发送结果给它。

DRPC通过DRPC Server来实现，DRPC Server的整体工作过程如下：

1. 接收到一个RPC调用请求；
2. 发送请求到Storm上的拓扑；
3. 从Storm上接收计算结果；
4. 将计算结果返回给客户端。

以上过程，在client客户端看来，一个DRPC调用看起来和一般的RPC调用没什么区别。下面代码是client通过DRPC调用“reach”函数，参数为“http://twitter.com”：

DRPCClient client = new DRPCClient("drpc-host", 3772); String result = client.execute("reach", "http://twitter.com");

DRPC内部工作流程如下：

1. Client向DRPC Server发送被调用执行的DRPC函数名称及参数。
2. Storm上的topology通过DRPCSpout实现这一函数，从DPRC Server接收到函数调用流；
3. DRPC Server会为每次函数调用生成唯一的id；
4. Storm上运行的topology开始计算结果，最后通过一个ReturnResults的Bolt连接到DRPC Server，发送指定id的计算结果；
5. DRPC Server通过使用之前为每个函数调用生成的id，将结果关联到对应的发起调用的client，将计算结果返回给client。

LinearDRPCTopologyBuilder

Storm提供了一个topology builder——LinearDRPCTopologyBuilder，它可以自动完成几乎所有的DRPC步骤。包括：

1. 构建spout；
2. 向DRPC Server返回结果；
3. 为Bolt提供函数用于对tuples进行聚集。

下面是一个简单的例子，这个DRPC拓扑只是简单的在输入参数后追加“!”后返回：

public static class ExclaimBolt extends BaseBasicBolt { public void execute(Tuple tuple, BasicOutputCollector collector) { String input = tuple.getString(1); collector.emit(new Values(tuple.getValue(0), input + "!")); } public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("id", "result")); } } public static void main(String[] args) throws Exception { LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("exclamation"); builder.addBolt(new ExclaimBolt(), 3); // ... }

由上述例子可见，我们只需很少的工作即可完成拓扑。当创建LinearDRPCTopologyBuilder的时候，需要指定拓扑中DRPC函数的名称“exclamation”。一个DRPC Server可以协调多个函数，每个函数有不同的函数名称。拓扑中的第一个bolt的输入是两个字段：第一个是请求的id号；第二个是请求的参数。

LinearDRPCTopologyBuilder同时需要最后一个bolt发射一个包含两个字段的输出流：第一个字段是请求id；第二个字段是计算结果。因此，所有的中间tuples必须包含请求id作为第一个字段。

例子中，ExclaimBolt在输入tuple的第二个字段后面追加“!”，LinearDRPCTopologyBuilder负责处理其余的协调工作：与DRPC Server建立连接，发送结果给DRPC Server。

本地模式DRPC

DRPC可以以本地模式运行，下面的代码是如何在本地模式运行上面的例子：

LocalDRPC drpc = new LocalDRPC(); LocalCluster cluster = new LocalCluster(); cluster.submitTopology("drpc-demo", conf, builder.createLocalTopology(drpc)); System.out.println("Results for 'hello':" + drpc.execute("exclamation", "hello")); cluster.shutdown(); drpc.shutdown();

首先创建一个LocalDRPC对象，该对象在本地模拟一个DRPC Server，正如LocalCluster在本地模拟一个Storm集群一样。然后创建一个LocalCluster对象在本地模式下运行拓扑。LinearDRPCTopologyBuilder含有单独的方法用于创建本地拓扑和远程拓扑。

本地模式下，LocalDRPC并不绑定任何端口，因此Storm的拓扑需要了解要通讯的对象——这就是为什么createLocalTopology方法需要以LocalDRPC对象作为输入。

加载完拓扑之后，通过对LocalDRPC调用execute方法，就可以执行DRPC函数调用了。

远程模式DRPC

在实际的Storm集群上运行DRPC也一样很简单。只需完成以下步骤：

1. 启动DRPC Server(s)；
2. 配置DRPC Server(s)地址；
3. 向Storm集群提交DRPC拓扑。

首先，通过storm脚本启动DRPC Server：

bin/storm drpc

然后，在Storm集群中配置DRPC Server地址，这就是DRPCSpout读取函数调用请求的地方。这一步的配置可以通过storm.yaml文件或者拓扑的配置来完成。通过storm.yaml文件的配置方式如下：

drpc.servers: - "drpc1.foo.com" - "drpc2.foo.com"

最后，通过StormSubmitter启动DRPC拓扑。为了以远程模式运行上面的例子，代码如下：

StormSubmitter.submitTopology("exclamation-drpc", conf, builder.createRemoteTopology());

createRemoteTopology被用于为Storm集群创建合适的拓扑。

一个复杂的例子

上面的exclamation只是一个简单的DRPC例子。下面通过一个复杂的例子介绍如何在Storm集群内进行DRPC——计算Twitter上每个URL的到达度（reach），也就是每个URL暴露给的不同人的个数。

为了完成这一计算，需要完成以下步骤：

1. 获取所有点选了（tweet）该URL的人；
2. 获取步骤1中所有人的关注者（followers，粉丝）；
3. 对所有关注者followers进行去重；
4. 对步骤3中的关注者人数进行求和。

一个简单的URL到达度计算可能涉及成千上万次数据库调用以及数以百万的followers记录，计算量非常大。有了Storm，将很容易实现这一计算过程。单机上可能需要运行几分钟才能完成，在Storm集群上，即使是最难计算的URL也只需要几秒钟。

这个例子的代码在storm-starter：点击这里。这里是如何创建拓扑的代码：

LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("reach"); builder.addBolt(new GetTweeters(), 3); builder.addBolt(new GetFollowers(), 12) .shuffleGrouping(); builder.addBolt(new PartialUniquer(), 6) .fieldsGrouping(new Fields("id", "follower")); builder.addBolt(new CountAggregator(), 2) .fieldsGrouping(new Fields("id"));

拓扑的执行分为以下四步：

1. GetTweeters：获取所有tweet了指定URL的用户列表，这个Bolt将输入流[id, url]转换成输出流[id, tweeter]，每个url元组被映射为多个tweeter元组。
2. GetFollowers：获取步骤1中所有用户列表的followers，这个Bolt将输入流[id, twetter]转换成输出流[id, follower]，当某个人同时是多个人的关注者follower，而且这些人都tweet了指定的URL，那么将产生重复的follower元组。
3. PartialUniquer：将所有followers按照follower id分组，使得同一个follower在同一个task中被处理。这个Bolt接收follower并进行去重计数。
4. CountAggregator：从各个PartialUniquer中接收各部分的计数结果，累加后完成到达度计算。

下面是PartialUniquer这个Bolt的代码实现：

public class PartialUniquer extends BaseBatchBolt { BatchOutputCollector _collector; Object _id; Set<String> _followers = new HashSet<String>(); @Override public void prepare(Map conf, TopologyContext context, BatchOutputCollector collector, Object id) { _collector = collector; _id = id; } @Override public void execute(Tuple tuple) { _followers.add(tuple.getString(1)); } @Override public void finishBatch() { _collector.emit(new Values(_id, _followers.size())); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("id", "partial-count")); } }

PartialUniquer通过继承BaseBatchBolt实现了IBatchBolt接口，batch bolt提供了API用于将一批tuples作为整体来处理。每个请求id会创建一个新的batch bolt实例，同时Storm负责这些实例的清理工作。

当PartialUniquer接收到一个follower元组时执行execute方法，将follower添加到请求id对应的HashSet集合中。

Batch bolt同时提供了finishBatch方法用于当这个task已经处理完所有的元组时调用。PartialUniquer发射一个包含当前task所处理的follower ids子集去重后个数的元组。

在内部实现上，CoordinatedBolt用于检测指定的bolt是否已经收到指定请求id的所有tuples元组。CoordinatedBolt使用direct streams管理实现这一协作过程。

拓扑的其他部分易于理解。到达度的每一步的计算过程都是并行进行的，通过DRPC实现也是非常容易的。

Non-linear DRPC拓扑

LinearDRPCTopologyBuilder只能处理“线性的”DRPC拓扑——正如到达度这样可以通过一系列步骤序列来完成的计算。不难想象，DRPC调用中包含有更复杂的带有分支和合并Bolt的拓扑。目前，必须自己直接使用CoordinatedBolt来完成这种非线性拓扑的计算。

LinearDRPCTopologyBuilder工作过程

• DRPCSpout发射[args, return-info]，其中return-info包含DRPC Server的主机和端口号，以及DRPC Server为该次请求生成的唯一id号；
• 构造一个Storm拓扑包含以下部分：
  • DRPCSpout
  • PrepareRequest(生成一个请求id，为return info创建一个流，为args创建一个流)
  • CoordinatedBolt wrappers以及direct groupings
  • JoinResult(将结果与return info拼接起来)
  • ReturnResult(连接到DRPC Server，返回结果)
• LinearDRPCTopologyBuilder是建立在Storm基本元素之上的高层抽象。

高级进阶

KeyedFairBolt for weaving the processing of multiple requests at the same time

How to use CoordinatedBolt directly

• KeyedFairBolt用于组织同一时刻多请求的处理过程；
• 如何直接使用CoordinatedBolt。