正确的解决方案可能是以某种方式使用NIO。我已经在这里评论了Hadoop如何使用nio underneath。
但更简单的解决方案在Dexter的答案中。我还发现了EJP的答案,建议使用BufferedOutputStream
来控制数据何时输出。因此,我将两者结合起来得到了下面的TimedOutputStream
。它不能完全控制远程输出缓冲(大部分由操作系统完成),但结合适当的缓冲区大小和写入超时提供了至少一些控制(请参见第二个程序以测试TimedOutputStream
)。
我还没有完全测试TimedOutputStream
,所以请自行进行尽职调查。
编辑:更新了写入方法,以更好地协调缓冲区大小和写入超时时间,并微调了测试程序。添加了关于非安全异步关闭套接字输出流的注释。
import java.io.*;
import java.util.concurrent.*;
public class TimedOutputStream extends FilterOutputStream {
protected int timeoutMs = 50_000;
protected final boolean closeExecutor;
protected final ScheduledExecutorService executor;
protected ScheduledFuture<?> timeoutTask;
protected volatile boolean writeTimedout;
protected volatile IOException writeTimeoutCloseException;
public void setTimeoutMs(int timeoutMs) {
this.timeoutMs = timeoutMs;
}
public int getTimeoutMs() {
return timeoutMs;
}
public boolean isWriteTimeout() {
return writeTimedout;
}
public IOException getWriteTimeoutCloseException() {
return writeTimeoutCloseException;
}
public ScheduledExecutorService getScheduledExecutor() {
return executor;
}
@Override
public void close() throws IOException {
try {
super.close();
} finally {
if (closeExecutor) {
executor.shutdownNow();
}
}
}
protected byte buf[];
protected int count;
public TimedOutputStream(OutputStream out) {
this(out, null);
}
public TimedOutputStream(OutputStream out, ScheduledExecutorService executor) {
this(out, 8192, executor);
}
public TimedOutputStream(OutputStream out, int size) {
this(out, size, null);
}
public TimedOutputStream(OutputStream out, int size, ScheduledExecutorService executor) {
super(out);
if (size <= 0) {
throw new IllegalArgumentException("Buffer size <= 0");
}
if (executor == null) {
this.executor = Executors.newScheduledThreadPool(1);
ScheduledThreadPoolExecutor stp = (ScheduledThreadPoolExecutor) this.executor;
stp.setRemoveOnCancelPolicy(true);
closeExecutor = true;
} else {
this.executor = executor;
closeExecutor = false;
}
buf = new byte[size];
}
protected void flushBuffer(boolean flushOut) throws IOException {
if (count > 0 || flushOut) {
timeoutTask = executor.schedule(new TimeoutTask(this), getTimeoutMs(), TimeUnit.MILLISECONDS);
try {
if (count > 0) {
out.write(buf, 0, count);
count = 0;
}
if (flushOut) {
out.flush();
}
} finally {
timeoutTask.cancel(false);
}
}
}
protected class TimeoutTask implements Runnable {
protected final TimedOutputStream tout;
public TimeoutTask(TimedOutputStream tout) {
this.tout = tout;
}
@Override public void run() {
tout.interruptWriteOut();
}
}
protected void interruptWriteOut() {
try {
writeTimedout = true;
out.close();
} catch (IOException e) {
writeTimeoutCloseException = e;
}
}
@Override
public void write(int b) throws IOException {
if (count >= buf.length) {
flushBuffer(false);
}
buf[count++] = (byte)b;
}
@Override
public void write(byte b[], int off, int len) throws IOException {
if (count >= buf.length) {
flushBuffer(false);
}
if (len <= buf.length - count) {
System.arraycopy(b, off, buf, count, len);
count += len;
} else {
final int fill = buf.length - count;
System.arraycopy(b, off, buf, count, fill);
count += fill;
flushBuffer(false);
final int remaining = len - fill;
int start = off + fill;
for (int i = 0; i < remaining / buf.length; i++) {
System.arraycopy(b, start, buf, count, buf.length);
count = buf.length;
flushBuffer(false);
start += buf.length;
}
count = remaining % buf.length;
System.arraycopy(b, start, buf, 0, count);
}
}
@Override
public void flush() throws IOException {
if (!isWriteTimeout()) {
flushBuffer(true);
}
}
}
还有测试程序:
import java.io.*;
import java.net.*;
import java.util.concurrent.*;
public class TestTimedSocketOut implements Runnable, Closeable {
public static void main(String[] args) {
TestTimedSocketOut m = new TestTimedSocketOut();
try {
m.run();
} finally {
m.close();
}
}
final int clients = 3;
final int timeOut = 1000;
final int bufSize = 4096;
final long maxWait = 5000L;
byte[] largeMsg = new byte[28_602];
final ThreadPoolExecutor tp = (ThreadPoolExecutor) Executors.newCachedThreadPool();
final ScheduledThreadPoolExecutor stp = (ScheduledThreadPoolExecutor) Executors.newScheduledThreadPool(1);
final ConcurrentLinkedQueue<Closeable> closeables = new ConcurrentLinkedQueue<Closeable>();
final CountDownLatch[] serversReady = new CountDownLatch[clients];
final CountDownLatch clientsDone = new CountDownLatch(clients);
final CountDownLatch serversDone = new CountDownLatch(clients);
ServerSocket ss;
int port;
@Override public void run() {
stp.setRemoveOnCancelPolicy(true);
try {
ss = new ServerSocket();
ss.bind(null);
port = ss.getLocalPort();
tp.execute(new SocketAccept());
for (int i = 0; i < clients; i++) {
serversReady[i] = new CountDownLatch(1);
ClientSideSocket css = new ClientSideSocket(i);
closeables.add(css);
tp.execute(css);
Thread.sleep(50L);
}
if (!clientsDone.await(maxWait, TimeUnit.MILLISECONDS)) {
println("CLIENTS DID NOT FINISH");
} else {
if (!serversDone.await(maxWait, TimeUnit.MILLISECONDS)) {
println("SERVERS DID NOT FINISH");
} else {
println("Finished");
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
@Override public void close() {
try { if (ss != null) ss.close(); } catch (Exception ignored) {}
Closeable c = null;
while ((c = closeables.poll()) != null) {
try { c.close(); } catch (Exception ignored) {}
}
tp.shutdownNow();
println("Scheduled tasks executed: " + stp.getTaskCount() + ", max. threads: " + stp.getLargestPoolSize());
stp.shutdownNow();
}
class SocketAccept implements Runnable {
@Override public void run() {
try {
for (int i = 0; i < clients; i++) {
SeverSideSocket sss = new SeverSideSocket(ss.accept(), i);
closeables.add(sss);
tp.execute(sss);
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
class SeverSideSocket implements Runnable, Closeable {
Socket s;
int number, cnumber;
boolean completed;
public SeverSideSocket(Socket s, int number) {
this.s = s;
this.number = number;
cnumber = -1;
}
@Override public void run() {
String t = "nothing";
try {
DataInputStream in = new DataInputStream(s.getInputStream());
DataOutputStream out = new DataOutputStream(s.getOutputStream());
serversReady[number].countDown();
Thread.sleep(timeOut);
t = in.readUTF();
in.readFully(new byte[largeMsg.length], 0, largeMsg.length);
t += in.readUTF();
out.writeByte(1);
out.flush();
cnumber = in.readInt();
completed = true;
} catch (Exception e) {
println("server side " + number + " stopped after " + e);
} finally {
println("server side " + number + " received: " + t);
if (completed && cnumber != number) {
println("server side " + number + " expected client number " + number + " but got " + cnumber);
}
close();
serversDone.countDown();
}
}
@Override public void close() {
TestTimedSocketOut.close(s);
s = null;
}
}
class ClientSideSocket implements Runnable, Closeable {
Socket s;
int number;
public ClientSideSocket(int number) {
this.number = number;
}
@SuppressWarnings("resource")
@Override public void run() {
Byte b = -1;
TimedOutputStream tout = null;
try {
s = new Socket();
s.connect(new InetSocketAddress(port));
DataInputStream in = new DataInputStream(s.getInputStream());
tout = new TimedOutputStream(s.getOutputStream(), bufSize, stp);
if (number == 1) {
tout.setTimeoutMs(timeOut / 2);
} else {
tout.setTimeoutMs(timeOut * 2);
}
DataOutputStream out = new DataOutputStream(tout);
if (!serversReady[number].await(maxWait, TimeUnit.MILLISECONDS)) {
throw new RuntimeException("Server side for client side " + number + " not ready.");
}
out.writeUTF("client side " + number + " starting transfer");
out.write(largeMsg);
out.writeUTF(" - client side " + number + " completed transfer");
out.flush();
b = in.readByte();
out.writeInt(number);
out.flush();
} catch (Exception e) {
println("client side " + number + " stopped after " + e);
} finally {
println("client side " + number + " result: " + b);
if (tout != null) {
if (tout.isWriteTimeout()) {
println("client side " + number + " had write timeout, close exception: " + tout.getWriteTimeoutCloseException());
} else {
println("client side " + number + " had no write timeout");
}
}
close();
clientsDone.countDown();
}
}
@Override public void close() {
TestTimedSocketOut.close(s);
s = null;
}
}
private static void close(Socket s) {
try { if (s != null) s.close(); } catch (Exception ignored) {}
}
private static final long START_TIME = System.currentTimeMillis();
private static void println(String msg) {
System.out.println((System.currentTimeMillis() - START_TIME) + "\t " + msg);
}
}
OutputStream
。remote
参数来自第三方方法,我们无法更改它。 - Freewindjava.net.SocketTimeoutException: write blocked too long
。你在哪个平台上?当套接字发送缓冲区已满时,写操作应该无限期地阻塞。真正的问题在于读取端。为什么它跟不上呢? - user207421WriteStream#block()
。 - Ortwin Angermeier