【Java】簡単なメッセージキューを作ってみる

Last updated at 2025-12-29Posted at 2025-12-29

今回は、Javaを使って簡単なメッセージキューコンソールアプリを作ってみます。

メッセージキューシステムとは
メッセージキューシステムとは、異なるアプリケーション間でデータを非同期にやり取りするための仕組みで、メッセージ（データ）を「キュー（待ち行列）」に一時保存し、受信側が都合の良い時に取り出して処理します。
これにより、送信側と受信側が同時に稼働していなくてもよく、システム間の結合度を下げ、耐障害性や拡張性を高めることができ、マイクロサービスや分散システムで広く利用されています。
主な特徴と仕組み
非同期通信: 送信側（プロデューサー）はメッセージをキューに送るだけで、受信側（コンシューマー）の処理完了を待つ必要がありません。
疎結合: アプリケーション同士が直接つながる必要がなくなり、独立性が高まります。
キュー（待ち行列）: メッセージは到着順（FIFO: First-In, First-Out）に並べられ、処理されるまでキューに保管されます。
耐障害性・信頼性: 受信側が一時的にダウンしてもメッセージはキューに残り、失われません。復旧後に処理を再開できます。
負荷分散: 急激なアクセス（スパイク）をキューが吸収し、バッチ処理や重い処理を平滑化します。
独占配送: 基本的に1つのメッセージは1つの受信者によって一度だけ処理されます。

具体的な利用例
タスク管理: ユーザーからのリクエスト（メール送信、画像処理など）をキューに入れ、バックグラウンドで処理する。
イベント処理: 複数のマイクロサービスがイベント（注文完了など）をキューを通じて連携する。
システム連携: 異なるシステム間でデータを連携し、データの欠損を防ぐ。

代表的なメッセージキュー製品
■Amazon SQS (Simple Queue Service) (AWS)
■RabbitMQ
■Apache ActiveMQ
■Kafka (メッセージキューとは異なるが、類似の用途で使われる)

下記がコンソールアプリになります。

sample.java

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ArrayBlockingQueue;

public class Main {
    // メッセージキュー (スレッドセーフ)
    private static final BlockingQueue<String> messageQueue = new ArrayBlockingQueue<>(10);

    // メッセージ送信メソッド
    public static void sendMessage(String message) throws InterruptedException {
        messageQueue.put(message); // キューが満杯の場合はブロックする
        System.out.println("Sent message: " + message);
    }

    // メッセージ受信メソッド
    public static String receiveMessage() throws InterruptedException {
        String message = messageQueue.take(); // キューが空の場合はブロックする
        System.out.println("Received message: " + message);
        return message;
    }

    public static void main(String[] args) throws InterruptedException {
        // 送信者スレッド
        new Thread(() -> {
            try {
                sendMessage("Hello");
                sendMessage("World");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        // 受信者スレッド
        new Thread(() -> {
            try {
                receiveMessage();
                receiveMessage();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }
}