[原创]java：Stream、Socket等源码分析

一、对于java启动之后的线程的说明

　　java在启动后会有几个特殊线程：

　　1、main线程，主线程

　　2、JVM线程，虚拟机的线程

　　3、GC垃圾回收线程，是个守护线程

　　4、EDT&Toolkit

　　5、在启动图形界面时会自动创建两个线程，用于接收事件之前阻塞界面

　　　　AWT-Shutdown与AWT-EventQueue-0，所以在触发按钮事件时，所有的操作都是在AWT-EventQueue-0线程中进行的，而不是在主线程中。

　　　　在AWT.setVisible之后这两个线程会开辟出来，setVisible之后的代码还是在当前线程中运行的。

二、InputStream的read操作。

　　int read()，如果因为已经到达流末尾而没有可用的字节，则返回值

-1

。在输入数据可用、检测到流末尾或者抛出异常前，此方法一直阻塞。

　　要使用int作为返回值也是因为返回值-1的缘故，这样就不会和byte中的-1冲突了。

　　那么什么时候才算流末尾呢？

　　不同子类read操作不同，流末尾的判别也不同。

　　ByteArrayInputStream：

public synchronized int read() {
return (pos < count) ? (buf[pos++] & 0xff) : -1;
}

　　因为此类是使用已有缓冲区创建的，所以在读到缓冲区结尾时即返回-1，流末尾即超过缓冲区，不存在线程阻塞。

　　FileInputStream：

public native int read() throws IOException;

　　使用其他语言实现，很大可能是使用C语言来实现的，C语言读到文件末尾会返回EOF标记，处理为-1返回给调用方。

　　BufferedInputStream：

　　因为仅仅是加了缓冲区的InputStream，所以read操作还是调用的其他InputStream类的read。

　　PipedInputStream：稍微复杂点

　　使用了两个线程，所以PipedInputStream与PipedOutputStream一定要使用两个线程来创建，否则容易阻塞当前线程。

　　在in<0时死循环，在连接未关闭时，只有在接收到数据时才会把in在其他线程中置为0，此时可以跳出此循环，可以继续往下执行。连接关闭时会返回-1。

public synchronized int read()  throws IOException {
if (!connected) {
throw new IOException("Pipe not connected");
} else if (closedByReader) {
throw new IOException("Pipe closed");
} else if (writeSide != null && !writeSide.isAlive()
&& !closedByWriter && (in < 0)) {
throw new IOException("Write end dead");
}

readSide = Thread.currentThread();
int trials = 2;
while (in < 0) {
if (closedByWriter) {
/* closed by writer, return EOF */
return -1;
}
if ((writeSide != null) && (!writeSide.isAlive()) && (--trials < 0)) {
throw new IOException("Pipe broken");
}
/* might be a writer waiting */
notifyAll();
try {
wait(1000);
} catch (InterruptedException ex) {
throw new java.io.InterruptedIOException();
}
}
int ret = buffer[out++] & 0xFF;
if (out >= buffer.length) {
out = 0;
}
if (in == out) {
/* now empty */
in = -1;
}

return ret;
}

三、Reader的read操作

　　Reader类中有一个lock对象，表示当前对象，在Reader的所有操作用都会有一句：synchronized (lock)，即锁定当前对象，所以在阻塞时，也不允许其他线程对该对象的其他任何操作，包括关闭等。

protected Reader() {
this.lock = this;
}

　　BufferedReader:

public int read(char cbuf[], int off, int len) throws IOException {
synchronized (lock) {
ensureOpen();
if ((off < 0) || (off > cbuf.length) || (len < 0) ||
((off + len) > cbuf.length) || ((off + len) < 0)) {
throw new IndexOutOfBoundsException();
} else if (len == 0) {
return 0;
}

int n = read1(cbuf, off, len);
if (n <= 0) return n;
while ((n < len) && in.ready()) {
int n1 = read1(cbuf, off + n, len - n);
if (n1 <= 0) break;
n += n1;
}
return n;
}
}

　　而在read1中，则是调用了fill()方法，去填充缓冲区：

private void fill() throws IOException {
int dst;
if (markedChar <= UNMARKED) {
/* No mark */
dst = 0;
} else {
/* Marked */
int delta = nextChar - markedChar;
if (delta >= readAheadLimit) {
/* Gone past read-ahead limit: Invalidate mark */
markedChar = INVALIDATED;
readAheadLimit = 0;
dst = 0;
} else {
if (readAheadLimit <= cb.length) {
/* Shuffle in the current buffer */
System.arraycopy(cb, markedChar, cb, 0, delta);
markedChar = 0;
dst = delta;
} else {
/* Reallocate buffer to accommodate read-ahead limit */
char ncb[] = new char[readAheadLimit];
System.arraycopy(cb, markedChar, ncb, 0, delta);
cb = ncb;
markedChar = 0;
dst = delta;
}
nextChar = nChars = delta;
}
}

int n;
do {
n = in.read(cb, dst, cb.length - dst);
} while (n == 0);
if (n > 0) {
nChars = dst + n;
nextChar = dst;
}
}

