【转】使用 Java 实现 Comet 风格的 Web 应用

<!-- This is the usual Connector, comment it out and add the NIO one -->
   <!-- Connector URIEncoding="utf-8" connectionTimeout="20000" port="8084" 
protocol="HTTP/1.1" redirectPort="8443"/ -->
<Connector connectionTimeout="20000" port="8080" protocol="org.apache.
coyote.http11.Http11NioProtocol" redirectPort="8443"/>

public class TomcatWeatherServlet extends HttpServlet implements CometProcessor {

    private MessageSender messageSender = null;
    private static final Integer TIMEOUT = 60 * 1000;

    @Override
    public void destroy() {
        messageSender.stop();
        messageSender = null;

    }

    @Override
    public void init() throws ServletException {
        messageSender = new MessageSender();
        Thread messageSenderThread =
                new Thread(messageSender, "MessageSender[" + getServletContext()
.getContextPath() + "]");
        messageSenderThread.setDaemon(true);
        messageSenderThread.start();

    }

    public void event(final CometEvent event) throws IOException, ServletException {
        HttpServletRequest request = event.getHttpServletRequest();
        HttpServletResponse response = event.getHttpServletResponse();
        if (event.getEventType() == CometEvent.EventType.BEGIN) {
            request.setAttribute("org.apache.tomcat.comet.timeout", TIMEOUT);
            log("Begin for session: " + request.getSession(true).getId());
            messageSender.setConnection(response);
            Weatherman weatherman = new Weatherman(95118, 32408);
            new Thread(weatherman).start();
        } else if (event.getEventType() == CometEvent.EventType.ERROR) {
            log("Error for session: " + request.getSession(true).getId());
            event.close();
        } else if (event.getEventType() == CometEvent.EventType.END) {
            log("End for session: " + request.getSession(true).getId());
            event.close();
        } else if (event.getEventType() == CometEvent.EventType.READ) {
            throw new UnsupportedOperationException("This servlet does not accept 
data");
        }

    }
}

private class MessageSender implements Runnable {

    protected boolean running = true;
    protected final ArrayList<String> messages = new ArrayList<String>();
    private ServletResponse connection;
    private synchronized void setConnection(ServletResponse connection){
        this.connection = connection;
        notify();
    }
    public void send(String message) {
        synchronized (messages) {
            messages.add(message);
            log("Message added #messages=" + messages.size());
            messages.notify();
        }
    }
    public void run() {
        while (running) {
            if (messages.size() == 0) {
                try {
                    synchronized (messages) {
                        messages.wait();
                    }
                } catch (InterruptedException e) {
                    // Ignore
                }
            }
            String[] pendingMessages = null;
            synchronized (messages) {
                pendingMessages = messages.toArray(new String[0]);
                messages.clear();
            }
            try {
                if (connection == null){
                    try{
                        synchronized(this){
                            wait();
                        }
                    } catch (InterruptedException e){
                        // Ignore
                    }
                }
                PrintWriter writer = connection.getWriter();
                for (int j = 0; j < pendingMessages.length; j++) {
                    final String forecast = pendingMessages[j] + "<br>";
                    writer.println(forecast);
                    log("Writing:" + forecast);
                }
                writer.flush();
                writer.close();
                connection = null;
                log("Closing connection");
            } catch (IOException e) {
                log("IOExeption sending message", e);
            }
        }
    }
}

private class Weatherman implements Runnable{

    private final List<URL> zipCodes;
    private final String YAHOO_WEATHER = "http://weather.yahooapis.com/forecastrss?p=";

    public Weatherman(Integer... zips) {
        zipCodes = new ArrayList<URL>(zips.length);
        for (Integer zip : zips) {
            try {
                zipCodes.add(new URL(YAHOO_WEATHER + zip));
            } catch (Exception e) {
                // dont add it if it sucks
            }
        }
    }

