Apakah mungkin membaca dari InputStream dengan batas waktu?

Secara khusus, masalahnya adalah menulis metode seperti ini:

int maybeRead(InputStream in, long timeout)

di mana nilai kembali sama dengan in.read () jika data tersedia dalam milidetik 'batas waktu', dan -2 sebaliknya. Sebelum metode kembali, semua thread yang muncul harus keluar.

Untuk menghindari argumen, subjek di sini java.io.InputStream, seperti yang didokumentasikan oleh Sun (semua versi Java). Harap dicatat ini tidak sesederhana kelihatannya. Berikut adalah beberapa fakta yang didukung langsung oleh dokumentasi Sun.

1. Metode in.read () mungkin non-interruptible.

2. Membungkus InputStream di Reader atau InterruptibleChannel tidak membantu, karena semua kelas yang dapat dilakukan adalah metode panggilan dari InputStream. Jika dimungkinkan untuk menggunakan kelas-kelas itu, akan mungkin untuk menulis solusi yang baru saja mengeksekusi logika yang sama langsung pada InputStream.

3. Itu selalu dapat diterima untuk in.available () untuk mengembalikan 0.

4. Metode in.close () dapat memblokir atau tidak melakukan apa pun.

5. Tidak ada cara umum untuk membunuh utas lainnya.

Menggunakan inputStream.available ()

Itu selalu dapat diterima untuk System.in.available () untuk mengembalikan 0.

Saya telah menemukan yang sebaliknya - selalu mengembalikan nilai terbaik untuk jumlah byte yang tersedia. Javadoc untuk InputStream.available():

Returns an estimate of the number of bytes that can be read (or skipped over) 
from this input stream without blocking by the next invocation of a method for 
this input stream.

Estimasi tidak dapat dihindari karena waktu / staleness. Angka tersebut bisa menjadi perkiraan sekali saja karena data baru terus berdatangan. Namun itu selalu "mengejar" pada panggilan berikutnya - itu harus memperhitungkan semua data yang tiba, bar yang tiba tepat pada saat panggilan baru. Mengembalikan 0 secara permanen ketika ada data yang gagal dengan kondisi di atas.

Peringatan Pertama: Subkelas beton InputStream bertanggung jawab atas ketersediaan ()

InputStreamadalah kelas abstrak. Tidak memiliki sumber data. Tidak ada artinya untuk memiliki data yang tersedia. Karenanya, javadoc untuk available()juga menyatakan:

The available method for class InputStream always returns 0.

This method should be overridden by subclasses.

Dan memang, kelas stream input konkret melakukan override available (), memberikan nilai yang bermakna, bukan 0s konstan.

Peringatan Kedua: Pastikan Anda menggunakan carriage-return saat mengetik input di Windows.

Jika menggunakan System.in, program Anda hanya menerima input ketika shell perintah Anda menyerahkannya. Jika Anda menggunakan pengalihan file / pipa (mis. Somefile> java myJavaApp atau somecommand | java myJavaApp), maka input data biasanya diserahkan segera. Namun, jika Anda mengetik input secara manual, maka penyerahan data dapat ditunda. Misalnya Dengan shell windows cmd.exe, data di buffer dalam shell cmd.exe. Data hanya diteruskan ke program java yang menjalankan mengikuti carriage-return (control-m atau <enter>). Itu batasan lingkungan eksekusi. Tentu saja, InputStream.available () akan mengembalikan 0 selama shell buffer data - itu perilaku yang benar; tidak ada data yang tersedia pada saat itu. Segera setelah data tersedia dari shell, metode mengembalikan nilai> 0. NB: Cygwin menggunakan cmd.

Solusi paling sederhana (tidak ada pemblokiran, jadi tidak perlu waktu habis)

Gunakan ini saja:

    byte[] inputData = new byte[1024];
    int result = is.read(inputData, 0, is.available());  
    // result will indicate number of bytes read; -1 for EOF with no data read.

