Czy można czytać z InputStream z limitem czasu?

W szczególności problem polega na napisaniu takiej metody:

int maybeRead(InputStream in, long timeout)

gdzie wartość zwracana jest taka sama jak in.read (), jeśli dane są dostępne w milisekundach „timeout”, lub -2 w przeciwnym razie. Przed zwróceniem metody wszystkie utworzone wątki muszą zostać zakończone.

Aby uniknąć argumentów, temat tutaj java.io.InputStream, zgodnie z dokumentacją firmy Sun (dowolna wersja Java). Pamiętaj, że to nie jest tak proste, jak się wydaje. Poniżej znajduje się kilka faktów, które są bezpośrednio poparte dokumentacją firmy Sun.

1. Metoda in.read () może być nieprzerywalna.

2. Zawijanie InputStream w Reader lub InterruptibleChannel nie pomaga, ponieważ wszystkie te klasy mogą zrobić, to wywołać metody InputStream. Gdyby można było użyć tych klas, byłoby możliwe napisanie rozwiązania, które po prostu wykonuje tę samą logikę bezpośrednio na InputStream.

3. Zawsze dopuszczalne jest, aby in.available () zwracała 0.

4. Metoda in.close () może blokować lub nic nie robić.

5. Nie ma ogólnego sposobu na zabicie innego wątku.

Korzystanie z metody inputStream.available ()

Zawsze akceptowalne jest, aby System.in.available () zwracało 0.

Znalazłem coś przeciwnego - zawsze zwraca najlepszą wartość dla liczby dostępnych bajtów. Javadoc dla 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.

Oszacowanie jest nieuniknione ze względu na harmonogram / nieaktualność. Liczba ta może być jednorazowym niedoszacowaniem, ponieważ stale napływają nowe dane. Jednak zawsze „nadrabia zaległości” przy następnym wywołaniu - powinien uwzględniać wszystkie otrzymane dane, z wyjątkiem tego, że przychodzi tylko w momencie nowego połączenia. Trwałe zwracanie 0, gdy istnieją dane, nie spełnia powyższego warunku.

Pierwsze zastrzeżenie: konkretne podklasy InputStream są odpowiedzialne za available ()

InputStreamjest klasą abstrakcyjną. Nie ma źródła danych. Posiadanie dostępnych danych nie ma sensu. Dlatego javadoc dla available()również stwierdza:

The available method for class InputStream always returns 0.

This method should be overridden by subclasses.

I rzeczywiście, konkretne klasy strumienia wejściowego przesłaniają available (), dostarczając znaczące wartości, a nie stałe 0.

Drugie zastrzeżenie: upewnij się, że używasz powrotu karetki podczas wpisywania danych wejściowych w systemie Windows.

Jeśli używasz System.in, twój program otrzymuje dane wejściowe tylko wtedy, gdy powłoka poleceń przekazuje je. Jeśli używasz przekierowań plików / potoków (np. Somefile> java myJavaApp lub somecommand | java myJavaApp), dane wejściowe są zwykle przekazywane natychmiast. Jeśli jednak ręcznie wpiszesz dane wejściowe, przekazanie danych może zostać opóźnione. Np. W powłoce cmd.exe systemu Windows dane są buforowane w powłoce cmd.exe. Dane są przekazywane do wykonującego się programu java tylko po powrocie karetki (control-m lub <enter>). To jest ograniczenie środowiska wykonawczego. Oczywiście InputStream.available () zwraca 0 tak długo, jak długo powłoka buforuje dane - to poprawne zachowanie; w tym momencie nie ma dostępnych danych. Gdy tylko dane są dostępne z powłoki, metoda zwraca wartość> 0. NB: Cygwin używa cmd.

Najprostsze rozwiązanie (bez blokowania, więc nie jest wymagany limit czasu)

Po prostu użyj tego:

    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.

LUB równoważnie,

    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);
         }

Bogatsze rozwiązanie (maksymalnie wypełnia bufor w określonym czasie)

Oświadcz, że:

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;
 }

Następnie użyj tego:

    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.

Zakładając, że twój strumień nie jest obsługiwany przez gniazdo (więc nie możesz go użyć Socket.setSoTimeout()), myślę, że standardowym sposobem rozwiązania tego typu problemu jest użycie Future.

Załóżmy, że mam następujący moduł wykonawczy i strumienie:

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

Mam program zapisujący, który zapisuje dane, a następnie czeka 5 sekund przed zapisaniem ostatniej części danych i zamknięciem strumienia:

    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);

Normalny sposób czytania tego jest następujący. Odczyt będzie blokowany na czas nieokreślony dla danych, więc kończy się to w ciągu 5 sekund:

    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");

