java管道流的使用疑问Write end dead?

在做一个web日志浏览小工具时候 使用到了java的管道流. 但是使用过程中却一直报了 Write end dead 的异常.

我的需求是这样的:

1. 创建好的管道流, 输出流放入 日志的appender 里面,这样log4j 记录日志时候会自动写入,
2. 然后我再人为跑一个线程不断去读取管道的输入流, 每当有websocket连接进来时候就将其session放入一个集合,
3. 自定义的线程遍历session 推送日志到前端.
4. 如果当前没有session照样会不断的读取管道输入流,只是没有session就直接丢弃.

这样在调试过程中一开始是没问题的,但第一拨日志输出到web完时总会报Write end dead 异常,经过排查 发现读取时候会去检测管道输出流的线程存活状态.

    //管道输入流源码
   synchronized void receive(int c) throws IOException {
        if (!connected)
            throw new IOException("Pipe not connected");
        if (closedByWriter || closedByReader)
            throw new IOException("Pipe closed");
        if (readSide != null && !readSide.isAlive())
            throw new IOException("Read end dead");
            //将管道输出流当前的线程保存下来
        writeSide = Thread.currentThread();
        while (in == out) {
            if (readSide != null && !readSide.isAlive())
                throw new IOException("Pipe broken");
            notifyAll();
            try {
                wait(1000L);
            } catch (InterruptedException ex) {
                throw new InterruptedIOException();
            }
        }
        if (in < 0) {
            in = 0;
            out = 0;
        }
        buffer[in++] = (char) c;
        if (in >= buffer.length)
            in = 0;
    }

public synchronized boolean ready() throws IOException {
        if (!connected)
            throw new IOException("Pipe not connected");
        if (closedByReader)
            throw new IOException("Pipe closed");
        if (writeSide != null && !writeSide.isAlive() && !closedByWriter && in < 0)
            throw new IOException("Write end dead");
        return in >= 0;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

可以看到报的异常和 管道是否关闭 reader是否关闭没关系(这两个报其他异常).这异常只和write流所在线程是否存活有关系
而且这 [所在线程] 并不是真正的所在线程,只是 他接收writte写入数据时候读取到的write的线程
log4j写日志时候并不是固定使用一个线程的,而且对于web应用产生日志的线程也绝对不会一样,这样就造成 其实管道输出流是被N+个线程轮流使用的,而且这些线程都在不断的创建和消亡,但管道输入流一直是固定一个线程在使用, 无解
这样使用管道流 无论如何都会报异常

我尝试 直接将两个管道流的源码copy下来自己创建了自己的管道流,并将 线程判定的那段代码注释掉 正常了!!!!

import java.io.IOException;
import java.io.InterruptedIOException;
import java.io.Reader;
public class MyPipeRead extends Reader {

    public MyPipeRead(MyPipeWrite src) throws IOException {
        this(src, 1024);
    }

    public MyPipeRead(MyPipeWrite src, int pipeSize) throws IOException {
        closedByWriter = false;
        closedByReader = false;
        connected = false;
        in = -1;
        out = 0;
        initPipe(pipeSize);
        connect(src);
    }

    public MyPipeRead() {
        closedByWriter = false;
        closedByReader = false;
        connected = false;
        in = -1;
        out = 0;
        initPipe(1024);
    }

    public MyPipeRead(int pipeSize) {
        closedByWriter = false;
        closedByReader = false;
        connected = false;
        in = -1;
        out = 0;
        initPipe(pipeSize);
    }

    private void initPipe(int pipeSize) {
        if (pipeSize <= 0) {
            throw new IllegalArgumentException("Pipe size <= 0");
        } else {
            buffer = new char[pipeSize];
            return;
        }
    }

    public void connect(MyPipeWrite src) throws IOException {
        src.connect(this);
    }

    synchronized void receive(int c) throws IOException {
        if (!connected)
            throw new IOException("Pipe not connected");
        if (closedByWriter || closedByReader)
            throw new IOException("Pipe closed");
        if (readSide != null && !readSide.isAlive())
            throw new IOException("Read end dead");
        writeSide = Thread.currentThread();
        while (in == out) {
            if (readSide != null && !readSide.isAlive())
                throw new IOException("Pipe broken");
            notifyAll();
            try {
                wait(1000L);
            } catch (InterruptedException ex) {
                throw new InterruptedIOException();
            }
        }
        if (in < 0) {
            in = 0;
            out = 0;
        }
        buffer[in++] = (char) c;
        if (in >= buffer.length)
            in = 0;
    }

    synchronized void receive(char c[], int off, int len) throws IOException {
        while (--len >= 0)
            receive(c[off++]);
    }

    synchronized void receivedLast() {
        closedByWriter = true;
        notifyAll();
    }

    public synchronized int read() throws IOException {
        if (!connected)
            throw new IOException("Pipe not connected");
        if (closedByReader)
            throw new IOException("Pipe closed");
//        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)
                return -1;
            if (writeSide != null && !writeSide.isAlive() && --trials < 0)
                throw new IOException("Pipe broken");
            notifyAll();
            try {
                wait(1000L);
            } catch (InterruptedException ex) {
                throw new InterruptedIOException();
            }
        }
        int ret = buffer[out++];
        if (out >= buffer.length)
            out = 0;
        if (in == out)
            in = -1;
        return ret;
    }

    public synchronized int read(char cbuf[], int off, int len) throws IOException {
        if (!connected)
            throw new IOException("Pipe not connected");
        if (closedByReader)
            throw new IOException("Pipe closed");
//        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();
        if (len == 0)
            return 0;
        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)
                in = -1;
        }
        return rlen;
    }

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

    public void close() throws IOException {
        in = -1;
        closedByReader = true;
    }

    boolean closedByWriter;
    boolean closedByReader;
    boolean connected;
    Thread readSide;
    Thread writeSide;
    private static final int DEFAULT_PIPE_SIZE = 1024;
    char buffer[];
    int in;
    int out;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170

