テキストファイルの非同期処理

1. ファイルを分割して読む: ByteBuffer とエンコーディング

今日では小さなテキストファイルを扱うことはまれです。通常は巨大なサーバーログ、レポート、CSVファイル、あるいはギガバイト級のダンプです。だからこそ、ただ読むだけでなく、アプリをフリーズさせずに効率よく読み込むことが重要です。

非同期アプローチはまさにそのためのものです。UIでもサーバー側ロジックでもメインスレッドをブロックせず、大量データの読み書きを並行して行え、複数ファイルを同時に扱う必要があるときにアプリのスケーラビリティを高めます。

重要なのは、非同期I/Oがディスク自体を速くするわけではない、という点です—魔法はありません。ディスクの処理完了を待っているあいだに、プログラムが手持ち無沙汰にならず他の作業を進められるようにするだけです。

非同期読み込みはどのように動く？

非同期チャネル（AsynchronousFileChannel）は文字列ではなく、ByteBuffer というオブジェクトにバイトのブロックを読み込みます。これは、単語ではなく文字が入った箱を運ぶようなものです。読み込んだ後は、これらのバイトを文字列に変換する必要があります—エンコーディングを考慮して！

例: ファイルをブロック単位で非同期読み込み

4096 バイトのブロックで非同期に読み取り、内容をコンソールに出力する最小例を書いてみます。

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.concurrent.Future;
import java.nio.charset.StandardCharsets;
import java.io.IOException;

public class AsyncTextReadExample {
    public static void main(String[] args) throws Exception {
        Path path = Path.of("bigfile.txt");
        try (AsynchronousFileChannel channel = AsynchronousFileChannel.open(path, StandardOpenOption.READ)) {
            ByteBuffer buffer = ByteBuffer.allocate(4096);
            int position = 0;
            Future<Integer> future = channel.read(buffer, position);

            while (future.get() > 0) {
                buffer.flip();
                // バイトを文字列に変換する (UTF-8)
                String chunk = StandardCharsets.UTF_8.decode(buffer).toString();
                System.out.print(chunk);
                buffer.clear();
                position += chunk.getBytes(StandardCharsets.UTF_8).length;
                future = channel.read(buffer, position);
            }
        }
    }
}

重要なポイント:

• ファイルは全体ではなく（バッファ）部分ごとに読みます。
• 読み込んだバイトは Charset を使って文字列にデコードします。
• buffer.clear() を忘れない—忘れると次の read が動作しません！

なぜ単にバイトをデコードするだけでは不十分なのか？

問題は、特にマルチバイトエンコーディング（例: "UTF-8"）では、1つの文字がブロックの境界で分断されうることです。バッファの最後のバイトが1文字の半分だった場合、次のブロックはその「残り」から始まります。特別な処理をしないと、文字化けやデコードエラーになります。

2. バイトから文字列への変換: 分断への対処

分断の問題

例えば、文字列 "こんにちは\n世界\n" があり、バッファの終端が "こん" で終わり、"にちは\n世界\n" が次のブロックに入ってしまったとします。単純に文字列をつなげるだけでは、文字を失ったり不正な文字列になったりします。

解決策: CharsetDecoder を使う

Java には、こうしたケースを正しく扱える CharsetDecoder クラスがあります。デコードされなかったバイトを「覚えて」おき、ブロックの継ぎ目で文字を正しく復元します。

CharsetDecoder の使用例

import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.nio.CharBuffer;
import java.nio.ByteBuffer;

CharsetDecoder decoder = Charset.forName("UTF-8").newDecoder();
ByteBuffer buffer = ... // 読み込んだバイト
CharBuffer charBuffer = CharBuffer.allocate(buffer.capacity());
decoder.decode(buffer, charBuffer, false);
// これで charBuffer には正しくデコードされた文字が入っています

実際の処理では、読み込みの合間に「残り」を保持し、それを考慮してデコードします。

3. テキストファイルの非同期書き込み

読み込みは半分に過ぎません。書き込みもバイトのブロック単位で行われ、まず文字列をエンコードしてバイトにする必要があります。

例: 文字列をファイルに非同期で書き込む

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.concurrent.Future;
import java.nio.charset.StandardCharsets;

public class AsyncTextWriteExample {
    public static void main(String[] args) throws Exception {
        Path path = Path.of("output.txt");
        String text = "こんにちは、世界!\n";
        ByteBuffer buffer = ByteBuffer.wrap(text.getBytes(StandardCharsets.UTF_8));
        try (AsynchronousFileChannel channel = AsynchronousFileChannel.open(path, StandardOpenOption.WRITE, StandardOpenOption.CREATE)) {
            Future<Integer> future = channel.write(buffer, 0);
            // デモ目的で完了を待つ（通常は不要！）
            future.get();
            System.out.println("データは非同期に書き込まれました。");
        }
    }
}

コメント: 実際の非同期シナリオでは、メインスレッドで future.get() を呼ぶべきではありません。非同期コードを同期的にしてしまいます。代わりに CompletionHandler を使いましょう（前回の講義を参照）。

4. 実践: 大きなテキストファイルを非同期に読み取り、行数を数える

実践課題として、大きなテキストファイルを非同期に読み取り、"\n" の数（行数）を数え、結果をコンソールに出力してみましょう。

CompletionHandler を使った例

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.channels.CompletionHandler;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.nio.charset.CharsetDecoder;
import java.nio.charset.StandardCharsets;
import java.nio.CharBuffer;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicLong;