　　可以看到在n==0时，do会死循环，也就是说，在未从流中读到数据时，此线程会通过死循环阻塞。在流中读到0个字节的数据与读到-1是不同的，-1表示流结尾。

　　readLine方法中：

String readLine(boolean ignoreLF) throws IOException {
StringBuffer s = null;
int startChar;

synchronized (lock) {
ensureOpen();
boolean omitLF = ignoreLF || skipLF;

bufferLoop:
for (;;) {

if (nextChar >= nChars)
fill();
if (nextChar >= nChars) { /* EOF */
if (s != null && s.length() > 0)
return s.toString();
else
return null;
}
boolean eol = false;
char c = 0;
int i;

/* Skip a leftover '\n', if necessary */
if (omitLF && (cb[nextChar] == '\n'))
nextChar++;
skipLF = false;
omitLF = false;

charLoop:
for (i = nextChar; i < nChars; i++) {
c = cb[i];
if ((c == '\n') || (c == '\r')) {
eol = true;
break charLoop;
}
}

startChar = nextChar;
nextChar = i;

if (eol) {
String str;
if (s == null) {
str = new String(cb, startChar, i - startChar);
} else {
s.append(cb, startChar, i - startChar);
str = s.toString();
}
nextChar++;
if (c == '\r') {
skipLF = true;
}
return str;
}

if (s == null)
s = new StringBuffer(defaultExpectedLineLength);
s.append(cb, startChar, i - startChar);
}
}
}

　　同样是调用了fill()，不同的是readLine在未读到/r与/n之前都会一直阻塞。

　　CharArrayReader：

　　直接通过现有缓冲区读数据，不存在阻塞等问题。

　　InputStreamReader：

　　使用StreamDecoder进行read。

　　PipedReader：

　　同样是使用两个线程进行操作，同PipedInputStream。

public synchronized int read(char cbuf[], int off, int len)  throws IOException {
if (!connected) {
throw new IOException("Pipe not connected");
} else if (closedByReader) {
throw new IOException("Pipe closed");
} else if (writeSide != null && !writeSide.isAlive()
&& !closedByWriter && (in < 0)) {
throw new IOException("Write end dead");
}

if ((off < 0) || (off > cbuf.length) || (len < 0) ||
((off + len) > cbuf.length) || ((off + len) < 0)) {
throw new IndexOutOfBoundsException();
} else if (len == 0) {
return 0;
}

/* possibly wait on the first character */
int c = read();
if (c < 0) {
return -1;
}
cbuf[off] =  (char)c;
int rlen = 1;
while ((in >= 0) && (--len > 0)) {
cbuf[off + rlen] = buffer[out++];
rlen++;
if (out >= buffer.length) {
out = 0;
}
if (in == out) {
/* now empty */
in = -1;
}
}
return rlen;
}

　　StringReader：

　　使用现有字符串进行操作，所以也不存在阻塞。

public int read(char cbuf[], int off, int len) throws IOException {
synchronized (lock) {
ensureOpen();
if ((off < 0) || (off > cbuf.length) || (len < 0) ||
((off + len) > cbuf.length) || ((off + len) < 0)) {
throw new IndexOutOfBoundsException();
} else if (len == 0) {
return 0;
}
if (next >= length)
return -1;
int n = Math.min(length - next, len);
str.getChars(next, next + n, cbuf, off);
next += n;
return n;
}
}

　　因为上边这些都是通过while死循环来进行锁死的，所以在使用到这些的时候建议放在一个单独的线程中，以免影响程序正常运行。

四、DatagramSocket的receive与send

　　receive方法在接收到数据报前一直阻塞。

public synchronized void receive(DatagramPacket p) throws IOException {
synchronized (p) {
if (!isBound())
bind(new InetSocketAddress(0));
if (connectState == ST_NOT_CONNECTED) {
SecurityManager security = System.getSecurityManager();
　　　　　　　　　　……
　　　　　　　　　　……
}
　　　　if (connectState == ST_CONNECTED_NO_IMPL) {
boolean stop = false;
　　　　　　　　　　……
　　　　　　　　　　……
}
getImpl().receive(p);
}
}

　　两个锁，一个锁当前对象，一个锁DatagramPacket。阻塞线程的操作是交给getImpl().receive(p)来进行的。

　　对于send()方法：

public void send(DatagramPacket p) throws IOException  {
InetAddress packetAddress = null;
synchronized (p) {

　　只锁了DatagramPacket，所以在不使用同一个DatagramPacket来发送和接收数据的情况下，可以使用同一个DatagramSocket对象在不同线程中进行发送和接收操作。