ATAU setara,

    BufferedReader br = new BufferedReader(new InputStreamReader(System.in, Charset.forName("ISO-8859-1")),1024);
    // ...
         // inside some iteration / processing logic:
         if (br.ready()) {
             int readCount = br.read(inputData, bufferOffset, inputData.length-bufferOffset);
         }

Richer Solution (secara maksimal mengisi buffer dalam periode waktu habis)

Nyatakan ini:

public static int readInputStreamWithTimeout(InputStream is, byte[] b, int timeoutMillis)
     throws IOException  {
     int bufferOffset = 0;
     long maxTimeMillis = System.currentTimeMillis() + timeoutMillis;
     while (System.currentTimeMillis() < maxTimeMillis && bufferOffset < b.length) {
         int readLength = java.lang.Math.min(is.available(),b.length-bufferOffset);
         // can alternatively use bufferedReader, guarded by isReady():
         int readResult = is.read(b, bufferOffset, readLength);
         if (readResult == -1) break;
         bufferOffset += readResult;
     }
     return bufferOffset;
 }

Kemudian gunakan ini:

    byte[] inputData = new byte[1024];
    int readCount = readInputStreamWithTimeout(System.in, inputData, 6000);  // 6 second timeout
    // readCount will indicate number of bytes read; -1 for EOF with no data read.

Dengan asumsi aliran Anda tidak didukung oleh soket (sehingga Anda tidak dapat menggunakan Socket.setSoTimeout()), saya pikir cara standar untuk memecahkan masalah jenis ini adalah dengan menggunakan Masa Depan.

Misalkan saya memiliki pelaksana dan aliran berikut:

    ExecutorService executor = Executors.newFixedThreadPool(2);
    final PipedOutputStream outputStream = new PipedOutputStream();
    final PipedInputStream inputStream = new PipedInputStream(outputStream);

