记录一次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);
    

标签: none

添加新评论