继上一次查应用的CPU飙高问题（http://www.cnblogs.com/hzmark/p/JVM_CPU.html）过去10天了。上次只是定位到了是一个第三方包占用了大量的CPU使用，但没有细致的去查第三方包为什么占用了这么高的CPU，并且内存为什么如此诡异。总的来说上一次排查带来的收获是熟悉了JVM的工具使用和大致定位到了问题。
在上次排查问题之后，应用出现异常的频率还是较高，终下定决心再查一次，而这次排查的重点落在内存方面。因为怀疑CPU偏高是因为内存的异常导致频繁的GC引起的。
首先是JVM内存的状态：
S0和S1交替的用满，不停的在进行YGC。
使用jmap可以看到内存中那些对象使用了大量的内存：
下面是列出内存占用的方法（下图不是异常状态的时候的现象）：
在异常状态的情况下看到的现象是class name:[B占用好几个G的内存，也就是byte数组占用了很大的内存。
结合上一次查CPU问题时定位到的问题代码：
仔细看出问题的线程：com.trilead.ssh2.StreamGobbler$GobblerThread.run
到这里为止都是通过上次排查CPU问题可以推测出来的，其实仔细一点耐心一点顺着com.trilead.ssh2.StreamGobbler$GobblerThread.run继续往下分析能定位到具体问题，但是到这里之后我才用了另一种方式去定位问题，那就是分析出现问题时的堆内存使用情况，下面是我的分析过程。
首先dump内存数据：
jmap -dump:format=b,file=xxx.bin 3230
其中xxx.bin是dump出来的文件，3230是pid
之后使用mat打开该文件（这边不介绍MAT的使用，可以自己查找帮助文档；我使用的是eclipse的mat插件，可以在官网上找到update的地址在线安装就可以了http://download.eclipse.org/mat/1.3.1/update-site/）
这是一个概览信息，可以看到总共使用了618M内存，而有一块直接占用了576M，那么问题肯定出现在这块内存中。
点到leak suspects会有一个系统分析的怀疑问题：
和上一步我们的猜想是一致的，问题出在最大的这个内存这里。和排查CPU时得出的结论也是一致的，问题出在第三方的包内。
点击detail内容可以看到具体是哪里占用了内存。
通过with outgoing references查看线程引用的对象，如下图所示：
在找到使用内存最大的对象，查找他的引用，可以看到代码中是哪里引用了这块内存导致无法被回收
到这里位置就可以知道是自己起的线程保持了对内存的引用导致无法别回收。查自己写的代码已经可以精确的定位到问题。
（其实根据上次使用到一些JVM的工具已经可以发现问题，但是问了跟精确，顺便学习内存排查，所以有了MAT分析的过程）
下面是定位到问题代码之后的处理。
查看具体的代码：
1 package com.trilead.ssh2; 2 3 import java.io.IOException; 4 import java.io.InputStream; 5 import java.io.InterruptedIOException; 6 7 /** 8 * A <code>StreamGobbler</code> is an InputStream that uses an internal worker 9 * thread to constantly consume input from another InputStream. It uses a buffer 10 * to store the consumed data. The buffer size is automatically adjusted, if needed. 11 * <p> 12 * This class is sometimes very convenient - if you wrap a session's STDOUT and STDERR 13 * InputStreams with instances of this class, then you don't have to bother about 14 * the shared window of STDOUT and STDERR in the low level SSH-2 protocol, 15 * since all arriving data will be immediatelly consumed by the worker threads. 16 * Also, as a side effect, the streams will be buffered (e.g., single byte 17 * read() operations are faster). 18 * <p> 19 * Other SSH for Java libraries include this functionality by default in 20 * their STDOUT and STDERR InputStream implementations, however, please be aware 21 * that this approach has also a downside: 22 * <p> 23 * If you do not call the StreamGobbler's <code>read()</code> method often enough 24 * and the peer is constantly sending huge amounts of data, then you will sooner or later 25 * encounter a low memory situation due to the aggregated data (well, it also depends on the Java heap size). 26 * Joe Average will like this class anyway - a paranoid programmer would never use such an approach. 27 * <p> 28 * The term "StreamGobbler" was taken from an article called "When Runtime.exec() won't", 29 * see http://www.javaworld.com/javaworld/jw-12-2000/jw-1229-traps.html. 30 * 31 * @author Christian Plattner, plattner@trilead.com 32 * @version $Id: StreamGobbler.java,v 1.1 2007/10/15 12:49:56 cplattne Exp $ 33 */ 34 35 public class StreamGobbler extends InputStream 36 { 37 class GobblerThread extends Thread 38 { 39 public void run() 40 { 41 byte[] buff = new byte[8192]; 42 43 while (true) 44 { 45 try 46 { 47 int avail = is.read(buff); 48 49 synchronized (synchronizer) 50 { 51 if (avail <= 0) 52 { 53 isEOF = true; 54 synchronizer.notifyAll(); 55 break; 56 } 57 58 int space_available = buffer.length - write_pos; 59 60 if (space_available < avail) 61 { 62 /* compact/resize buffer */ 63 64 int unread_size = write_pos - read_pos; 65 int need_space = unread_size + avail; 66 67 byte[] new_buffer = buffer; 68 69 if (need_space > buffer.length) 70 { 71 int inc = need_space / 3; 72 inc = (inc < 256) ? 256 : inc; 73 inc = (inc > 8192) ? 8192 : inc; 74 new_buffer = new byte[need_space + inc]; 75 } 76 77 if (unread_size > 0) 78 System.arraycopy(buffer, read_pos, new_buffer, 0, unread_size); 79 80 buffer = new_buffer; 81 82 read_pos = 0; 83 write_pos = unread_size; 84 } 85 86 System.arraycopy(buff, 0, buffer, write_pos, avail); 87 write_pos += avail; 88 89 synchronizer.notifyAll(); 90 } 91 } 92 catch (IOException e) 93 { 94 synchronized (synchronizer) 95 { 96 exception = e; 97 synchronizer.notifyAll(); 98 break; 99 } 100 } 101 } 102 } 103 } 104 105 private InputStream is; 106 private final GobblerThread t; 107 108 private final Object synchronizer = new Object(); 109 110 private boolean isEOF = false; 111 private boolean isClosed = false; 112 private IOException exception = null; 113 114 private byte[] buffer = new byte[2048]; 115 private int read_pos = 0; 116 private int write_pos = 0; 117 118 public StreamGobbler(InputStream is) 119 { 120 this.is = is; 121 t = new GobblerThread(); 122 t.setDaemon(true); 123 t.start(); 124 } 125 126 public int read() throws IOException 127 { 128 synchronized (synchronizer) 129 { 130 if (isClosed) 131 throw new IOException("This StreamGobbler is closed."); 132 133 while (read_pos == write_pos) 134 { 135 if (exception != null) 136 throw exception; 137 138 if (isEOF) 139 return -1; 140 141 try 142 { 143 synchronizer.wait(); 144 } 145 catch (InterruptedException e) 146 { 147 throw new InterruptedIOException(); 148 } 149 } 150 151 int b = buffer[read_pos++] & 0xff; 152 153 return b; 154 } 155 } 156 157 public int available() throws IOException 158 { 159 synchronized (synchronizer) 160 { 161 if (isClosed) 162 throw new IOException("This StreamGobbler is closed."); 163 164 return write_pos - read_pos; 165 } 166 } 167 168 public int read(byte[] b) throws IOException 169 { 170 return read(b, 0, b.length); 171 } 172 173 public void close() throws IOException 174 { 175 synchronized (synchronizer) 176 { 177 if (isClosed) 178 return; 179 isClosed = true; 180 isEOF = true; 181 synchronizer.notifyAll(); 182 is.close(); 183 } 184 } 185 186 public int read(byte[] b, int off, int len) throws IOException 187 { 188 if (b == null) 189 throw new NullPointerException(); 190 191 if ((off < 0) || (len < 0) || ((off + len) > b.length) || ((off + len) < 0) || (off > b.length)) 192 throw new IndexOutOfBoundsException(); 193 194 if (len == 0) 195 return 0; 196 197 synchronized (synchronizer) 198 { 199 if (isClosed) 200 throw new IOException("This StreamGobbler is closed."); 201 202 while (read_pos == write_pos) 203 { 204 if (exception != null) 205 throw exception; 206 207 if (isEOF) 208 return -1; 209 210 try 211 { 212 synchronizer.wait(); 213 } 214 catch (InterruptedException e) 215 { 216 throw new InterruptedIOException(); 217 } 218 } 219 220 int avail = write_pos - read_pos; 221 222 avail = (avail > len) ? len : avail; 223 224 System.arraycopy(buffer, read_pos, b, off, avail); 225 226 read_pos += avail; 227 228 return avail; 229 } 230 } 231 }
如果使用这个类之前知悉读类的说明信息是可以看明白这个类存在的问题的。当然也可以从源码上看出来，我就是从源码看出问题的，当时花的时间肯定比看说明耗费的时间长。
下面是产生问题的代码：
1 class GobblerThread extends Thread 2 { 3 public void run() 4 { 5 byte[] buff = new byte[8192]; 6 7 while (true) 8 { 9 try 10 { 11 int avail = is.read(buff); 12 13 synchronized (synchronizer) 14 { 15 if (avail <= 0) 16 { 17 isEOF = true; 18 synchronizer.notifyAll(); 19 break; 20 } 21 22 int space_available = buffer.length - write_pos; 23 24 if (space_available < avail) 25 { 26 /* compact/resize buffer */ 27 28 int unread_size = write_pos - read_pos; 29 int need_space = unread_size + avail; 30 31 byte[] new_buffer = buffer; 32 33 if (need_space > buffer.length) 34 { 35 int inc = need_space / 3; 36 inc = (inc < 256) ? 256 : inc; 37 inc = (inc > 8192) ? 8192 : inc; 38 new_buffer = new byte[need_space + inc]; 39 } 40 41 if (unread_size > 0) 42 System.arraycopy(buffer, read_pos, new_buffer, 0, unread_size); 43 44 buffer = new_buffer; 45 46 read_pos = 0; 47 write_pos = unread_size; 48 } 49 50 System.arraycopy(buff, 0, buffer, write_pos, avail); 51 write_pos += avail; 52 53 synchronizer.notifyAll(); 54 } 55 } 56 catch (IOException e) 57 { 58 synchronized (synchronizer) 59 { 60 exception = e; 61 synchronizer.notifyAll(); 62 break; 63 } 64 } 65 } 66 } 67 }
StreamGobbler类的buffer属性是一个只会不断增长，且不会清空的byte数组，问题就出在这里。
当一个Hive查询的执行时间非常的长（比如出现数据倾斜导致查询进行不下去、执行任务很多需要排队导致时间边长），那么这个线程将持续很长时间，而它将hold这个buffer永远不会被回收，虽然它里面的内容再也不会被使用。
那么这就是不合理的地方。StreamGobbler类的设计就是启动一个线程将一个输入流的内容不停的清空，读到自己的缓存中。这样的目的是不要造成流的堵塞，但这样就不适用长任务且日志不停输出的场景。
至此排查分析结束，原因就是第三方的StreamGobbler并不适合当前的场景。解决的办法就是自己实现读取流的线程及时的处理掉流中的内容，和StreamGobbler起一个线程去读试一致的，只是不做缓存处理。
经过这次CPU飙高查到原因，基本已经熟悉了JVM工具的使用，CPU问题及内存问题的排查流程，算是有些收获。看问题，排查定位问题，看源码分析，最终解决掉问题的过程让自己非常的“爽”，并且得到成长。
和各位博友一起成长，加油。
如果本文对您有帮助，点一下右下角的“推荐”
