java - LinkedBlockingQueue挂起的InputStream的自定义实现

Question

我尝试实现一个输入流，我可以将字符串 block 传递给它。
从技术上讲，它应该可以工作，因为我所需要做的就是阻止 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 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");
    }
}

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

现在奇怪的事情是: 我看不到任何输出，除非我对 th.interrupt() 行进行注释，并且这种情况仅发生在主线程的 hibernate 中(这没有意义，因为我正在另一个线程中从 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");

因此 before 被打印，而 after 则不打印。

<小时/>

想通了

我不确定您的确切要求，但如果您想将 Scanner 与您的 InputStream 一起使用，您应该覆盖

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

除了 read() 之外，因为您正在扩展接口(interface)。但是，扫描器不会调用读取操作。您可以尝试这个类，看看它是否有效。

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。我猜你每天都会学到新东西。

希望这有帮助!