Servlet-3 异步上下文，如何执行异步写入？
2022-08-31 20:38:21
问题描述
Servlet-3.0 API 允许分离请求/响应上下文并在以后回答它。
但是，如果我试图编写大量数据，例如：
AsyncContext ac = getWaitingContext() ;
ServletOutputStream out = ac.getResponse().getOutputStream();
out.print(some_big_data);
out.flush()
它实际上可能会阻止 - 并且它确实在琐碎的测试用例中阻止 - 对于Tomcat 7和Jetty 8。教程建议创建一个线程池来处理这样的设置 - 女巫通常是传统10K架构的反正。
但是，如果我有10，000个打开的连接和一个10个线程的线程池，那么即使有1%的客户端具有低速连接或只是阻塞连接，也足以阻止线程池并完全阻止彗星响应或显着减慢它。
预期的做法是获取“写就绪”通知或 I/O 完成通知，然后继续推送数据。
如何使用Servlet-3.0 API完成此操作，即我如何获得：
- 有关 I/O 操作的异步完成通知。
- 通过写入就绪通知获取非阻塞 I/O。
如果Servlet-3.0 API不支持这一点，那么是否有任何特定于Web服务器的API（如Jetty Continuation或Tomcat CometEvent）允许真正异步处理此类事件，而无需使用线程池伪造异步I / O。
有人知道吗？
如果这是不可能的，你能通过参考文档来确认它吗？
示例代码中的问题演示
我附加了下面模拟事件流的代码。
笔记：
- 它使用抛出来检测断开连接的客户端
ServletOutputStreamIOException - 它发送消息以确保客户端仍然存在
keep-alive - 我创建了一个线程池来“模拟”异步操作。
在这样一个示例中，我显式定义了大小为 1 的线程池来显示问题：
- 启动应用程序
- 从两个终端运行（两次）
curl http://localhost:8080/path/to/app - 现在发送数据
curd -d m=message http://localhost:8080/path/to/app - 两个客户端都收到了数据
- 现在挂起其中一个客户端 （Ctrl+Z） 并再次发送消息
curd -d m=message http://localhost:8080/path/to/app - 观察到另一个未挂起的客户端要么未收到任何内容，要么在消息传输后停止接收保持活动状态请求，因为其他线程被阻止。
我想在不使用线程池的情况下解决这样的问题，因为使用1000-5000个打开的连接，我可以非常快速地耗尽线程池。
下面的示例代码。
import java.io.IOException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.LinkedBlockingQueue;
import javax.servlet.AsyncContext;
import javax.servlet.ServletConfig;
import javax.servlet.ServletException;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import javax.servlet.ServletOutputStream;
@WebServlet(urlPatterns = "", asyncSupported = true)
public class HugeStreamWithThreads extends HttpServlet {
private long id = 0;
private String message = "";
private final ThreadPoolExecutor pool =
new ThreadPoolExecutor(1, 1, 50000L,TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());
// it is explicitly small for demonstration purpose
private final Thread timer = new Thread(new Runnable() {
public void run()
{
try {
while(true) {
Thread.sleep(1000);
sendKeepAlive();
}
}
catch(InterruptedException e) {
// exit
}
}
});
class RunJob implements Runnable {
volatile long lastUpdate = System.nanoTime();
long id = 0;
AsyncContext ac;
RunJob(AsyncContext ac)
{
this.ac = ac;
}
public void keepAlive()
{
if(System.nanoTime() - lastUpdate > 1000000000L)
pool.submit(this);
}
String formatMessage(String msg)
{
StringBuilder sb = new StringBuilder();
sb.append("id");
sb.append(id);
for(int i=0;i<100000;i++) {
sb.append("data:");
sb.append(msg);
sb.append("\n");
}
sb.append("\n");
return sb.toString();
}
public void run()
{
String message = null;
synchronized(HugeStreamWithThreads.this) {
if(this.id != HugeStreamWithThreads.this.id) {
this.id = HugeStreamWithThreads.this.id;
message = HugeStreamWithThreads.this.message;
}
}
if(message == null)
message = ":keep-alive\n\n";
else
message = formatMessage(message);
if(!sendMessage(message))
return;
boolean once_again = false;
synchronized(HugeStreamWithThreads.this) {
if(this.id != HugeStreamWithThreads.this.id)
once_again = true;
}
if(once_again)
pool.submit(this);
}
boolean sendMessage(String message)
{
try {
ServletOutputStream out = ac.getResponse().getOutputStream();
out.print(message);
out.flush();
lastUpdate = System.nanoTime();
return true;
}
catch(IOException e) {
ac.complete();
removeContext(this);
return false;
}
}
};
private HashSet<RunJob> asyncContexts = new HashSet<RunJob>();
@Override
public void init(ServletConfig config) throws ServletException
{
super.init(config);
timer.start();
}
@Override
public void destroy()
{
for(;;){
try {
timer.interrupt();
timer.join();
break;
}
catch(InterruptedException e) {
continue;
}
}
pool.shutdown();
super.destroy();
}
protected synchronized void removeContext(RunJob ac)
{
asyncContexts.remove(ac);
}
// GET method is used to establish a stream connection
@Override
protected synchronized void doGet(HttpServletRequest request, HttpServletResponse response)
throws ServletException, IOException {
// Content-Type header
response.setContentType("text/event-stream");
response.setCharacterEncoding("utf-8");
// Access-Control-Allow-Origin header
response.setHeader("Access-Control-Allow-Origin", "*");
final AsyncContext ac = request.startAsync();
ac.setTimeout(0);
RunJob job = new RunJob(ac);
asyncContexts.add(job);
if(id!=0) {
pool.submit(job);
}
}
private synchronized void sendKeepAlive()
{
for(RunJob job : asyncContexts) {
job.keepAlive();
}
}
// POST method is used to communicate with the server
@Override
protected synchronized void doPost(HttpServletRequest request, HttpServletResponse response)
throws ServletException, IOException
{
request.setCharacterEncoding("utf-8");
id++;
message = request.getParameter("m");
for(RunJob job : asyncContexts) {
pool.submit(job);
}
}
}
上面的示例使用线程来防止阻塞...但是，如果阻塞客户端的数量大于线程池的大小，它将阻塞。
如何在不阻塞的情况下实施它？
