Java中的非阻塞文件IO

Question

我想写一个命名管道(已经创建)而不会阻塞读取器.我的读者是另一个可能会失败的应用程序.如果读者确实失败了,我希望编写器应用程序继续写入该命名管道.像Java这样的东西

fopen(fPath, O_NONBLOCK)

因此,当读者出现时,它可能会从失败的地方恢复.

Answer 1

首先,我试着回答你的问题.接下来,我将尝试向您展示我创建的代码片段,它使用阻止IO来解决您的问题.

你的问题

我想写一个命名管道(已经创建)而不会阻塞读取器

您不需要非阻塞IO来解决您的问题.我认为它甚至无法帮助您解决问题.阻止IO也将运行良好(可能甚至比非阻塞IO更好,因为并发性低).加号阻止IO更容易编程.您的读者可以/应该保持阻止.

我的读者是另一个可能会失败的应用程序.如果读者确实失败了,我希望编写器应用程序能够写入命名管道.因此,当读者出现时,它可能会从失败的地方恢复.

只是将消息放入阻塞队列中.接下来只有当读者从中读取时才写入命名管道(由于阻塞IO而自动发生).使用阻塞队列时不需要非阻塞文件IO.当读者正在阅读时,数据是从阻塞队列异步传递的,这会将您的数据从您的编写器发送到阅读器.

类似于Java中的fopen(fPath,O_NONBLOCK)

您在阅读器上不需要非阻塞IO,即使您使用它也是如此.只使用阻塞IO.

代码链

创建了一个小片段,我相信它可以展示您的需求.

组件:

• Writer.java:从控制台读取行作为示例.当您启动程序时,输入文本,然后输入,将其发送到您的命名管道.如有必要,作者将继续写作.
• Reader.java:读取从命名管道(Writer.java)写入的行.
• 命名管道:我假设您在同一目录中创建了一个名为"pipe"的管道.

Writer.java

import java.io.BufferedWriter;
import java.io.Console;
import java.io.FileWriter;
import java.io.PrintWriter;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.logging.Level;
import java.util.logging.Logger;

public class Writer {
    private final BlockingDeque<StringBuffer> queue;
    private final String filename;

    public static void main(String[] args) throws Exception {
        final Console console = System.console();
        final Writer writer = new Writer("pipe");

        writer.init();

        while(true) {
            String readLine = console.readLine();
            writer.write(new StringBuffer(readLine));
        }
    }

    public Writer(final String filename){
        this.queue = new LinkedBlockingDeque<StringBuffer>();
        this.filename = filename;
    }

    public void write(StringBuffer buf) {
        queue.add(buf);
    }

    public void init() {
        ExecutorService single = Executors.newSingleThreadExecutor();

        Runnable runnable = new Runnable() {
            public void run() {
                while(true) {
                    PrintWriter w = null;
                    try {
                        String toString = queue.take().toString();
                        w = new PrintWriter(new BufferedWriter(new FileWriter(filename)), true);
                        w.println(toString);
                    } catch (Exception ex) {
                        Logger.getLogger(Writer.class.getName()).log(Level.SEVERE, null, ex);
                    }
                }
            }
        };

        single.submit(runnable);
    }
}

Reader.java

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;

public class Reader {
    private final BufferedReader br;

    public Reader(final String filename) throws FileNotFoundException {
        br = new BufferedReader(new FileReader(filename));
    }

    public String readLine() throws IOException {
        return br.readLine();
    }

    public void close() {
        try {
            br.close();
        } catch (IOException ex) {
            Logger.getLogger(Reader.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    public static void main(String[] args) throws FileNotFoundException {
        Reader reader = new Reader("pipe");
        while(true) {
            try {
                String readLine = reader.readLine();
                System.out.println("readLine = " + readLine);
            } catch (IOException ex) {
                reader.close();
                break;
            }
        }
    }
}

Answer 2

如果您希望管道保持活动状态并对消息进行排队，您可能需要消息系统而不是原始管道。在 Java 中，标准 API 称为“Java 消息系统”( JMS )，并且有许多标准实现——我见过的最常见的是 Apache ActiveMQ。如果您想要一个跨平台、类似套接字的接口来进行缓冲和恢复，我可能会建议0MQ，它虽然不是“纯 Java”，但具有许多语言的绑定和出色的性能。