które wyjścia:

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

Gdyby istniał bardziej podstawowy problem, jak brak odpowiedzi pisarza, czytelnik zablokowałby się na zawsze. Jeśli zawiniesz odczyt w przyszłości, będę mógł kontrolować limit czasu w następujący sposób:

    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);
    }

które wyjścia:

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)

Mogę przechwycić wyjątek TimeoutException i wykonać dowolne czyszczenie.

Jeśli Twój InputStream jest obsługiwany przez Socket, możesz ustawić limit czasu Socket (w milisekundach) za pomocą setSoTimeout . Jeśli wywołanie read () nie odblokuje się w określonym czasie, zgłosi SocketTimeoutException.

Po prostu upewnij się, że wywołujesz setSoTimeout na Socket przed wykonaniem wywołania read ().

Wolałbym raczej zakwestionować stwierdzenie problemu niż po prostu przyjąć je na ślepo. Potrzebujesz tylko limitów czasu z konsoli lub przez sieć. Jeśli masz to drugie, Socket.setSoTimeout()i HttpURLConnection.setReadTimeout()które oboje robią dokładnie to, co jest wymagane, o ile skonfigurujesz je poprawnie podczas ich konstruowania / zdobywania. Pozostawienie tego w dowolnym punkcie później w aplikacji, gdy wszystko, co masz, to kiepski projekt InputStream, co prowadzi do bardzo niewygodnej implementacji.

Nie korzystałem z klas z pakietu Java NIO, ale wygląda na to, że mogą być tutaj pomocne. W szczególności java.nio.channels.Channels i java.nio.channels.InterruptibleChannel .

Oto sposób na pobranie NIO FileChannel z System.in i sprawdzenie dostępności danych przy użyciu limitu czasu, co jest szczególnym przypadkiem problemu opisanego w pytaniu. Uruchom go na konsoli, nie wpisuj żadnych danych wejściowych i czekaj na wyniki. Został pomyślnie przetestowany pod Javą 6 w systemach Windows i 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];
    }
}

Co ciekawe, podczas uruchamiania programu w NetBeans 6.5, a nie na konsoli, limit czasu w ogóle nie działa, a wywołanie System.exit () jest faktycznie konieczne, aby zabić wątki zombie. Dzieje się tak, że wątek przerywacza blokuje (!) Wywołanie funkcji reader.interrupt (). Inny program testowy (tutaj nie pokazany) dodatkowo próbuje zamknąć kanał, ale to też nie działa.

Jak powiedziałem, NIO jest najlepszym (i poprawnym) rozwiązaniem. Jeśli jednak naprawdę utkniesz w InputStream, możesz też

1. Stwórz wątek, którego wyłącznym zadaniem jest odczyt z InputStream i umieszczenie wyniku w buforze, który można odczytać z oryginalnego wątku bez blokowania. Powinno to działać dobrze, jeśli masz tylko jedną instancję strumienia. W przeciwnym razie możesz zabić wątek przy użyciu przestarzałych metod w klasie Thread, chociaż może to spowodować wycieki zasobów.

2. Polegaj na isAvailable, aby wskazać dane, które można odczytać bez blokowania. Jednak w niektórych przypadkach (na przykład w przypadku gniazd) może zająć potencjalnie blokujący odczyt dla isAvailable, aby zgłosić coś innego niż 0.

Zainspirowany tą odpowiedzią wymyśliłem nieco bardziej obiektowe rozwiązanie.

Jest to ważne tylko wtedy, gdy zamierzasz czytać znaki

Możesz zastąpić BufferedReader i zaimplementować coś takiego:

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");
    }
}

Oto prawie kompletny przykład.

Zwracam 0 dla niektórych metod, powinieneś zmienić to na -2, aby spełnić twoje potrzeby, ale myślę, że 0 jest bardziej odpowiednie z kontraktem BufferedReader. Nic się nie stało, po prostu odczytano 0 znaków. Metoda readLine jest strasznym zabójcą wydajności. Powinieneś utworzyć całkowicie nowy BufferedReader, jeśli faktycznie chcesz używać readLin e. W tej chwili nie jest bezpieczny wątkowo. Jeśli ktoś wywoła operację, gdy readLines czeka na wiersz, przyniesie to nieoczekiwane wyniki

Nie lubię wracać -2 gdzie jestem. Rzuciłbym wyjątek, ponieważ niektórzy ludzie mogą po prostu sprawdzać, czy int <0, aby rozważyć EOS. W każdym razie te metody twierdzą, że „nie można blokować”, należy sprawdzić, czy to stwierdzenie jest rzeczywiście prawdziwe i po prostu ich nie zastępować.

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");
    }

}