import java.io.IOException;
import java.io.Writer;

/**
 * 描述
 * @author Norton Lai
 * @created 2018-8-26 上午10:45:00
 */
public class MyPipeWrite extends Writer {

    public MyPipeWrite(MyPipeRead snk) throws IOException {
        closed = false;
        connect(snk);
    }

    public MyPipeWrite() {
        closed = false;
    }

    public synchronized void connect(MyPipeRead snk) throws IOException {
        if (snk == null)
            throw new NullPointerException();
        if (sink != null || snk.connected)
            throw new IOException("Already connected");
        if (snk.closedByReader || closed) {
            throw new IOException("Pipe closed");
        } else {
            sink = snk;
            snk.in = -1;
            snk.out = 0;
            snk.connected = true;
            return;
        }
    }

    public void write(int c) throws IOException {
        if (sink == null) {
            throw new IOException("Pipe not connected");
        } else {
            sink.receive(c);
            return;
        }
    }

    public void write(char cbuf[], int off, int len) throws IOException {
        if (sink == null)
            throw new IOException("Pipe not connected");
        if ((off | len | off + len | cbuf.length - (off + len)) < 0) {
            throw new IndexOutOfBoundsException();
        } else {
            sink.receive(cbuf, off, len);
            return;
        }
    }

    public synchronized void flush() throws IOException {
        if (sink != null) {
            if (sink.closedByReader || closed)
                throw new IOException("Pipe closed");
            synchronized (sink) {
                sink.notifyAll();
            }
        }
    }

    public void close() throws IOException {
        closed = true;
        if (sink != null)
            sink.receivedLast();
    }

    private MyPipeRead sink;
    private boolean closed;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

所以 在判定管道状态和流状态之后 还要判定线程状态的目的是什么 有什么作用?