如何在Jetty使用Comet模式提高伸缩性

感谢philip McCarthy([e-mail])

　异步服务器端事件驱动的Ajax程序很难实现，也很难获得伸缩性。在作者的系列文章里，Plilip McCarthy展示了一个有效的方式:
　Comet模式允许您push数据到客户端，而且Jetty 6的Continuations API让您的Comet程序对大量客户端获得高可伸缩性。您可以方便的同DWR 2使用Comet和Continuations。

　随着Ajax在Web程序开发技术里建立了牢固的位置，出现了几种常见的Ajax使用模式。例如，Ajax通常用于响应用户输入来使用新数据修改局部页面。但有时候，Web程序的用户界面需要根据偶尔的异步服务器端事件来更新，而不需要用户动作 -- 例如，在Ajax聊天程序里显示其他用户输入的一条新消息。由于Web浏览器和服务器间的HTTP连接只能由浏览器建立，服务器不能"推"更改数据到浏览器。

　Ajax程序有两个解决该问题的基本方式:浏览器每隔几秒请求服务器来获得更改，或者服务器维持与浏览器的连接并且传递数据。长连接技术称为Comet。本文展示了怎样使用Jetty服务器引擎和DWR来实现简单而高效的Comet Web程序。

为什么要Comet?
　轮询方式的主要缺点是在大量客户端时产生了大量的传输浪费。每个客户端都必须有规律的请求服务器来获得更改，这是服务器资源的一个重担。最坏的情况是程序很少更新，例如Ajax邮件收件箱。在这种情况下，大量的客户端轮询是多余的，服务器仅仅简单的响应"没有数据"。 可以通过增加轮询间隔时间来减轻服务器负荷，但是这引入了服务器事件和客户端知晓之间的延迟。当然，一个合理的折衷方案可以多数程序适用，并且轮询的工作方式也可以接受。

　然而，对Comet策略的呼唤来自它可感知的高效。客户端不会产生轮询方式特有的传输浪费，一旦事件发生，就会被发布到客户端。但是维持长连接也消耗了服务器资源。当servlet位置持久的请求在等候状态时，servlet独占一个线程。这样传统的servlet引擎就限制了Comet的伸缩性，因为客户端的数量会迅速超过服务器栈可以有效处理的线程的数量。

Jetty 6有什么不同
　Jetty 6设计来处理大量并发连接，它使用Java语言的不堵塞I/O(java.nio)库并且使用优化的输出缓冲架构。Jetty也有一个处理长连接的杀手 锏:一个称为Continuations的特性。我将用一个接收请求然后等待两秒发送响应的简单servlet来示范Continuations。然后， 我将展示当服务器拥有更多的客户端时将发生什么。最后我将使用Continuations重新实现servlet，并且您将看到它们的不同。

　为了让它更简单，我将限制Jetty servlet引擎为一个单一的请求处理线程。列表1显示了相关的jetty.xml配置。事实上我需要允许在ThreadPool里总共有3个线 程:Jetty服务器本身使用一个，HTTP连接器使用一个来监听进来的请求，最后剩一个线程来执行servlet代码。
列表1. 单一servlet线程的Jetty配置

 

代码

<?xml version="1.0"?>
<!DOCTYPE Configure PUBLIC "-//Mort Bay Consulting//DTD Configure//EN"
  "http://jetty.mortbay.org/configure.dtd">
<Configure id="Server" class="org.mortbay.jetty.server">
  <Set name="ThreadPool">
    <New class="org.mortbay.threadBoundedThreadPool">
      <Set name="minThreads">3</Set>
      <Set name="lowThreads">0</Set>
      <Set name="maxThreads">3</Set>
    </New>
  </Set>
</Configure> 

 

　下一步，为了模仿异步事件，列表2显示了BlockingServlet的service()方法，它简单的使用Thread.sleep()调用来在完成前暂停2,000毫秒。同时它也在执行开始和结束时输出系统时间。为了帮助区分不同请求的输出，它也把一个请求参数作为标识符记录到日志。
列表2. BlockingServlet

 

代码 

public class BlockingServlet extends HttpServlet {