   public void run() {
       int i = 0;
       while (i >= 0) {
           int j = i % zipCodes.size();
           SyndFeedInput input = new SyndFeedInput();
           try {
               SyndFeed feed = input.build(new InputStreamReader(zipCodes.get(j)
.openStream()));
               SyndEntry entry = (SyndEntry) feed.getEntries().get(0);
               messageSender.send(entryToHtml(entry));
               Thread.sleep(30000L);
           } catch (Exception e) {
               // just eat it, eat it
           }
           i++;
       }
   }

    private String entryToHtml(SyndEntry entry){
        StringBuilder html = new StringBuilder("<h2>");
        html.append(entry.getTitle());
        html.append("</h2>");
        html.append(entry.getDescription().getValue());
        return html.toString();
    }
}

<%@page contentType="text/html" pageEncoding="UTF-8"%>
<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN"
   "http://www.w3.org/TR/html4/loose.dtd">

<html>
    <head>
        <meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
        <title>Comet Weather</title>
        <SCRIPT TYPE="text/javascript">
            function go(){
                var url = "http://localhost:8484/WeatherServer/Weather"
                var request =  new XMLHttpRequest();
                request.open("GET", url, true);
                request.setRequestHeader("Content-Type","application/x-javascript;");
                request.onreadystatechange = function() {
                    if (request.readyState == 4) {
                        if (request.status == 200){
                            if (request.responseText) {
                                document.getElementById("forecasts").innerHTML = 
request.responseText;
                            }
                        }
                        go();
                    }
                };
                request.send(null);
            }
        </SCRIPT>
    </head>
    <body>
        <h1>Rapid Fire Weather</h1>
        <input type="button" onclick="go()" value="Go!"></input>
        <div id="forecasts"></div>
    </body>
</html>

public class JettyWeatherServlet extends HttpServlet {
    private MessageSender messageSender = null;
    private static final Integer TIMEOUT = 5 * 1000;
    public void begin(HttpServletRequest request, HttpServletResponse response) 
throws IOException, ServletException {
        request.setAttribute("org.apache.tomcat.comet", Boolean.TRUE);
        request.setAttribute("org.apache.tomcat.comet.timeout", TIMEOUT);
        messageSender.setConnection(response);
        Weatherman weatherman = new Weatherman(95118, 32408);
        new Thread(weatherman).start();
    }
    public void end(HttpServletRequest request, HttpServletResponse response) 
throws IOException, ServletException {
        synchronized (request) {
            request.removeAttribute("org.apache.tomcat.comet");
            Continuation continuation = ContinuationSupport.getContinuation
(request, request);
            if (continuation.isPending()) {
                continuation.resume();
            }
        }
    }
    public void error(HttpServletRequest request, HttpServletResponse response) 
throws IOException, ServletException {
        end(request, response);
    }
    public boolean read(HttpServletRequest request, HttpServletResponse response) 
throws IOException, ServletException {
        throw new UnsupportedOperationException();
    }
    @Override
    protected void service(HttpServletRequest request, HttpServletResponse response) 
throws IOException, ServletException {
        synchronized (request) {
            Continuation continuation = ContinuationSupport.getContinuation
(request, request);
            if (!continuation.isPending()) {
                begin(request, response);
            }
            Integer timeout = (Integer) request.getAttribute
("org.apache.tomcat.comet.timeout");
            boolean resumed = continuation.suspend(timeout == null ? 10000 : 
timeout.intValue());

            if (!resumed) {
                error(request, response);
            }
        }
    }
    public void setTimeout(HttpServletRequest request, HttpServletResponse response, 
int timeout) throws IOException, ServletException,
            UnsupportedOperationException {
        request.setAttribute("org.apache.tomcat.comet.timeout", new Integer(timeout));
    }
}

@WebServlet(asyncSupported=true, asyncTimeout=5000)
public class WeatherServlet extends HttpServlet {
   private MessageSender messageSender;

// init and destroy are the same as other
    @Override
    protected void doGet(HttpServletRequest request, HttpServletResponse response)
    throws ServletException, IOException {
        AsyncContext async = request.startAsync(request, response);
        messageSender.setConnection(async);
        Weatherman weatherman = new Weatherman(95118, 32444);
        async.start(weatherman);;
    }
}