Call async service which return DeferredResults, multiple times without increasing execution time
It seems the problem here is that you're running out of connections in the database pool.
You have your method tagged with @Transaction but your controller is also expecting the result of the method i.e. the DeferredResult to be delivered as soon as possible such that the thread is set free.
Now, this is what happens when you run a request:
- The
@Transactionfunctionality is implemented in a Spring proxy which must open a connection, call your subject method and then commit or rollback the transaction. - Therefore, when your controller invokes the
fooService.pullmethod, it is in fact calling a proxy. - The proxy must first request a connection from the pool, then it invokes your service method, which within that transaction does some database operation. Finally, it must commit or rollback the transaction and finally return the connection to the pool.
- After all this your method returns a
DeferredResult, which is then passed to the controller for it to return.
Now, the problem is that DeferredResult is designed in such a way that it should be used asynchronously. In other words, the promise is expected to be resolved later in some other thread, and we are supposed to free the request thread as soon as possible.
In fact, Spring documentation on DeferredResult says:
@GetMapping("/quotes")
@ResponseBody
public DeferredResult<String> quotes() {
DeferredResult<String> deferredResult = new DeferredResult<String>();
// Save the deferredResult somewhere..
return deferredResult;
}
// From some other thread...
deferredResult.setResult(data);
The problem in your code is precisely that the DeferredResult is being solved in the same request thread.
So, the thing is that when the Spring proxy requests a connection to the database pool, when you do your heavy load tests, many requests will find the pool is full and does not have connections available. So the request is put on hold, but at that point your DeferredResult has not been created yet, so its timeout functionality does not exist.
Your request is basically waiting for some connection from the database pool to become available. So, let's say 5 seconds pass, then the request gets a connection, and now you get DeferredResult which the controller uses to handle the response. Eventually, 5 seconds later it timeout. So you have to add your time waiting for a connection from the pool and your time waiting for the DeferredResult to get resolved.
That's why you probably see that, when you test with JMeter, the request time gradually increases as connections get depleted from the database pool.
You can enable some logging for the thread pool by adding the following your application.properties file:
logging.level.com.zaxxer.hikari=DEBUG
You could also configure the size of your database pool and even add some JMX support such that you can monitor it from Java Mission Control:
spring.datasource.hikari.maximumPoolSize=10
spring.datasource.hikari.registerMbeans=true
Using JMX support you will be able to see how the database pool gets depleted.
The trick here consists in moving the logic that resolves the promise to another thread:
@Override
public DeferredResult pull(Long previousId, String username) {
DeferredResult result = createPollingResult(previousId, username);
CompletableFuture.runAsync(() -> {
//this is where you encapsulate your db transaction
List<MessageDTO> messages = messageService.findRecents(previousId, username); // should be final or effective final
if (messages.isEmpty()) {
pollingResults.putIfAbsent(username, result);
} else {
result.setResult(messages);
}
});
return result;
}
By doing this, your DeferredResult is returned immediately and Spring can do its magic of asynchronous request handling while it sets free that precious Tomcat thread.
I think you need producer and consumer structure model. i write code for you. i hope it will help you.
This is sample code :
DeferredResultStrore
@Component
public class DeferredResultStrore {
private Queue<DeferredResult<String>> responseBodyQueue;
private HashMap<String, List<DeferredResult<InterfaceModel>>> groupMap;
private final long resultTimeOut;
public DeferredResultStrore() {
responseBodyQueue = new LinkedBlockingQueue<DeferredResult<String>>();
groupMap = new HashMap<String, List<DeferredResult<InterfaceModel>>>();
// write time.
resultTimeOut = 1000 * 60 * 60;
}
public Queue<DeferredResult<String>> getResponseBodyQueue() {
return responseBodyQueue;
}
public HashMap<String, List<DeferredResult<InterfaceModel>>> getGroupMap() {
return groupMap;
}
public long getResultTimeOut() {
return resultTimeOut;
}
}
DeferredResultService
public interface DeferredResultService {
public DeferredResult<?> biteResponse(HttpServletResponse resp, HttpServletRequest req);
public DeferredResult<?> biteGroupResponse(String key, HttpServletResponse resp);
}
DeferredResultServiceImpl
@Service
public class DeferredResultServiceImpl implements DeferredResultService {
@Autowired
private DeferredResultStrore deferredResultStore;
@Override
public DeferredResult<?> biteResponse(final HttpServletResponse resp, HttpServletRequest req) {
final DeferredResult<String> defResult = new DeferredResult<String>(deferredResultStore.getResultTimeOut());
removeObserver(resp, defResult, null);
deferredResultStore.getResponseBodyQueue().add(defResult);
return defResult;
}
@Override
public DeferredResult<?> biteGroupResponse(String key, final HttpServletResponse resp) {
final DeferredResult<InterfaceModel> defResult = new DeferredResult<InterfaceModel>(
deferredResultStore.getResultTimeOut());
List<DeferredResult<InterfaceModel>> defResultList = null;
removeObserver(resp, defResult, key);
if (deferredResultStore.getGroupMap().containsKey(key)) {
defResultList = deferredResultStore.getGroupMap().get(key);
defResultList.add(defResult);
} else {
defResultList = new ArrayList<DeferredResult<InterfaceModel>>();
defResultList.add(defResult);
deferredResultStore.getGroupMap().put(key, defResultList);
}
return defResult;
}
private void removeObserver(final HttpServletResponse resp, final DeferredResult<?> defResult, final String key) {
defResult.onCompletion(new Runnable() {
public void run() {
if (key != null) {
List<DeferredResult<InterfaceModel>> defResultList = deferredResultStore.getGroupMap().get(key);
if (defResultList != null) {
for (DeferredResult<InterfaceModel> deferredResult : defResultList) {
if (deferredResult.hashCode() == defResult.hashCode()) {
defResultList.remove(deferredResult);
}
}
}
} else {
if (!deferredResultStore.getResponseBodyQueue().isEmpty()) {
deferredResultStore.getResponseBodyQueue().remove(defResult);
}
}
}
});
defResult.onTimeout(new Runnable() {
public void run() {
// 206
resp.setStatus(HttpServletResponse.SC_PARTIAL_CONTENT);
if (key != null) {
List<DeferredResult<InterfaceModel>> defResultList = deferredResultStore.getGroupMap().get(key);
if (defResultList != null) {
for (DeferredResult<InterfaceModel> deferredResult : defResultList) {
if (deferredResult.hashCode() == defResult.hashCode()) {
InterfaceModel model = new InterfaceModel();
model.setId(key);
model.setMessage("onTimeout");
deferredResult.setErrorResult(model);
defResultList.remove(deferredResult);
}
}
}
} else {
defResult.setErrorResult("onTimeout");
deferredResultStore.getResponseBodyQueue().remove(defResult);
}
}
});
}
}
PushService
public interface PushService {
public boolean pushMessage(String message);
public boolean pushGroupMessage(String key, String topic, HttpServletResponse resp);
}
PushServiceImpl
@Service
public class PushServiceImpl implements PushService {
@Autowired
private DeferredResultStrore deferredResultStore;
@Override
public boolean pushMessage(String message) {
if (!deferredResultStore.getResponseBodyQueue().isEmpty()) {
for (DeferredResult<String> deferredResult : deferredResultStore.getResponseBodyQueue()) {
deferredResult.setResult(message);
}
deferredResultStore.getResponseBodyQueue().remove();
}
return true;
}
@Override
public boolean pushGroupMessage(String key, String topic, HttpServletResponse resp) {
List<DeferredResult<InterfaceModel>> defResultList = null;
// select data in DB. that is sample group push service. need to connect db.
InterfaceModel model = new InterfaceModel();
model.setMessage("write group message.");
model.setId(key);
if (deferredResultStore.getGroupMap().containsKey(key)) {
defResultList = deferredResultStore.getGroupMap().get(key);
for (DeferredResult<InterfaceModel> deferredResult : defResultList) {
deferredResult.setResult(model);
}
deferredResultStore.getGroupMap().remove(key);
}
return true;
}
}
InterfaceModel
public class InterfaceModel {
private String message;
private int idx;
private String id;
// DB Column
public InterfaceModel() {
// TODO Auto-generated constructor stub
}
public InterfaceModel(String message, int idx, String id) {
this.message = message;
this.idx = idx;
this.id = id;
}
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
public int getIdx() {
return idx;
}
public void setIdx(int idx) {
this.idx = idx;
}
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
}
web.xml
async-supported very important in settings.
<servlet>
<servlet-name>appServlet</servlet-name>
<servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
<init-param>
<param-name>contextConfigLocation</param-name>
<param-value>/WEB-INF/spring/appServlet/servlet-context.xml</param-value>
</init-param>
<load-on-startup>1</load-on-startup>
<async-supported>true</async-supported>
</servlet>
Java base
@Bean
public ServletRegistrationBean dispatcherServlet() {
ServletRegistrationBean registration = new ServletRegistrationBean(
new DispatcherServlet(), "/");
registration.setAsyncSupported(true);
return registration;
}
In fact :
A DeferredResult is associated with an open request. When the request completes, the DeferredResult is removed from the map, and then, the client issues a new long polling request, which adds a new DeferredResult instance
Spring Boot will automatically register any Servlet beans in your application context with the servlet container. By default async supported is set to true so there's nothing for you to do beyond creating a bean for your Servlet.
@Aligtor, for you => public @interface EnableAsync Enables Spring's asynchronous method execution capability, similar to functionality found in Spring's XML namespace.
As many guys mentioned it's not correct way to test the performance. You asked to do automated requests at certain period of time as you were doing in XMLHttpRequest. You can use interval of Observable as:
import {Observable} from "rxjs/Observable";
import {Subscription} from "rxjs/Subscription";
private _intervalSubscription: Subscription;
ngOnInit() {
this._intervalSubscription = Observable.interval(500).subscribe(x => {
this.getDataFromServer();
});
}
ngOnDestroy(): void {
this._intervalSubscription.unsubscribe();
}
getDataFromServer() {
// do your network call
this.http.get<any>(this.url, {params})
.subscribe((data) => {
console.log('s', data);
}, (error) => {
console.log('e', error);
});
}
This is the best possible way of doing polling from client side.
EDIT 1
private prevRequestTime: number;
ngAfterViewInit(): void {
this.getDataFromServer();
}
getDataFromServer() {
this.prevRequestTime = Date.now();
// do your network call
this.http.get<any>(this.url, {params})
.subscribe((data) => {
console.log('s', data);
this.scheduleRequestAgain();
}, (error) => {
console.log('e', error);
this.scheduleRequestAgain();
});
}
scheduleRequestAgain() {
let diff = Date.now() - this.prevRequestTime;
setTimeout(this.getDataFromServer(), diff);
}