Benutzerdefinierte Implementierung von InputStream mit LinkedBlockingQueue hängen

Question

Ich habe versucht, eine InputStream zu implementieren, die ich Blöcke von Zeichenfolgen daran übergeben kann.
Technisch sollte es funktionieren, weil alles, was ich tun musste, die InputStream # read() -Methode zu blockieren, und das ist es. Oder so dachte ich so ... Hier ist meine individuelle Umsetzung des StringInputStream ist:

public class StringInputStream extends InputStream { 
    private LinkedBlockingQueue<Integer> buffer = new LinkedBlockingQueue<>(); 

    public void supplyData(String s) { 
     for (char ch : s.toCharArray()) { 
      buffer.add((int) ch); 
     } 
    } 

    @Override 
    public int read() throws IOException { 
     try { 
      return buffer.take(); 
     } catch (InterruptedException e) { 
      e.printStackTrace(); 
     } 
     return -1; 
    } 
} 

Und hier ist mein Code, um es zu testen:

public class StreamsMain { 
    public static void main(String[] args) throws InterruptedException { 
     InputStream is = new ByteArrayInputStream("eu ma duc la scoala\n sa ma distrez\nsi imi place la MAXIM!".getBytes(StandardCharsets.UTF_8)); 
     Scanner scanner1 = new Scanner(is); 
     AtomicReference<StringInputStream> inputStream = new AtomicReference<>(new StringInputStream()); 
     Thread th = new Thread(() -> { 
      Scanner scanner = new Scanner(inputStream.get()); 
      while (scanner.hasNextLine()) { 
       System.out.println("2. " + scanner.nextLine()); 
      } 
     }); 
     th.start(); 

     while (scanner1.hasNextLine()) { 
      String line = scanner1.nextLine(); 
      inputStream.get().supplyData(line + "\n"); 
      System.out.println("1. " + line); 
     } 

     System.out.println("\n\nwaiting 3 seconds to exit from MAIN thread"); 
     TimeUnit.SECONDS.sleep(3); 
     //th.interrupt(); 
     System.out.println("exited MAIN thread"); 
    } 
} 

In meinem Beispiel lese ich aus der erster Eingabestream, gebe ich die Zeilen zu meiner benutzerdefinierten Implementierung und dann lese ich von meiner benutzerdefinierten Implementierung in einem anderen Thread. Die seltsame Sache: Ich sehe keine Ausgabe, OHNE dass ich die th.interrupt() Zeile zerlege und das passiert NUR das Schlafen vom Hauptthread (was keinen Sinn ergibt, weil ich von meinem StringInputStream in einem anderen Thread).

Können Sie mir bitte helfen, das Problem zu erkennen?

Mit freundlichen Grüßen

Answer 1

Ich wurde tatsächlich neugierig dies selbst so begann ich diese aufzublicken

Als ich

versuchte

Thread th = new Thread(() -> { 
    System.out.println("Before scanner initialization"); 
    Scanner scanner = new Scanner(inputStream.get()); 
    while (scanner.hasNextLine()) { 
     System.out.println("2. " + scanner.nextLine()); 
    } 
}); 
th.start(); 

Der Druck vor der Scanner-Initialisierung gedruckt, was bedeutet, dass die Initialisierung des neuen Scanners blockiert den Thread.

Jetzt habe ich eigentlich noch nie versucht, von der InputStream-Klasse zu erben, aber als ich dies im Haupt-Thread versuchte, blockierte die Initialisierung nicht.

---

Tatsächlich ist die Codeblöcke hier

AtomicReference<StringInputStream> inputStream = new AtomicReference<>(new StringInputStream()); 
System.out.println("before"); 
Scanner scanner = new Scanner(inputStream.get()); 
while (scanner.hasNextLine()) { // Code is blocking here 
    System.out.println("2. " + scanner.nextLine()); 
} 
System.out.println("after"); 

Also, bevor gedruckt wird, während nach nicht der Fall ist.

---

es Figured

out Ich bin von Ihren genauen Anforderungen nicht sicher, aber wenn Sie Scanner mit Ihrem Input verwenden möchten, sollten Sie

read(byte[] b, int off, int len) 

Neben read() overrriding werden, da Sie erweitern eine Schnittstelle. Der Lesevorgang wird jedoch nicht vom Scanner aus aufgerufen. Sie können diese Klasse ausprobieren, um zu sehen, ob sie funktioniert.

class StringInputStream extends InputStream { 

    byte[] bytes; 
    int index = 0; 

    public StringInputStream(byte[] bytes) { 
     this.bytes = bytes; 
    } 

    public int read() { 
     return bytes[index]; 
    } 


    public int read(byte[] b, int off, int len) { 
     if(index == bytes.length) 
      return 0; 
     b[0] = bytes[index++]; 
     return 1; 
    } 
} 

Die Klasse wird von scanner.hasNextLine() falsch zurück, wenn Sie 0 Byte von read(byte[] b, int off, int len) zurückzukehren. Schätze, du lernst jeden Tag etwas Neues.

Hoffentlich hilft das!