Saya memiliki penulis yang menulis beberapa data kemudian menunggu selama 5 detik sebelum menulis bagian terakhir dari data dan menutup aliran:

    Runnable writeTask = new Runnable() {
        @Override
        public void run() {
            try {
                outputStream.write(1);
                outputStream.write(2);
                Thread.sleep(5000);
                outputStream.write(3);
                outputStream.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    };
    executor.submit(writeTask);

Cara membaca yang normal adalah sebagai berikut. Bacaan akan memblokir tanpa batas untuk data dan ini selesai dalam 5s:

    long start = currentTimeMillis();
    int readByte = 1;
    // Read data without timeout
    while (readByte >= 0) {
        readByte = inputStream.read();
        if (readByte >= 0)
            System.out.println("Read: " + readByte);
    }
    System.out.println("Complete in " + (currentTimeMillis() - start) + "ms");

yang keluaran:

Read: 1
Read: 2
Read: 3
Complete in 5001ms

Jika ada masalah yang lebih mendasar, seperti penulis tidak merespons, pembaca akan memblokir selamanya. Jika saya membungkus pembacaan di masa depan, saya kemudian dapat mengontrol batas waktu sebagai berikut:

    int readByte = 1;
    // Read data with timeout
    Callable<Integer> readTask = new Callable<Integer>() {
        @Override
        public Integer call() throws Exception {
            return inputStream.read();
        }
    };
    while (readByte >= 0) {
        Future<Integer> future = executor.submit(readTask);
        readByte = future.get(1000, TimeUnit.MILLISECONDS);
        if (readByte >= 0)
            System.out.println("Read: " + readByte);
    }

yang keluaran:

Read: 1
Read: 2
Exception in thread "main" java.util.concurrent.TimeoutException
    at java.util.concurrent.FutureTask$Sync.innerGet(FutureTask.java:228)
    at java.util.concurrent.FutureTask.get(FutureTask.java:91)
    at test.InputStreamWithTimeoutTest.main(InputStreamWithTimeoutTest.java:74)

Saya dapat menangkap TimeoutException dan melakukan pembersihan apa pun yang saya inginkan.

Jika InputStream Anda didukung oleh Socket, Anda dapat mengatur timeout Socket (dalam milidetik) menggunakan setSoTimeout . Jika panggilan read () tidak membuka blokir dalam batas waktu yang ditentukan, itu akan melempar SocketTimeoutException.

Pastikan Anda memanggil setSoTimeout di Socket sebelum melakukan panggilan read ().

Saya akan mempertanyakan pernyataan masalah daripada hanya menerimanya secara membabi buta. Anda hanya perlu waktu tunggu dari konsol atau melalui jaringan. Jika yang terakhir Anda miliki Socket.setSoTimeout()dan HttpURLConnection.setReadTimeout()yang keduanya melakukan persis apa yang diperlukan, selama Anda mengaturnya dengan benar ketika Anda membangun / memperolehnya. Membiarkannya ke titik arbitrer kemudian dalam aplikasi ketika semua yang Anda miliki adalah InputStream adalah desain yang buruk yang mengarah ke implementasi yang sangat canggung.

Saya belum menggunakan kelas-kelas dari paket Java NIO, tetapi tampaknya mereka bisa membantu di sini. Khususnya, java.nio.channels.Channels dan java.nio.channels.InterruptibleChannel .

Berikut adalah cara untuk mendapatkan NIO FileChannel dari System.in dan memeriksa ketersediaan data menggunakan batas waktu, yang merupakan kasus khusus dari masalah yang dijelaskan dalam pertanyaan. Jalankan di konsol, jangan ketik input apa pun, dan tunggu hasilnya. Itu diuji dengan sukses di bawah Java 6 pada Windows dan Linux.

import java.io.FileInputStream;
import java.io.FilterInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.lang.reflect.Field;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedByInterruptException;

public class Main {

    static final ByteBuffer buf = ByteBuffer.allocate(4096);

    public static void main(String[] args) {

        long timeout = 1000 * 5;

        try {
            InputStream in = extract(System.in);
            if (! (in instanceof FileInputStream))
                throw new RuntimeException(
                        "Could not extract a FileInputStream from STDIN.");

            try {
                int ret = maybeAvailable((FileInputStream)in, timeout);
                System.out.println(
                        Integer.toString(ret) + " bytes were read.");

            } finally {
                in.close();
            }

        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }

    /* unravels all layers of FilterInputStream wrappers to get to the
     * core InputStream
     */
    public static InputStream extract(InputStream in)
            throws NoSuchFieldException, IllegalAccessException {

        Field f = FilterInputStream.class.getDeclaredField("in");
        f.setAccessible(true);

        while( in instanceof FilterInputStream )
            in = (InputStream)f.get((FilterInputStream)in);

        return in;
    }

    /* Returns the number of bytes which could be read from the stream,
     * timing out after the specified number of milliseconds.
     * Returns 0 on timeout (because no bytes could be read)
     * and -1 for end of stream.
     */
    public static int maybeAvailable(final FileInputStream in, long timeout)
            throws IOException, InterruptedException {

        final int[] dataReady = {0};
        final IOException[] maybeException = {null};
        final Thread reader = new Thread() {
            public void run() {                
                try {
                    dataReady[0] = in.getChannel().read(buf);
                } catch (ClosedByInterruptException e) {
                    System.err.println("Reader interrupted.");
                } catch (IOException e) {
                    maybeException[0] = e;
                }
            }
        };

        Thread interruptor = new Thread() {
            public void run() {
                reader.interrupt();
            }
        };

        reader.start();
        for(;;) {

            reader.join(timeout);
            if (!reader.isAlive())
                break;

            interruptor.start();
            interruptor.join(1000);
            reader.join(1000);
            if (!reader.isAlive())
                break;

            System.err.println("We're hung");
            System.exit(1);
        }

        if ( maybeException[0] != null )
            throw maybeException[0];

        return dataReady[0];
    }
}

Menariknya, ketika menjalankan program di dalam NetBeans 6.5 daripada di konsol, batas waktu tidak berfungsi sama sekali, dan panggilan ke System.exit () sebenarnya diperlukan untuk membunuh utas zombie. Apa yang terjadi adalah bahwa thread interruptor memblokir (!) Pada panggilan ke reader.interrupt (). Program pengujian lain (tidak ditampilkan di sini) juga mencoba menutup saluran, tetapi itu juga tidak berhasil.

Seperti dikatakan jt, NIO adalah solusi terbaik (dan benar). Jika Anda benar-benar terjebak dengan InputStream, Anda juga bisa

1. Tumbuhkan utas yang pekerjaan eksklusifnya adalah membaca dari InputStream dan memasukkan hasilnya ke dalam buffer yang dapat dibaca dari utas asli Anda tanpa memblokir. Ini akan bekerja dengan baik jika Anda hanya memiliki satu contoh aliran. Jika tidak, Anda mungkin dapat mematikan utas menggunakan metode yang tidak digunakan di kelas Utas, meskipun ini dapat menyebabkan kebocoran sumber daya.

2. Andalkan isAvailable untuk menunjukkan data yang dapat dibaca tanpa memblokir. Namun dalam beberapa kasus (seperti dengan Soket) dapat membaca pemblokiran yang potensial untuk tersedia untuk melaporkan sesuatu selain 0.

Terinspirasi dalam jawaban ini saya datang dengan solusi yang lebih berorientasi objek.

Ini hanya valid jika Anda bermaksud membaca karakter

Anda dapat mengganti BufferedReader dan mengimplementasikan sesuatu seperti ini:

public class SafeBufferedReader extends BufferedReader{

    private long millisTimeout;

    ( . . . )

    @Override
    public int read(char[] cbuf, int off, int len) throws IOException {
        try {
            waitReady();
        } catch(IllegalThreadStateException e) {
            return 0;
        }
        return super.read(cbuf, off, len);
    }

    protected void waitReady() throws IllegalThreadStateException, IOException {
        if(ready()) return;
        long timeout = System.currentTimeMillis() + millisTimeout;
        while(System.currentTimeMillis() < timeout) {
            if(ready()) return;
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                break; // Should restore flag
            }
        }
        if(ready()) return; // Just in case.
        throw new IllegalThreadStateException("Read timed out");
    }
}

Ini contoh yang hampir lengkap.

Saya mengembalikan 0 pada beberapa metode, Anda harus mengubahnya ke -2 untuk memenuhi kebutuhan Anda, tapi saya pikir 0 lebih cocok dengan kontrak BufferedReader. Tidak ada yang salah terjadi, ia hanya membaca 0 karakter. Metode readLine adalah pembunuh kinerja yang mengerikan. Anda harus membuat BufferedReader yang sama sekali baru jika Anda benar-benar ingin menggunakan readLin e. Saat ini, tidak aman untuk thread. Jika seseorang menjalankan operasi ketika readLines menunggu baris, itu akan menghasilkan hasil yang tidak terduga

Saya tidak suka kembali -2 di tempat saya. Saya akan melemparkan pengecualian karena beberapa orang mungkin hanya memeriksa jika int <0 untuk mempertimbangkan EOS. Bagaimanapun, metode-metode tersebut mengklaim bahwa "tidak dapat memblokir", Anda harus memeriksa apakah pernyataan itu benar dan hanya jangan menimpanya.

import java.io.BufferedReader;
import java.io.IOException;
import java.io.Reader;
import java.nio.CharBuffer;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;

/**
 * 
 * readLine
 * 
 * @author Dario
 *
 */
public class SafeBufferedReader extends BufferedReader{

    private long millisTimeout;

    private long millisInterval = 100;

    private int lookAheadLine;

    public SafeBufferedReader(Reader in, int sz, long millisTimeout) {
        super(in, sz);
        this.millisTimeout = millisTimeout;
    }

    public SafeBufferedReader(Reader in, long millisTimeout) {
        super(in);
        this.millisTimeout = millisTimeout;
    }



    /**
     * This is probably going to kill readLine performance. You should study BufferedReader and completly override the method.
     * 
     * It should mark the position, then perform its normal operation in a nonblocking way, and if it reaches the timeout then reset position and throw IllegalThreadStateException
     * 
     */
    @Override
    public String readLine() throws IOException {
        try {
            waitReadyLine();
        } catch(IllegalThreadStateException e) {
            //return null; //Null usually means EOS here, so we can't.
            throw e;
        }
        return super.readLine();
    }

    @Override
    public int read() throws IOException {
        try {
            waitReady();
        } catch(IllegalThreadStateException e) {
            return -2; // I'd throw a runtime here, as some people may just be checking if int < 0 to consider EOS
        }
        return super.read();
    }

    @Override
    public int read(char[] cbuf) throws IOException {
        try {
            waitReady();
        } catch(IllegalThreadStateException e) {
            return -2;  // I'd throw a runtime here, as some people may just be checking if int < 0 to consider EOS
        }
        return super.read(cbuf);
    }

    @Override
    public int read(char[] cbuf, int off, int len) throws IOException {
        try {
            waitReady();
        } catch(IllegalThreadStateException e) {
            return 0;
        }
        return super.read(cbuf, off, len);
    }

    @Override
    public int read(CharBuffer target) throws IOException {
        try {
            waitReady();
        } catch(IllegalThreadStateException e) {
            return 0;
        }
        return super.read(target);
    }

    @Override
    public void mark(int readAheadLimit) throws IOException {
        super.mark(readAheadLimit);
    }

    @Override
    public Stream<String> lines() {
        return super.lines();
    }

    @Override
    public void reset() throws IOException {
        super.reset();
    }

    @Override
    public long skip(long n) throws IOException {
        return super.skip(n);
    }

    public long getMillisTimeout() {
        return millisTimeout;
    }

    public void setMillisTimeout(long millisTimeout) {
        this.millisTimeout = millisTimeout;
    }

    public void setTimeout(long timeout, TimeUnit unit) {
        this.millisTimeout = TimeUnit.MILLISECONDS.convert(timeout, unit);
    }

    public long getMillisInterval() {
        return millisInterval;
    }

    public void setMillisInterval(long millisInterval) {
        this.millisInterval = millisInterval;
    }

    public void setInterval(long time, TimeUnit unit) {
        this.millisInterval = TimeUnit.MILLISECONDS.convert(time, unit);
    }

    /**
     * This is actually forcing us to read the buffer twice in order to determine a line is actually ready.
     * 
     * @throws IllegalThreadStateException
     * @throws IOException
     */
    protected void waitReadyLine() throws IllegalThreadStateException, IOException {
        long timeout = System.currentTimeMillis() + millisTimeout;
        waitReady();

        super.mark(lookAheadLine);
        try {
            while(System.currentTimeMillis() < timeout) {
                while(ready()) {
                    int charInt = super.read();
                    if(charInt==-1) return; // EOS reached
                    char character = (char) charInt;
                    if(character == '\n' || character == '\r' ) return;
                }
                try {
                    Thread.sleep(millisInterval);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt(); // Restore flag
                    break;
                }
            }
        } finally {
            super.reset();
        }
        throw new IllegalThreadStateException("readLine timed out");

    }

    protected void waitReady() throws IllegalThreadStateException, IOException {
        if(ready()) return;
        long timeout = System.currentTimeMillis() + millisTimeout;
        while(System.currentTimeMillis() < timeout) {
            if(ready()) return;
            try {
                Thread.sleep(millisInterval);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt(); // Restore flag
                break;
            }
        }
        if(ready()) return; // Just in case.
        throw new IllegalThreadStateException("read timed out");
    }

}