背景：
业务类型是类似于问答大模型那种可以流式蹦字的一个ui展示效果（类似于手机上的语音助手这种产品），业务中用的是http请求长连接，拿到下游服务的reponse后需要不断地读取其中的内容，下游不断地向其中去写。但是存在一种场景就是用户在蹦字的过程中退出应用。这种情况下我们需要回收应用的资源并将其正确关闭。但是在观测打点和回流数据时，很多正常业务的场景下野打了告警日志。下面是业务简单的代码：
class Task implements Runnable {
@Override
public void run() {
Response response = null;
try {
Response response = new OkHttpClient().newCall(new Request());
InputStream inputStream = response.body().byteStream();
byte[] bytes = new byte[1024];
int readLen = 0;
while (readLen = inputStream.read(bytes)) {
// do something
}
} catch (Exception ex) {
// 打印告警日志
} finally {
if (response != null) {
response.close();
}
}
}
}
class ProjectHolder{
private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4, 5, TimeUnit.MINUTES, new ArrayBlockingQueue<>(2));
public static String status = "running";
public static void run(Task task) {
threadPoolExecutor.submit(task);
}
public static void destory() {
status = "destory";
threadPoolExecutor.shutdownNow();
threadPoolExecutor.awaitTermination(3, TimeUnit.SECONDS);
}
}
简单的业务代码如上，在ProjectHolder释放资源销毁的时候有可能导致业务上会打印无效告警，虽然这种是不会影响用户体验的，但是会影响回流数据和告警分析。为了优化，将业务中读取流的时候进行了一个应用状态的判断。改动如下：
class Task implements Runnable {
@Override
public void run() {
Response response = null;
try {
...
while ("running".equals(ProjectHolder.status) && readLen = inputStream.read(bytes)) {
// do something
}
} catch (Exception ex) {
// 打印告警日志
} finally {
...
}
}
}
class ProjectHolder{
...
private static volatile String status = "running";
...
}
本以为这样就OK了，但是会有一定的概率报这个错okhttp3.internal.http2.Http2Stream.waitForIo$okhttp（Http2Stream.kt660)的错，去stackflow上看也有一些兄弟遇到这个问题，差点以为是OKhttp的bug。但是有一个评论说原因是多线程场景下读取一个公用的connection的header。这个提醒了我，是不是多线程场景下的问题。我就去看我们项目destory的时候是通过关闭线程池的方式是通过shutdownNow()方法，这个方法和shutdown()的区别简单讲就是前者会直接interrupt当前工作的线程，后者会interrupt空闲的线程，但是允许正在执行任务的线程继续执行。两者都会拒绝接受新的任务。
这就可能会有导致问题发生的可能性，所以我将shutdownNow()方法改成shutdown()方法后测试，发现确实没有再复现上面的错误了。
但是这个无法解释报错的根因。想要知道根因的话那就只能看OKhttp的源码了。
okhttp3.internal.http2.Http2Stream#waitForIo
void waitForIo() throws InterruptedIOException {
try {
this.wait();
} catch (InterruptedException var2) {
throw new InterruptedIOException();
}
}
这里会调用wait()方法将自己加入等待队列
public long read(Buffer sink, long byteCount) throws IOException {
if (byteCount < 0L) {
throw new IllegalArgumentException("byteCount < 0: " + byteCount);
} else {
long read;
synchronized(Http2Stream.this) {
Http2Stream.this.waitForIo();
....
}
}
}
waitForIo方法是在read()方法中去调用的。
void receive(BufferedSource in, long byteCount) throws IOException {
......
synchronized(Http2Stream.this) {
boolean wasEmpty = this.readBuffer.size() == 0L;
this.readBuffer.writeAll(this.receiveBuffer);
if (wasEmpty) {
Http2Stream.this.notifyAll();
}
}
}
}
在receive数据的时候会调用notifyAll()方法将自己从等待队列唤醒。
所以这里的大概流程就知道了，read方法读到Response中缓存全部数据时会阻塞，将自己wait加入到等待队列中去，在收到数据时将自己notify移出等待队列再次开始工作。这样就可能在while循环条件里，阻塞在 readLen = inputStream.read(bytes)这一行的时候，业务线程已经调用wait加入了等待队列中，shutdownNow()方法将线程Interrupt了，导致OkHttp出现了这个okhttp3.internal.http2.Http2Stream$waitForIo错误。
本文用的OKHttp版本是3.14.4
下面我们写一个小demo验证我们的猜想：
public class TestDemo {
private static Object LOCK = new Object();
public static void main(String[] args) throws Exception {
Runnable runnable = new Runnable() {
@Override
public void run() {
try {
synchronized (LOCK) {
TimeUnit.SECONDS.sleep(2);
LOCK.wait();
}
} catch (Exception ex) {
System.out.println("runnable occurs exception");
} finally {
System.out.println("runnable end");
}
}
};
Runnable runnable2 = new Runnable() {
@Override
public void run() {
try {
synchronized (LOCK) {
TimeUnit.SECONDS.sleep(5);
LOCK.notifyAll();
}
} catch (Exception ex) {
System.out.println("runnable2 occurs exception");
} finally {
System.out.println("runnable2 end");
}
}
};
Thread thread1 = new Thread(runnable);
thread1.start();
Thread thread2 = new Thread(runnable2);
thread2.start();
TimeUnit.SECONDS.sleep(3);
thread1.interrupt(); // interrupt等待队列中的线程报错，此处注释掉可以正常运行
System.out.println("main end");
}
}
输出：
main end
runnable2 end
runnable occurs exception
runnable end
