From 0530c6b61b22dea0595e33eb7f60668a1186ed01 Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Wed, 18 Apr 2012 02:15:40 -0400 Subject: [PATCH] Add Spring MVC 3.2 Servlet based async support Instead of polling at regular intervals for quotes, the server now holds up the response until a new quote is available, which results in fewer requests overall and more optimal latency. Similarly instead of returning a trade confirmation id and then polling to obtain the confirmation, the server now holds up the request and until the confirmation becomes available. Since the request processing is released immediately, the longer wait should not affect the ability of the server to process requests while also eliminating the need for polling --- stocks/README.md | 3 + stocks/pom.xml | 27 ++++++-- .../stubs/ExecutionVenueServiceStub.java | 20 +++--- .../rabbit/stocks/web/QuoteController.java | 66 ++++++++++++++----- stocks/src/main/resources/log4j.properties | 2 +- stocks/src/main/resources/servlet-config.xml | 10 ++- stocks/src/main/webapp/WEB-INF/web.xml | 15 +++-- stocks/src/main/webapp/index.jsp | 63 +++++------------- 8 files changed, 119 insertions(+), 87 deletions(-) create mode 100644 stocks/README.md diff --git a/stocks/README.md b/stocks/README.md new file mode 100644 index 0000000..0c33024 --- /dev/null +++ b/stocks/README.md @@ -0,0 +1,3 @@ +This branch contains a version of the Spring AMQP stocks sample modified to take advantage of Spring MVC 3.2, Servlet-based async support. The change shows how an existing application with client-side polling can improve its latency while also optimizing the number of requests required to deliver updates. + +The details of the changes can be viewed in the following [commit](https://github.com/SpringSource/spring-amqp-samples/commit/1a241f8cd68835fa9e6af4e987bccf7b9e6b8bf1). diff --git a/stocks/pom.xml b/stocks/pom.xml index 3b7fcb5..a5ed4a9 100644 --- a/stocks/pom.xml +++ b/stocks/pom.xml @@ -15,7 +15,7 @@ true - 3.0.5.RELEASE + 3.2.0.BUILD-SNAPSHOT 1.0.0.RELEASE UTF-8 @@ -34,11 +34,23 @@ org.springframework.amqp spring-amqp ${spring.amqp.version} + + + org.springframework + spring-core + + org.springframework.amqp spring-rabbit ${spring.amqp.version} + + + org.springframework + spring-tx + + @@ -69,9 +81,9 @@ ${spring.framework.version} - javax.servlet - servlet-api - 2.5 + org.apache.geronimo.specs + geronimo-servlet_3.0_spec + 1.0 provided @@ -132,6 +144,13 @@ + + + SpringSource repository + http://repo.springsource.org/milestone + + + diff --git a/stocks/src/main/java/org/springframework/amqp/rabbit/stocks/service/stubs/ExecutionVenueServiceStub.java b/stocks/src/main/java/org/springframework/amqp/rabbit/stocks/service/stubs/ExecutionVenueServiceStub.java index ce97611..ab6fa04 100644 --- a/stocks/src/main/java/org/springframework/amqp/rabbit/stocks/service/stubs/ExecutionVenueServiceStub.java +++ b/stocks/src/main/java/org/springframework/amqp/rabbit/stocks/service/stubs/ExecutionVenueServiceStub.java @@ -27,16 +27,16 @@ import org.springframework.amqp.rabbit.stocks.service.ExecutionVenueService; /** * Execute the trade, setting the execution price to changing value in line with what the market data feed is producing. - * + * * @author Mark Pollack * */ public class ExecutionVenueServiceStub implements ExecutionVenueService { private static Log log = LogFactory.getLog(ExecutionVenueServiceStub.class); - + private Random random = new Random(); - + public TradeResponse executeTradeRequest(TradeRequest request) { TradeResponse response = new TradeResponse(); response.setAccountName(request.getAccountName()); @@ -46,11 +46,11 @@ public class ExecutionVenueServiceStub implements ExecutionVenueService { response.setTicker(request.getTicker()); response.setRequestId(request.getId()); response.setConfirmationNumber(UUID.randomUUID().toString()); - - + + try { - log.info("Sleeping 2 seconds to simulate processing.."); - Thread.sleep(2000); + log.info("Sleeping 5 seconds to simulate processing.."); + Thread.sleep(5000); } catch (InterruptedException e) { log.error("Didn't finish sleeping", e); } @@ -68,11 +68,11 @@ public class ExecutionVenueServiceStub implements ExecutionVenueService { { //in line with market data implementation return new BigDecimal(22 + Math.abs(gaussian())); - } + } } - + private double gaussian() { return random.nextGaussian(); - } + } } diff --git a/stocks/src/main/java/org/springframework/amqp/rabbit/stocks/web/QuoteController.java b/stocks/src/main/java/org/springframework/amqp/rabbit/stocks/web/QuoteController.java index 397f12d..31b0de4 100644 --- a/stocks/src/main/java/org/springframework/amqp/rabbit/stocks/web/QuoteController.java +++ b/stocks/src/main/java/org/springframework/amqp/rabbit/stocks/web/QuoteController.java @@ -1,11 +1,11 @@ /* * Copyright 2002-2010 the original author or authors. - * + * * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at - * + * * http://www.apache.org/licenses/LICENSE-2.0 - * + * * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the * specific language governing permissions and limitations under the License. @@ -18,6 +18,8 @@ import java.util.Collections; import java.util.Comparator; import java.util.Iterator; import java.util.List; +import java.util.Map; +import java.util.Map.Entry; import java.util.Queue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -36,10 +38,11 @@ import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMethod; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.ResponseBody; +import org.springframework.web.context.request.async.DeferredResult; /** * @author Dave Syer - * + * @author Rossen Stoyanchev */ @Controller public class QuoteController { @@ -52,6 +55,10 @@ public class QuoteController { private Queue quotes = new PriorityBlockingQueue(100, new QuoteComparator()); + private Map suspendedTradeRequests = new ConcurrentHashMap(); + + private Map suspendedQuoteRequests = new ConcurrentHashMap(); + private long timeout = 30000; // 30 seconds of data public void setStockServiceGateway(StockServiceGateway stockServiceGateway) { @@ -63,17 +70,34 @@ public class QuoteController { String key = response.getRequestId(); responses.putIfAbsent(key, response); Collection queue = new ArrayList(responses.values()); + long timestamp = System.currentTimeMillis() - timeout; for (Iterator iterator = queue.iterator(); iterator.hasNext();) { TradeResponse tradeResponse = iterator.next(); + String requestId = tradeResponse.getRequestId(); if (tradeResponse.getTimestamp() < timestamp) { - responses.remove(tradeResponse.getRequestId()); + responses.remove(requestId); + } + if (suspendedTradeRequests.containsKey(requestId)) { + DeferredResult deferredResult = suspendedTradeRequests.remove(requestId); + deferredResult.trySet(tradeResponse); } } } public void handleQuote(Quote message) { logger.info("Client received: " + message); + quotes.add(message); + + for (Entry entry : suspendedQuoteRequests.entrySet()) { + List list = getLatestQuotes(entry.getValue()); + if (!list.isEmpty()) { + DeferredResult deferredResult = entry.getKey(); + deferredResult.trySet(list); + suspendedQuoteRequests.remove(entry.getKey()); + } + } + long timestamp = System.currentTimeMillis() - timeout; for (Iterator iterator = quotes.iterator(); iterator.hasNext();) { Quote quote = iterator.next(); @@ -81,12 +105,23 @@ public class QuoteController { iterator.remove(); } } - quotes.add(message); } @RequestMapping("/quotes") @ResponseBody - public List quotes(@RequestParam(required = false) Long timestamp) { + public Object quotes(@RequestParam(required = false) Long timestamp) { + List list = getLatestQuotes(timestamp); + if (list.isEmpty()) { + DeferredResult deferredResult = new DeferredResult(Collections.emptyList()); + suspendedQuoteRequests.put(deferredResult, timestamp); + return deferredResult; + } + else { + return list; + } + } + + private List getLatestQuotes(Long timestamp) { if (timestamp == null) { timestamp = 0L; } @@ -102,13 +137,16 @@ public class QuoteController { @RequestMapping(value = "/trade", method = RequestMethod.POST) @ResponseBody - public TradeRequest trade(@ModelAttribute TradeRequest tradeRequest) { + public Object trade(@ModelAttribute TradeRequest tradeRequest) { String ticker = tradeRequest.getTicker(); Long quantity = tradeRequest.getQuantity(); if (quantity == null || quantity <= 0 || !StringUtils.hasText(ticker)) { // error - return tradeRequest; + return null; } else { + DeferredResult deferredResult = new DeferredResult(); + suspendedTradeRequests.put(tradeRequest.getId(), deferredResult); + // Fake rest of request while UI is basic tradeRequest.setAccountName("ACCT-123"); tradeRequest.setBuyRequest(true); @@ -117,15 +155,9 @@ public class QuoteController { tradeRequest.setUserName("Joe Trader"); tradeRequest.setUserName("Joe"); stockServiceGateway.send(tradeRequest); - } - return tradeRequest; - } - @RequestMapping(value = "/trade", method = RequestMethod.GET) - @ResponseBody - public TradeResponse response(@RequestParam String requestId) { - TradeResponse result = responses.get(requestId); - return result; + return deferredResult; + } } private static class QuoteComparator implements Comparator { diff --git a/stocks/src/main/resources/log4j.properties b/stocks/src/main/resources/log4j.properties index 1343d41..b19a9c2 100644 --- a/stocks/src/main/resources/log4j.properties +++ b/stocks/src/main/resources/log4j.properties @@ -6,6 +6,6 @@ log4j.appender.stdout.layout=org.apache.log4j.PatternLayout #log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - <%m>%n log4j.appender.stdout.layout.ConversionPattern=%-5p %8t [%40.40c{4}]: %m%n -log4j.category.org.springframework.amqp.rabbit=DEBUG +log4j.category.org.springframework.amqp.rabbit=INFO log4j.category.org.springframework.beans.factory=INFO diff --git a/stocks/src/main/resources/servlet-config.xml b/stocks/src/main/resources/servlet-config.xml index 3b8bec5..7562333 100644 --- a/stocks/src/main/resources/servlet-config.xml +++ b/stocks/src/main/resources/servlet-config.xml @@ -5,7 +5,7 @@ xsi:schemaLocation="http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc-3.0.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd - http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd"> + http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd"> @@ -53,4 +53,12 @@ + + + + + + + + diff --git a/stocks/src/main/webapp/WEB-INF/web.xml b/stocks/src/main/webapp/WEB-INF/web.xml index fda42b6..e44b3aa 100644 --- a/stocks/src/main/webapp/WEB-INF/web.xml +++ b/stocks/src/main/webapp/WEB-INF/web.xml @@ -1,10 +1,8 @@ - - - + + contextConfigLocation classpath*:/server-bootstrap-*.xml @@ -17,11 +15,13 @@ shallowEtagHeaderFilter org.springframework.web.filter.ShallowEtagHeaderFilter + true hiddenHttpMethodFilter org.springframework.web.filter.HiddenHttpMethodFilter + true @@ -42,6 +42,7 @@ classpath*:/servlet-config.xml 1 + true diff --git a/stocks/src/main/webapp/index.jsp b/stocks/src/main/webapp/index.jsp index bc03f22..ce71bb8 100644 --- a/stocks/src/main/webapp/index.jsp +++ b/stocks/src/main/webapp/index.jsp @@ -14,25 +14,18 @@ var timer; var debug = false; var lastquote = 0; - var lasttrades = {}; var template = "{{#quotes}}\ {{timeString}}\ {{#stock}}{{ticker}}{{/stock}}\ {{price}}\ {{/quotes}}"; - var confirmation = "{{#response}}Trade Confirmation:
    \ -
  • Id: {{confirmationNumber}}
    • \ -
  • Quantity: {{quantity}}
    • \ -
  • Ticker: {{ticker}}
    • \ -
  • Price: {{price}}
    • \ -
{{/response}}"; + var confirmation = "{{#response}}Trade confirmation {{confirmationNumber}}, Quantity: {{quantity}}, Ticker: {{ticker}}, Price: {{price}}{{/response}}
"; function load() { if (running) { - $('#status').text("Waiting...") + $('#status').text("Getting quotes...") $.ajax({ url : "quotes?timestamp=" + lastquote, success : function(message) { - $('#status').text("Updating") if (debug) { $('#debug').text(JSON.stringify(message)) } @@ -79,46 +72,23 @@ } return setTimeout(load, 1000); } - function confirm(id) { - if (lasttrades.id) { - clearTimeout(lasttrades.id); - delete lasttrades.id; - } - $.get("trade?requestId=" + id, function(response) { - if (response && response.requestId) { - $('#messages').html($.mustache(confirmation, { - response : response - })); - delete lasttrades.id; - } else { - lasttrades.id = setTimeout("confirm('" + id + "')", 2000); - } - }); - } $(function() { $.ajaxSetup({cache:false}); $('#start').click(start); $('#stop').click(stop); $('#clear').click(clear); - start(); - $('#tradeForm') - .submit( - function() { - $ - .post( - $('#tradeForm').attr("action"), - $('#tradeForm').serialize(), - function(request) { - var message = "Processing..."; - if (request && request.ticker) { - confirm(request.id); - } else { - message = "The trade request was invalid. Please provide a quantity and a stock ticker."; - } - $('#messages').text(message); - }); - return false; - }); + $('#tradeForm').submit(function() { + $('#messages').text(''); + $.post($('#tradeForm').attr("action"), $('#tradeForm').serialize(), + function(response) { + if (response && response.requestId) { + $('#confirmations').append($.mustache(confirmation, {response : response})); + } else { + $('#messages').html("

The trade request was invalid. Please provide a quantity and a stock ticker.

"); + } + }); + return false; + }); }); @@ -156,6 +126,7 @@ +
-
- -
+

Quotes

Stopped