Java学习——Jetty服务端编程

sunny03

　　下面测试使用Jetty的continuation机制，首先自定义一个服务，分别添加blockservlet和nonblockservlet。开启5个线程的线程池提供server服务。（jetty 8.1.8版本）

package test;

import org.eclipse.jetty.server.*;
import org.eclipse.jetty.server.nio.SelectChannelConnector;
import org.eclipse.jetty.servlet.ServletContextHandler;
import org.eclipse.jetty.servlet.ServletHolder;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
public class PJetty{
public static void main(String[] args) throws Exception {
//设置connectors
SelectChannelConnector connector1 = new SelectChannelConnector();
connector1.setPort(1987);
connector1.setThreadPool(new QueuedThreadPool(5));
Server server = new Server();
server.setConnectors(new Connector[]{connector1});
//使用servlet处理请求
ServletContextHandler context = new ServletContextHandler();
context.setContextPath("/");
context.addServlet(new ServletHolder(new BlockingServlet()), "/block");
context.addServlet(new ServletHolder(new NonBlockingServlet()), "/nonblock");
server.setHandler(context);
server.start();
server.join();
}
}
　　先实现一个阻塞的servlet服务BlockingServlet.java如下

package test;
import java.text.SimpleDateFormat;
import java.util.Date;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
public class BlockingServlet extends HttpServlet {
/**
* generated serialize number
*/
private static final long serialVersionUID = -3402527424876194224L;
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");//可以方便地修改日期格式

public void service(HttpServletRequest req, HttpServletResponse res)
throws java.io.IOException {
String start = dateFormat.format( new Date() );
String sleeptime = req.getParameter("st");
res.setContentType("text/plain");
res.getWriter().println("Request: "+sleeptime+"\tstart:\t" + start);
res.getWriter().flush();
try {
if(sleeptime != null){
Thread.sleep(Integer.parseInt(sleeptime)*1000);
}
} catch (Exception e) {}
String end = dateFormat.format( new Date() );
res.getWriter().println("Request: "+sleeptime+"\tend:\t" + end);
}
}
　　然后实现一个NonBlockingServlet.java如下，continuation采用suspend/resume模式

package test;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.eclipse.jetty.continuation.Continuation;
import org.eclipse.jetty.continuation.ContinuationSupport;
public class NonBlockingServlet extends HttpServlet {
/**
* generated serialize number
*/
private static final long serialVersionUID = 3313258432391586994L;
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");//可以方便地修改日期格式
MyAsyncHandler myAsyncHandler ;
public void init() throws ServletException {
myAsyncHandler = new MyAsyncHandler() {  
public void register(final MyHandler myHandler) {
try {  
Thread.sleep(10000);
String end = dateFormat.format( new Date() );
myHandler.onMyEvent(end);  
} catch (InterruptedException e) {  
// TODO Auto-generated catch block  
e.printStackTrace();  
}  
}  
};
}
protected void doGet(HttpServletRequest request, HttpServletResponse response) throws IOException
{
String start = dateFormat.format( new Date() );
String reqId = request.getParameter("st");
// 如果我们需要异步方式得到一个result，并放入request中
Object results = request.getAttribute("results");
if (results==null) // 如果异步处理尚未返回结果
{
final Continuation continuation = ContinuationSupport.getContinuation(request);
if(continuation.isInitial()){
continuation.setTimeout(8000);
response.setContentType("text/plain");
response.getWriter().println("Request: "+reqId+"\tstart:\t"+start);
response.getWriter().flush();
}
// 判断是否超时
if (continuation.isExpired())
{
// 返回超时Response
response.getWriter().println("timeout");
response.getWriter().flush();  
return;
}
// 挂起HTTP连接
continuation.suspend();
// 注册一个异步事件处理器
myAsyncHandler.register(new MyHandler() {  
public void onMyEvent(Object result) {  
continuation.setAttribute("results", result);  
continuation.resume();  
}  
});
return; // or continuation.undispatch();
}
// 连接恢复后返回结果
response.getWriter().println("Request: "+reqId+"\tend:\t" + results);
response.getWriter().flush();  
}

private interface MyAsyncHandler{
public void register(MyHandler myHandler);
}
private interface MyHandler{
public void onMyEvent(Object result);  
}
}
　　一旦continuation被挂起，需要注册一系列回调函数，请求将被挂起一直到continuation.resume()或continuation.complete()被调用，如果都没有被调用，则超时。
　　运行服务，测试结果如下：

