记录一次tcp连接持续断掉的排查过程

有人上报说他们的 app 连接另外一个微服务的时候, 连接经常断掉, 不知道什么原因.

查看网络方面的监控, 网络一切安好.

查看日志, 看到一些异常信息: java.nio.channels.ClosedByInterruptException. 顺着这个信息, 我们找到了根本原因.

1. 既然是 java.nio.channels.ClosedByInterruptException, 那么我们可以用 btrace 查看为什么这个异常被创建, 得到结果如下:

   java.nio.channels.ClosedByInterruptException.<init>(ClosedByInterruptException.java:51)
   java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:202)
   sun.nio.ch.SocketChannelImpl.write(SocketChannelImpl.java:477)
   com.tianxiaohui.java.BufferedWriteChannel.flush(BufferedWriteChannel.java:281)
   com.tianxiaohui.java.BufferedWriteChannel.write(BufferedWriteChannel.java:223)
   com.tianxiaohui.java.BaseClient.sendBytesActual(BaseClient.java:339)
   com.tianxiaohui.java.BaseClient.sendBytesOverChannel(BaseClient.java:203)
   ...
   rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:48)
   rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:30)
   rx.Observable.unsafeSubscribe(Observable.java:10327) 
   rx.internal.operators.OperatorSubscribeOn$SubscribeOnSubscriber.call(OperatorSubscribeOn.java:100)
   rx.xinternal.schedulers.CachedThreadScheduler$EventLoopWorker$1.call(CachedThreadScheduler.java:230)
   rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55)
   ...
    java.lang.Thread.run(Thread.java:748)

2. 既然是 interrupt 异常, 那么就一定有其它线程给这个线程发 interrupt, 于是通过下面的 btrace 代码, 我们就能找到发 interrupt 的线程

   import org.openjdk.btrace.core.annotations.*;
   import static org.openjdk.btrace.core.BTraceUtils.*;
   
   import org.openjdk.btrace.core.BTraceUtils.Strings;
   
   @BTrace
   public class ThreadInterruptTrace {
   
    @OnMethod( clazz="/java\\.nio\\.channels\\.ClosedByInterruptException/", method="<init>" )
    public static void createException() {
      println(Strings.strcat(name(currentThread()), " --- get ClosedByInterruptException "));
      println(jstackStr());
    }
   
    @OnMethod( clazz="/java\\.lang\\.Thread1/", method="/interrupt/" )
    public static void t1r(@ProbeClassName String probeClass, @Self Thread self) {
      if (Strings.startsWith(name(self), "RxIoScheduler")) {
          println(Strings.strcat(Strings.strcat(name(currentThread()), " try to interrupt ->  "), str(self)));
            println(jstackStr());
        }
    }
   }

3. 找到发 interrupt 的线程, 那么就去看代码, 找了发生 interrupt 的原因, 最终原因是下面的代码. 每次 create 一个 Observable 都要新起一个线程池做为 scheduler, 其实不需要. 到时每次新建一个 scheduler 线程池导致创建太多线程池, 消耗大量内存, 之后又自动销毁. 然后这些 scheduler 销毁的时候, 还要通知 I/O 线程(这里的 I/O 线程池用的默认的), 也就是给 I/O 线程发一个 interrupt. 最终导致了我们看到的问题.

   Observable.create(new Observable.OnSubscribe<List<T>>() {
    @Override
    public void call(Subscriber<? super List<T>> subscriber) {
           //some code here
      }
    }).}).observeOn(Schedulers.newThread())
       .subscribeOn(Schedulers.io())
       .timeout(threadTimeout, TimeUnit.MILLISECONDS);