public class AsyncLineCounter {
    public static void main(String[] args) throws IOException {
        Path path = Path.of("bigfile.txt");
        AsynchronousFileChannel channel = AsynchronousFileChannel.open(path, StandardOpenOption.READ);

        ByteBuffer buffer = ByteBuffer.allocate(4096);
        AtomicLong position = new AtomicLong(0);
        CharsetDecoder decoder = StandardCharsets.UTF_8.newDecoder();
        StringBuilder leftover = new StringBuilder();
        AtomicLong lines = new AtomicLong(0);

        channel.read(buffer, position.get(), null, new CompletionHandler<Integer, Object>() {
            @Override
            public void completed(Integer result, Object attachment) {
                if (result == -1) {
                    // ファイルの末尾に到達
                    if (leftover.length() > 0) lines.incrementAndGet();
                    System.out.println("ファイルの行数: " + lines.get());
                    try { channel.close(); } catch (IOException e) { e.printStackTrace(); }
                    return;
                }
                buffer.flip();
                CharBuffer charBuffer = CharBuffer.allocate(buffer.remaining());
                decoder.decode(buffer, charBuffer, false);
                charBuffer.flip();
                String chunk = leftover.toString() + charBuffer.toString();
                leftover.setLength(0);

                // 行数を数える
                int last = 0;
                int idx;
                while ((idx = chunk.indexOf('\n', last)) != -1) {
                    lines.incrementAndGet();
                    last = idx + 1;
                }
                // 残り（最後の \n 以降の部分）
                if (last < chunk.length()) {
                    leftover.append(chunk.substring(last));
                }
                buffer.clear();
                position.addAndGet(result);
                channel.read(buffer, position.get(), null, this);
            }

            @Override
            public void failed(Throwable exc, Object attachment) {
                System.err.println("読み取りエラー: " + exc.getMessage());
                try { channel.close(); } catch (IOException e) { e.printStackTrace(); }
            }
        });

        // プログラムがすぐ終了しないようにする（あくまで例！）
        try { Thread.sleep(2000); } catch (InterruptedException e) {}
    }
}

• CompletionHandler を使って本当に非同期なコードにしています。
• 各読み込み後に、バッファを CharsetDecoder でデコードします。
• "\n" で終わらなかった行の残りは、次のブロックへ持ち越します。
• ファイルの終わりで leftover に何か残っていれば、それも1行として数えます。
• 簡単のため、この例では 2000 ms スリープして非同期処理の完了を待っています（実アプリでは不要—通常はメインループやUIがあります）。

5. 結果をファイルへ非同期に書き出す

たとえば、結果（行数など）を新しいファイルへ—非同期で—書き出したいとします。

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.channels.CompletionHandler;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.nio.charset.StandardCharsets;
import java.io.IOException;

public class AsyncWriteResult {
    public static void main(String[] args) throws IOException {
        String result = "ファイルの行数: 12345\n";
        ByteBuffer buffer = ByteBuffer.wrap(result.getBytes(StandardCharsets.UTF_8));
        Path path = Path.of("result.txt");

        AsynchronousFileChannel channel = AsynchronousFileChannel.open(
            path, StandardOpenOption.WRITE, StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING);

        channel.write(buffer, 0, null, new CompletionHandler<Integer, Object>() {
            @Override
            public void completed(Integer written, Object attachment) {
                System.out.println("結果を非同期で書き込みました！");
                try { channel.close(); } catch (IOException e) { e.printStackTrace(); }
            }

            @Override
            public void failed(Throwable exc, Object attachment) {
                System.err.println("書き込みエラー: " + exc.getMessage());
                try { channel.close(); } catch (IOException e) { e.printStackTrace(); }
            }
        });

        try { Thread.sleep(500); } catch (InterruptedException e) {}
    }
}

6. 部分データとエンコーディング処理のヒント

ブロック間にまたがる部分的な行

行が2つのブロックに分割されている場合、バイトを手作業で「結合」しようとしないでください！ CharsetDecoder を使えば、不足しているバイトを丁寧に扱い、1文字も失わずに処理できます。

さまざまなエンコーディングの扱い

"UTF-8" は現代のアプリでの標準ですが、ファイルが別のエンコーディング（例えば "Windows-1251" や "UTF-16"）である場合は、対応する Charset を使用します:

import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;

Charset charset = Charset.forName("Windows-1251");
CharsetDecoder decoder = charset.newDecoder();

CharsetDecoder と CharsetEncoder の活用

データを部分的に読み書きする場合、エンコーディングの扱いを正しく行うことが重要です。文字がブロック間で「裂ける」ことがあり、追加処理がないとバイトのごちゃ混ぜになります。

これを避けるために、CharsetDecoder と CharsetEncoder を使います。

読み込み時は decode(ByteBuffer, CharBuffer, endOfInput) を、書き込み時は encode(CharBuffer, ByteBuffer, endOfInput) を呼び出します。

これらは、たとえ文字が2つのブロックに分断されていても、正しく組み立てて処理してくれます。

7. テキストファイルを非同期処理するときの典型的なミス

エラー No.1: 行の残りを無視する。 ブロック間で行の「尻尾」を保持しないと、一部の行を失ったり、正しくデコードできなかったりします.

エラー No.2: バッファの扱いの誤り。 処理後に buffer.clear() を呼び忘れると、次の read が動かない、あるいはデータがおかしくなります。

エラー No.3: 不適切なエンコーディングの使用。 ファイルの書き込み時と異なる Charset でデコードすると、文字化けやエラーが発生する可能性があります。

エラー No.4: メインスレッドのブロッキング。 UIスレッドで future.get() や Thread.sleep() を呼ぶと、非同期性の意味が失われます。CompletionHandler やリアクティブなアプローチを使いましょう。

エラー No.5: 完了後にチャネルを閉じない。 すべての操作が終わったら、エラーがあった場合でも必ず（channel.close() で）チャネルを閉じてください。