root $ for i in `seq 1 10`; do lynx -dump http://192.168.201.167:1987/block?st=$i &  done
Request: 1      start:  2012/12/19 10:41:59
Request: 1      end:    2012/12/19 10:42:00
Request: 4      start:  2012/12/19 10:41:59
Request: 4      end:    2012/12/19 10:42:03
Request: 9      start:  2012/12/19 10:41:59
Request: 9      end:    2012/12/19 10:42:08
Request: 2      start:  2012/12/19 10:42:08
Request: 10     start:  2012/12/19 10:42:00
Request: 2      end:    2012/12/19 10:42:10
Request: 10     end:    2012/12/19 10:42:10

Request: 8      start:  2012/12/19 10:42:03
Request: 8      end:    2012/12/19 10:42:11
Request: 3      start:  2012/12/19 10:42:10
Request: 3      end:    2012/12/19 10:42:13
Request: 5      start:  2012/12/19 10:42:11
Request: 5      end:    2012/12/19 10:42:16
Request: 7      start:  2012/12/19 10:42:10
Request: 7      end:    2012/12/19 10:42:17
Request: 6      start:  2012/12/19 10:42:13
Request: 6      end:    2012/12/19 10:42:19
root $ for i in `seq 1 10`; do lynx -dump http://192.168.201.167:1987/nonblock?st=$i &  done
Request: 1      start:  2012/12/19 11:43:48
Request: 1      end:    2012/12/19 11:43:49
Request: 2      start:  2012/12/19 11:43:48
Request: 2      end:    2012/12/19 11:43:50
Request: 3      start:  2012/12/19 11:43:48
Request: 3      end:    2012/12/19 11:43:51
Request: 4      start:  2012/12/19 11:43:48
Request: 4      end:    2012/12/19 11:43:52
Request: 5      start:  2012/12/19 11:43:48
Request: 5      end:    2012/12/19 11:43:53
Request: 6      start:  2012/12/19 11:43:48
Request: 6      end:    2012/12/19 11:43:54
Request: 7      start:  2012/12/19 11:43:48
Request: 7      end:    2012/12/19 11:43:55
Request: 9      start:  2012/12/19 11:43:48
Request: 10     start:  2012/12/19 11:43:48
timeout
timeout
Request: 8      start:  2012/12/19 11:43:56
timeout
　　再用ab简单测试一下，nonblock模式rps(request per second)要高好几倍的样子吧
　　附：当异步handler操作结束返回结果时servlet或filter用来生成响应内容，典型地做法是通过设定request的attribute来传递结果results，用来判断请求是否被挂起。而suspend/continue模式用在异步的handler产生响应时使用。
　　continuation的suspend/continue模式实现如下：

package test;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.eclipse.jetty.continuation.Continuation;
import org.eclipse.jetty.continuation.ContinuationSupport;
public class NonBlockingServlet extends HttpServlet {
/**
* Suspend/Resume
*/
private static final long serialVersionUID = 3313258432391586994L;
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");//可以方便地修改日期格式
MyAsyncHandler myAsyncHandler ;
public void init() throws ServletException {
myAsyncHandler = new MyAsyncHandler() {  
public void register(final MyHandler myHandler,final String sleeptime) {
new Thread(new Runnable() {  
public void run() {  
try {
if(sleeptime != null){
Thread.sleep(Integer.parseInt(sleeptime)*1000);
}
String end = dateFormat.format( new Date() );
try {
myHandler.onMyEvent(end);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}  
} catch (InterruptedException e) {   
e.printStackTrace();  
}
}  
}).start();  
}  
};
}
protected void doGet(HttpServletRequest request, final HttpServletResponse response) throws IOException
{
String start = dateFormat.format( new Date() );
final String sleeptime = request.getParameter("st");
final Continuation continuation = ContinuationSupport.getContinuation(request);
if(continuation.isInitial()){
response.setContentType("text/plain");
response.getWriter().println("Request: "+sleeptime+"\tstart:\t"+start);
response.getWriter().flush();
}
continuation.setTimeout(8000);
// if this is not a timeout
if (continuation.isExpired())
{
// 返回超时Response
response.getWriter().println("timeout");
response.getWriter().flush();  
return;
}
// 挂起请求
continuation.suspend(response); // response may be wrapped.
// register with async service.  The code here will depend on the
// the service used (see Jetty HttpClient for example)
myAsyncHandler.register(new MyHandler()
{
public void onMyEvent(Object result) throws IOException
{
// 连接恢复后返回结果
response.getWriter().println("Request: "+sleeptime+"\tend:\t" + result);
response.getWriter().flush();
continuation.complete();
}
},sleeptime);
}

private interface MyAsyncHandler{
public void register(MyHandler myHandler,String sleeptime);
}
private interface MyHandler{
public void onMyEvent(Object result) throws IOException;  
}
}