問題の説明
サーブレット-3.0APIを使用すると、要求/応答コンテキストを切り離して、後でそれに応答できます。
ただし、大量のデータを書き込もうとすると、次のようになります。
AsyncContext ac = getWaitingContext() ;
ServletOutputStream out = ac.getResponse().getOutputStream();
out.print(some_big_data);
out.flush()
Tomcat7とJetty8の両方で、実際にブロックされる可能性があります。また、簡単なテストケースではブロックされます。チュートリアルでは、このようなセットアップを処理するスレッドプールを作成することをお勧めします。
ただし、10,000のオープン接続と、たとえば10スレッドのスレッドプールがある場合、低速接続または接続をブロックしたばかりのクライアントの1％でも、スレッドプールをブロックして、彗星の応答を完全にブロックするか、速度を落とすだけで十分です。大幅。
予想される方法は、「書き込み可能」通知またはI / O完了通知を取得してから、データをプッシュし続けることです。
Servlet-3.0 APIを使用してこれを行うにはどうすればよいですか。つまり、次のいずれかを取得するにはどうすればよいですか。
- I/O操作に関する非同期完了通知。
- 書き込み準備完了通知でノンブロッキングI/Oを取得します。
これがServlet-3.0APIでサポートされていない場合、スレッドプールを使用して非同期I / Oを偽造することなく、このようなイベントを真に非同期で処理できるWebサーバー固有のAPI（JettyContinuationやTomcatCometEventなど）はありますか。
誰か知っていますか？
そして、これが不可能な場合は、ドキュメントを参照して確認できますか？
サンプルコードでの問題のデモンストレーション
イベントストリームをエミュレートする以下のコードを添付しました。
ノート：
- 切断されたクライアントを検出するために
ServletOutputStreamそのスローを使用しますIOException keep-aliveクライアントがまだそこにいることを確認するためにメッセージを送信します- 非同期操作を「エミュレート」するスレッドプールを作成しました。
このような例では、問題を示すためにサイズ1のスレッドプールを明示的に定義しました。
- アプリケーションを起動します
- 2つのターミナルから実行
curl http://localhost:8080/path/to/app（2回） - 次に、でデータを送信します
curd -d m=message http://localhost:8080/path/to/app - 両方のクライアントがデータを受信しました
- ここで、クライアントの1つを一時停止し（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);
}
}
}
上記のサンプルでは、スレッドを使用してブロックを防止しています...ただし、ブロックするクライアントの数がスレッドプールのサイズよりも大きい場合は、ブロックされます。
ブロックせずにどのように実装できますか？