挂起LinkedBlockingQueue的InputStream的自定义实现

Question

我尝试实现一个InputStream，可以将字符串块传递给它。
从技术上讲，它应该可以工作，因为我所需要做的就是阻止InputStream#read（）方法，仅此而已。。。以下是我对StringInputStream的自定义实现：

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

这是我测试它的代码：

public class StreamsMain {
    public static void main(String[] args) throws InterruptedException {
        InputStream is = new ByteArrayInputStream("eu ma duc la scoalan sa ma distreznsi 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("nnwaiting 3 seconds to exit from MAIN thread");
        TimeUnit.SECONDS.sleep(3);
        //th.interrupt();
        System.out.println("exited MAIN thread");
    }
}

在我的示例中，我读取第一个输入流，将行提供给我的自定义实现，然后在另一个线程中读取我的自定义实施。

现在奇怪的是：我看不到任何输出，除非我停用th.interrupt（）行，而这只发生在主线程的睡眠中（这没有意义，因为我正在另一个线程中读取StringInputStream）。

你能帮我找出问题吗？

亲切问候，

Answer 1

实际上我自己也对此感到好奇，所以我开始查找

当我尝试时

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

扫描仪初始化之前的打印已打印，这意味着新扫描仪的初始化正在阻塞线程。

现在，我实际上从未尝试过从InputStream类继承，但当我尝试在主线程上继承时，初始化没有阻止。

---

实际上这里的代码块

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

所以之前是打印的，而之后不是。

---

搞清楚了

我不确定你的确切要求，但如果你想在InputStream中使用Scanner，你应该超过

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

除了read()，因为您正在扩展一个接口。但是，读取不会从扫描仪调用。你可以试试这个类，看看它是否有效。

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

如果从read(byte[] b, int off, int len)返回0个字节，则该类将从scanner.hasNextLine()返回false。我猜你每天都会学到一些新东西。

希望这能有所帮助！