  public void service(HttpServletRequest req, HttpServletResponse res) throws java.io.IOException {

    String reqId = req.getParameter("id");

    res.setContentType("text/plain");
    res.getWriter().println("Request: " + reqId + "\tstart:\t" + new Date());
    res.getWriter().flush();

    try {
      Thread.sleep(2000);
    } catch (Exception e) {}

-----------------------------------------------------------

 

　现在您可以观察几个同步请求下servlet的行为。列表3显示了使用lynx的5个并行请求时控制台的输出。命令行简单的启动5个lynx进程，加上一个标识符序数到请求的URL。
列表3. 到BlockingServlet的几个并发请求的输出

 

 

代码 

$ for i in 'seq 1 5' ; do lynx -dump localhost:8080/blocking?id=$i & done
Request: 1 start: Sun Jul 01 12:32:29 BST 2007
Request: 1 end:   Sun Jul 01 12:32:31 BST 2007

Request: 2 start: Sun Jul 01 12:32:31 BST 2007
Request: 2 end:   Sun Jul 01 12:32:33 BST 2007

Request: 3 start: Sun Jul 01 12:32:33 BST 2007
Request: 3 end:   Sun Jul 01 12:32:35 BST 2007

Request: 4 start: Sun Jul 01 12:32:35 BST 2007
Request: 4 end:   Sun Jul 01 12:32:37 BST 2007

Request: 5 start: Sun Jul 01 12:32:37 BST 2007
Request: 5 end:   Sun Jul 01 12:32:39 BST 2007

 

　列表3的输出并不惊奇。由于Jetty只有一个线程来执行servlet的service()方 法，Jetty将每个请求列队并按顺序服务。时间戳显示了在一个应答分派给一个请求(以及end消息)后，servlet开始处理下一个请求(下一个 start消息)。所以即使所有的5个请求是同时发出的，最后的那个请求必须等待8秒才能得到处理。

　现在，看看Jetty 6的Continuations特性在这种情形下是多么的有用。列表4显示了列表2的BlockingServlet使用Continuations API重写后的样子。我将在后面解释代码。
列表4. ContinuationServlet

 

代码 

public class ContinuationServlet extends HttpServlet {

  public void service(HttpServletRequest req, HttpServletResponse res) throws java.io.IOException {

    String reqId = req.getParameter("id");

    Continuation cc = ContinuationSupport.getContinuation(req, );

    res.setContentType("text/plain");
    res.getWriter().println("Request: " + reqId + "\tstart:\t" + new Date());
    res.getWriter().flush();

    cc.suspend(2000);

    res.getWriter().println("Request: " + reqId + "\tend:\t" + new Date());
  }
}

 

列表5显示了对ContinuationServlet作5个并发请求时的输出，可以和列表3比较一下。
列表5. 到ContinuationServlet的几个并发请求的输出

 

代码

$ for i in 'seq 1 5' ; do lynx -dump localhost:8080/continuation?id=$i & done
Request: 1 start: Sun Jul 01 12:37:37 BST 2007
Request: 1 start: Sun Jul 01 12:37:39 BST 2007
Request: 1 end:   Sun Jul 01 12:37:39 BST 2007

Request: 3 start: Sun Jul 01 12:37:37 BST 2007
Request: 3 start: Sun Jul 01 12:37:39 BST 2007
Request: 3 end:   Sun Jul 01 12:37:39 BST 2007

Request: 2 start: Sun Jul 01 12:37:37 BST 2007
Request: 2 start: Sun Jul 01 12:37:39 BST 2007
Request: 2 end:   Sun Jul 01 12:37:39 BST 2007

Request: 5 start: Sun Jul 01 12:37:37 BST 2007
Request: 5 start: Sun Jul 01 12:37:39 BST 2007
Request: 5 end:   Sun Jul 01 12:37:39 BST 2007

Request: 4 start: Sun Jul 01 12:37:37 BST 2007
Request: 4 start: Sun Jul 01 12:37:39 BST 2007
Request: 4 end:   Sun Jul 01 12:37:39 BST 2007

 

　在列表5里有两件重要的事情值得注意。首先，每个start消息出现了两次，暂时不要担心这点。其 次，更重要的是，现在请求是并发处理的，没有排队。注意所有的start和end消息时间戳是一样的。因此，没有哪个请求耗时超多两秒，即使只有单一的 servlet线程在运行。

深入Jetty的Continuations机制
　理解Jetty的Continuations机制的将解释您在列表5里 看到的东西。为了使用Continuatins，Jetty必须配置为使用它的SelectChannelConnector处理请求。这个 connector构建在java.nio API之上，允许它维持每个连接开放而不用消耗一个线程。当使用SelectChannelConnector 时，ContinuationSupport.getContinuation()提供一个 SelectChannelConnector.RetryContinuation实例(但是，您必须针对Continuation接口编程)。当在 RetryContinuation上调用suspend()时，它抛出一个特殊的运行时异常 -- RetryRequest，该异常传播到servlet外并且回溯到filter链，最后被SelectChannelConnector捕获。但是不会 发送一个异常响应给客户端，而是将请求维持在未决 Continuations队列里，则HTTP连接保持开放。这样，用来服务请求的线程返回给ThreadPool，然后又可以用来服务其他请求。

　暂停的请求停留在未决 Continuations队列里直到指定的过期时间，或者在它的Continuation上调用resume()方法。当任何一个条件触发时，请求会重 新提交给servlet(通过filter链)。这样，整个请求被"重播"直到RetryRequest异常不再抛出，然后继续按正常情况执行。

　列表5里的输出现在应该能理解了。对每个请求，按顺序进入到servlet的service()方法，start消息发送给应答，然后 Continuation的suspend()方法保留servlet，然后释放线程来开始下一请求。所有的5个请求迅速运行service()方法的第 一部分并马上进入暂停状态，所有的start消息在几毫秒内输出。两秒后，suspend()过期，第一个请求从未决队列里重新得到并重新提交给 ContinuationServlet。start消息第二次输出，对suspend()方法的第二次调用立即返回，然后end消息被发送给应答。然后 servlet代码执行下一个队列请求，以此类推。

　所以，在BlockingServlet和ContinuationServlet两种情况下，请求被排入队列来访问单 一的servlet线程。尽管如此，在BlockingServlet里的两秒钟暂停在servlet线程里执行 时，ContinuationServlet的暂停发生于servlet外面的SelectChannelConnector里。 ContinuationServlet全部的吞吐量会更高，因为servlet线程不会在sleep()调用时阻碍大多数时间。

 

-----------------------------------------------------------

 

让Continuations变得有用
　现在您已经看到Continuations运行servlet请求暂停而不消耗线程，我需要多解释一下Continuations API来展示怎样使用Continuations达到特殊的目的。

　一个resume()方法和一个suspend()方法配对。您可以认为它们是标准的Object wait()/notify()机制的Continuations等价物。即，suspend()维持一个Continuation直到过期或者另一个线程调用resume()。suspend()/resume()方法是使用Continuations实现真实的Comet风格服务的关键所在。基本的模式是从当前请求维持Continuation，调用suspend()，然后等待直到您的异步时间到达，然后调用resume()并生成应答。

　但是，不像编程语言里真实的语言级continuations，如Scheme，或Java语言里的wait()/notify()，在 Jetty Continuation上调用resume()并不意味着代码执行于它停止的确切位置。您已经看到，真正发生的是与Continuation相关的请求 被重播。这导致两个问题:列表4的ContinuationServlet里代码不合需要的重新执行，以及丢失状态 -- 暂停时作用域里的任何东西都丢失了。

　第一个问题的解决方案是isPending()方法，如果isPending()方法的返回值为true，这意味着suspend()在前面已经 被调用过了，并且二次请求的执行不会再次接触suspend()方法。换句话说，给您的suspend()调用前的代码加上isPending()条件可 以确保它只被执行一次。Continuation也提供了一个简单的机制来保持状态:putObject(Object)和getObject()方法。 使用它们来维持一个context对象，这样当Continuation暂停时任何您需要维持的状态都可以得到保护。您也可以使用该机制作为一种在线程之 间传递事件数据的方法，后面您将看到。

写一个基于Continuations的程序
　作为一个真实世界里的例子，我将开发一个基本的GPS坐标跟踪Web程序。 它将在无规律间隔内生成随机的纬度-经度对。假设生成的坐标可以为附近的公众移动位置，如拿着GPS设备马拉松运动员，成队的汽车，或者运输中的包裹位 置。有意思的部分在于我怎样告诉浏览器坐标信息。图1显示了这个简单的GPS跟踪程序的类图:
图1. 显示GPS跟踪程序主要组件的类图

　 首先，该程序需要生成坐标的一些东西，这是RandomWalkGenerator的工作。从一个初始坐标开始，每次对它的私有方法 generateNextCoord()的调用都从该位置随机走一步并返回一个GpsCoord对象。当初始化 时，RandomWalkGenerator创建一个线程，该线程在随机间隔内调用generateNextCoorld()方法并发送生成的坐标给任何 使用addListener()注册自己的CoordListener实例。
列表6显示了RandomWalkGenerator的循环逻辑:
列表6. RandomWalkGenerator的run()方法

 

代码 

public void run() {

  try {
    while (true) {
      int sleepMillis = 5000 + (int)(Math.random()*8000d);
      Thread.sleep(sleepMillis);
      dispatchUpdate(generateNextCoord());
    }
  } catch (Exception e) {
    throw new RuntimeException(e);
  }
}

 

　CoordListener是一个定义了onCoord(GpsCoord coord)方法的回调接口。在例子中，ContinuationBasedTracker类实现了CoordListener。 ContinuationBasedTracker的另外一个方法为getNextPosition(Continuation, int)。列表7显示了这些方法的具体实现:
列表7. ContinuationBasedTracker的内脏

 

代码 

public GpsCoord getNextPosition(Continuation continuation, int timeoutSecs) {

  synchronized(this) {
    if (!continuation.isPending()) {
      pendingContinuations.add(continuation);
    }

    // wait for next update
    continuation.suspend(timeoutSecs*1000);
  }

  return (GpsCoord)continuation.getObject();
}

public void onCoord(GpsCoord gpsCoord) {

  synchronized(this) {
    for (Continuation continuation : pendingContinuations) {

      continuation.setObject(pgsCoord);
      continuation.resume();
    }

    pendingContinuations.clear();
  }
}

 

-----------------------------------------------------------

 

　当客户端在Continuation里调用getNextPosition()时，isPending()方法检查这次请求不是重试，然后添加它 到一个等待坐标的Continuations集合里，然后Continuation被暂停。同时，onCoord -- 当生成新坐标时调用 -- 简单的循环每个未决Continuations，为它们设置GPS 坐标，然后恢复它们。然后每个重试的请求完成getNextPosition()的执行，从Continuation得到GpsCoord并返回它给调用 者。注意这里需要同步，不仅预防pendingContinuations集合里出现不一致的状态，也确保了新添加的Continuation在它被暂停 之前不会被恢复。

谜题最后一部分是servlet代码本身，显示于列表8:
列表8. GPSTrackerServlet实现

 

 

代码 

public class GpsTrackerServlet extends HttpServlet {

  private static final int TIMEOUT_SECS = 60;
  private ContinuationBasedTracker tracker = new ContinuationBasedTracker();

  public void service(HttpServletRequest req, HttpServletResponse res) throws java.io.IOException {

    Continuation c ContinuationSupport.getContinuation(req, );
    GpsCoord position = tracker.getNextPosition(c, TIMEOUT_SECS);

    String json = new Jsonifier().toJson(position);
    res.getWriter().print(json);
  }
}

 

　您可以看到，servlet所做很少。它简单的维持请求的Continuation，调用 getNextPosition()，转换GPSCoord为JavaScript Object Notation(JSON)并输出。这里不需要防止任何代码重执行，所以我不需要检查isPending()。列表9显示了对 GpsTrackerServlet的调用的输出，使用服务器可得到的单一线程上的5个并发请求。
列表9. GPSTrackerServlet输出

 

代码 

$ for i in 'seq 1 5' ; do lynx -dump localhost:8080/tracker & done
  { coord : { lat : 51.51122, lng : -0.08103112 } }
  { coord : { lat : 51.51122, lng : -0.08103112 } }
  { coord : { lat : 51.51122, lng : -0.08103112 } }
  { coord : { lat : 51.51122, lng : -0.08103112 } }
  { coord : { lat : 51.51122, lng : -0.08103112 } }

 

　这个例子不是很引人注目但却是概念的证明。在请求分派后，它们被维持开发几秒钟直到生成坐标，这时迅速产生应答。这是Comet模式的基本原理，使用Jetty单一线程处理5个并发请求，感谢Continuations。

创建一个Comet客户端
　现在您已经看到Continuations怎样用于创建非阻塞Web服务，您可能想知道怎样创建客户端代码来使用它。一个Comet客户端需要:
1. 维持一个XMLHttpRequest连接直到接收应答.
2. 分派应答给合适的JavaScript处理者.
3. 立即建立一个新连接.
　更高级的Comet可以在客户端和服务器使用合适的路由机制来使用一个连接来从多个不同的服务推数据到浏览器。一个可能性为使用JavaScript库如Dojo等写客户端代码来提供基于Comet的请求机制，形如dojo.io.comet。

　尽管如此，如果您正在使用Java作为服务器语言，在客户端和服务器端得到高级Comet支持的更好的方式是使用DWR 2。如果您不熟悉DWR，您可以该系列的第3部分，"Ajax with Direct Web Remoting"。DWR透明的提供一个HTTP-RPC传输层，暴露您的Java对象来使用JavaScript代码调用。DWR生成客户端代理，自 动marshall和unmarshall数据，处理安全问题，提供一个便利的客户端辅助库，并且对所有主要的浏览器工作。

DWR 2:反转Ajax
　DWR 2新引入的概念为反转Ajax。该机制将服务端事件"推"给客户端。客户端DWR代码透明的处理连接建立和应答解析，所以从开发人员的角度来看，事件可以从服务端Java代码简单的发布到客户端。

　DWR可以配置使用3个不同的机制来反转Ajax。一种是我们熟悉的轮询方式。第二种方式称为piggyback，它不创建任何到服务器的连接， 而是等待直到另一个DWR服务调用发生并piggyback未决事件到该请求应答。这可以获得高效率但是意味着客户端事件通知被延迟直到客户端作出一个不 相干的调用。最后一种机制使用Comet风格的长连接。最好的是，当DWR运行在Jetty下并且使用Continuations来获得非阻塞Comet 时可以自动检测事件。

　我将修改我的GPS例子来使用DWR 2反转Ajax。同时，您将看到反转Ajax怎样工作的更多细节。

　我不再需要我的servlet。DWR提供了一个controller servlet，它协调客户端请求直接访问Java对象。我也不再需要显示处理Continuations，因为DWR在幕后处理了这些。所以我只需要一个新的CoordListener实现来发布坐标更新到任何客户端浏览器。

　一个称为ServerContext的接口提供DWR的反转Ajax魔法。ServerContext知道当前查看一个给定页面的所有Web客户 端并且可以提供一个ScriptSession来与每个客户端交流。ScriptSession用来从Java代码推JavaScript片段到客户端。 列表10显示了ReverseAjaxTracker怎样响应坐标通知，以及使用它们来生成客户端updateCoordinate()方法调用。注意如 果一个合适的转换器是可用的，则DWR的ScriptBuffer对象的appendData()调用会自动marshall一个Java对象到 JSON，。
列表10. ReverseAjaxTracker里的通知回调方法

 

代码 

public void onCoord(GpsCoord gpsCoord) {

  // Generate JavaScriptcode to call client-side
  // function with coord data
  ScriptBuffer script = new ScriptBuffer();
 script.appendScript("updateCoordinate(").appendData(gpsCoord).appendScript(");");

  // Push script out to clients viewing the page
  Collection<ScriptSession> sessions = sctx.getScriptSessionsByPage(pageUrl);

  for (ScriptSession session : sessions) {
    session.addScript(script);
  }
}

 

　下一步，DWR必须配置来知道ReverseAjaxTracker。在更大的程序里，DWR的 Spring集成可以使用Spring创建的beans来提供DWR。但是这里，我将仅仅让DWR创建一个新的ReverseAjaxTracker实例 并把它放在application作用域里。所有后续的DWR请求将访问这个单一的实例。

　我也需要告诉DWR怎样从GpsCoord beans来marshall数据到JSON。由于GpsCoord是一个简单对象，DWR基于反射的BeanConverter足够。
列表11显示了ReverseAjaxTracker配置。
列表11. ReverseAjaxTracker的DWR配置

 

代码 

<dwr>
  <allow>
    <create creator="new" javascrit="Tracker" scope="application">
      <param name="class"value="developerworks.jetty6.gpstracker.ReverseAjaxTracker"/>
    </create>

    <convert converter="bean" match="developerworks.jetty6.gpstracker.GpsCoord"/>
  </allow>
</dwr>

-----------------------------------------------------------

 

　create元素的javascript元素指定了DWR用来暴露tracker作为一个JavaScript对象的名字。但是，在这里，我的客户端代码不会使用它，而是从tracker推数据给它。同时，也需要在web.xml里做一些额外的配置来让DWR使用反转Ajax，见列表12。
列表12. DwrServlet的web.xml配置

 

 

代码 

<servlet>
  <servlet-name>dwr-invoker</servlet-name>
  <servlet-class>
    org.directwebremoteing.servlet.DwrServlet
  </servlet-class>
  <init-param>
    <param-name>activeReverseAjaxEnabled</param-name>
    <param-value>true</param-value>
  </init-param>
  <init-param>
    <param-name>initApplicationScopeCreatorsAtStartup</param-name>
    <param-value>true</param-value>
  </init-param>
</servlet>

 

　第一个servlet init-param，activeReverseAjaxEnabled，激活轮询和Comet功能。 第二个，initApplicationScopeCreatorsAtStartup，告诉DWR当程序开始时初始化 ReverseAjaxTracker。这会覆盖通常在bean上作第一次请求时的延迟初始化行为 -- 在这里这是很有必要的，因为客户端从不在ReverseAjaxTracker上调用方法。

　最后，我需要实现从DWR调用的客户端JavaScript方法。回调方法updateCoordinate()被传递一个JSON形式的 GpsCoord对象，它由DWR的BeanConverter自动序列化。这个方法仅仅从坐标提取longitude和latitude域并通过DOM 调用添加它们到一个列表里。这在列表13里显示了，同我的页面的onload方法一起。onload包含对 dwr.engine.setActiveReverseAjax(true)，这告诉DWR打开一个到服务器的持久的连接来等待回调。
列表13. 反转Ajax GPS跟踪的客户端实现

 

代码 

window.onload = function() {
  dwr.engine.setActiveReverseAjax(true);
}

function updateCoordinate(coord) {
  if (coord) {
    var li = document.createElement_x("li");
    li.appendChild(document.createTextNode(coord.longitude + ", " + coord.latitude));
    document.getElementByIdx_x("coords").appendChild(li);
  }
}

 

　现在我可以让我的浏览器访问跟踪程序页面，当坐标数据开始生成时DWR将开始推数据到客户端。这个实现将简单的输出一个生成的坐标列表，见图2:
图2. ReverseAjaxTracker输出

　使用反转Ajax创建一个事件驱动的Ajax程序是如此简单。记住，感谢DWR对Jetty Continuations的使用，当等待新事件到达时线程不会阻塞在服务器。

　据此，很容易从Yahoo!或者Google集成一个地图窗口部件。通过改变客户端回调方法，坐标可以简单的传递到地图API，而不是直接添加到页面。图3显示了在这样的一个地图组件上DWR反转Ajax GPS跟踪程序描绘的随机路线:
图3. 使用地图UI的ReverseAjaxTracker

结论
　现在您看到了Jetty Continuations联合Comet可以提供一个高效的、可伸缩的事件驱动Ajax程序的解决方案。我没有给出Continuations的伸缩性的图，因为性能在 真是世界里取决于许多变数。服务器硬件、操作系统的选择、JVM实现、Jetty配置、您的Web程序的设计和传输效率在负荷下都会影响Jetty Continuations的性能。尽管如此，Webtide的Greg Wilkins(首要的Jetty开发者) 发布了一个比较Jetty 6集成Continuations与不集成Continuations的Comet程序处理10,000并发请求时的性能的白皮书。在Greg的测试里， 使用Continuations并去掉了线程消费和栈内存消费，使用大于10的因数。

　您也看到了使用DWR的反转Ajax技术实现事件驱动的Ajax程序是多么容易。DWR不仅节省您的客户端和服务端代码，反转Ajax也将整个服 务器推机制从您的代码中抽象出来。您可以随意转换您的Comet方式:轮询或者piggyback方式，只需简单的更改DWR配置。您可以随意试验并找到 适合您的程序的最佳策略而不会影响您的代码。