P2P (Point-to-Point) モード
キュー プロデューサー
P2P モードでは、メッセージはキューに送信され、キュー内の各メッセージは 1 つのコンシューマーによってのみ消費されます。以下にキュー プロデューサーのサンプルコードを示します。
package com.activemq.messaging.p2p;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class QueueProducer {
private static final String BROKER_URL = "tcp://localhost:61616";
private static final String QUEUE_NAME = "myQueue";
public static void main(String[] args) {
Connection connection = null;
Session session = null;
MessageProducer producer = null;
try {
// 1. ConnectionFactory の取得
ConnectionFactory factory = new ActiveMQConnectionFactory(BROKER_URL);
// 2. Connection の作成
connection = factory.createConnection();
connection.start(); // 接続の開始
// 3. Session の作成 (トランザクションなし、自動 ACK)
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 4. Destination (Queue) の作成
Destination queue = session.createQueue(QUEUE_NAME);
// 5. MessageProducer の作成
producer = session.createProducer(queue);
producer.setDeliveryMode(DeliveryMode.PERSISTENT); // メッセージ永続化
// 6. メッセージの送信
for (int i = 1; i <= 5; i++) {
TextMessage message = session.createTextMessage("P2P メッセージ: " + i);
producer.send(message);
System.out.println("送信: " + message.getText());
}
} catch (JMSException e) {
System.err.println("JMS エラーが発生しました: " + e.getMessage());
e.printStackTrace();
} finally {
// 7. リソースのクリーンアップ
closeResources(producer, session, connection);
}
}
private static void closeResources(MessageProducer producer, Session session, Connection connection) {
try {
if (producer != null) producer.close();
if (session != null) session.close();
if (connection != null) connection.close();
} catch (JMSException e) {
System.err.println("リソースクローズ中に JMS エラーが発生しました: " + e.getMessage());
}
}
}
キュー コンシューマー 1
キュー コンシューマーは、キューからメッセージを受信して処理します。以下は、ポーリング方式でメッセージを受信するコンシューマーの例です。
package com.activemq.messaging.p2p;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
import java.io.IOException;
public class QueueConsumer1 {
private static final String BROKER_URL = "tcp://localhost:61616";
private static final String QUEUE_NAME = "myQueue";
public static void main(String[] args) {
Connection connection = null;
Session session = null;
MessageConsumer consumer = null;
try {
// 1. ConnectionFactory の取得
ConnectionFactory factory = new ActiveMQConnectionFactory(BROKER_URL);
// 2. Connection の作成
connection = factory.createConnection();
connection.start(); // 接続の開始
// 3. Session の作成
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 4. Destination (Queue) の作成
Destination queue = session.createQueue(QUEUE_NAME);
// 5. MessageConsumer の作成
consumer = session.createConsumer(queue);
// 6. メッセージの受信 (ポーリング)
System.out.println("コンシューマー 1: メッセージ受信待機中...");
while (true) {
// receive() はブロックするメソッド
TextMessage message = (TextMessage) consumer.receive(1000); // タイムアウト付きで待機
if (message != null) {
System.out.println("コンシューマー 1 受信: " + message.getText());
} else {
System.out.println("コンシューマー 1: タイムアウト。メッセージがありません。");
// 必要に応じてループを抜ける条件を追加
// break;
}
}
} catch (JMSException e) {
System.err.println("JMS エラーが発生しました: " + e.getMessage());
e.printStackTrace();
} finally {
// 7. リソースのクリーンアップ
closeResources(consumer, session, connection);
}
}
private static void closeResources(MessageConsumer consumer, Session session, Connection connection) {
try {
if (consumer != null) consumer.close();
if (session != null) session.close();
if (connection != null) connection.close();
} catch (JMSException e) {
System.err.println("リソースクローズ中に JMS エラーが発生しました: " + e.getMessage());
}
}
}
キュー コンシューマー 2
別のコンシューマーインスタンスを作成し、同じキューを購読することで、メッセージはコンシューマー間で分散されます。
package com.activemq.messaging.p2p;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class QueueConsumer2 {
private static final String BROKER_URL = "tcp://localhost:61616";
private static final String QUEUE_NAME = "myQueue";
public static void main(String[] args) {
Connection connection = null;
Session session = null;
MessageConsumer consumer = null;
try {
// 1. ConnectionFactory の取得
ConnectionFactory factory = new ActiveMQConnectionFactory(BROKER_URL);
// 2. Connection の作成
connection = factory.createConnection();
connection.start(); // 接続の開始
// 3. Session の作成
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 4. Destination (Queue) の作成
Destination queue = session.createQueue(QUEUE_NAME);
// 5. MessageConsumer の作成
consumer = session.createConsumer(queue);
// 6. メッセージの受信 (リスナー方式)
System.out.println("コンシューマー 2: メッセージ受信待機中 (リスナー)...");
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
if (message instanceof TextMessage) {
try {
System.out.println("コンシューマー 2 受信: " + ((TextMessage) message).getText());
} catch (JMSException e) {
System.err.println("メッセージ処理中に JMS エラーが発生しました: " + e.getMessage());
}
}
}
});
// アプリケーションを継続させるために、標準入力を待機
System.out.println("Enter キーを押すと終了します...");
System.in.read();
} catch (JMSException e) {
System.err.println("JMS エラーが発生しました: " + e.getMessage());
e.printStackTrace();
} catch (IOException e) {
System.err.println("I/O エラーが発生しました: " + e.getMessage());
} finally {
// 7. リソースのクリーンアップ
closeResources(consumer, session, connection);
}
}
private static void closeResources(MessageConsumer consumer, Session session, Connection connection) {
try {
if (consumer != null) consumer.close();
if (session != null) session.close();
if (connection != null) connection.close();
} catch (JMSException e) {
System.err.println("リソースクローズ中に JMS エラーが発生しました: " + e.getMessage());
}
}
}
Pub/Sub (Publish/Subscribe) モード
トピック プロデューサー
Pub/Sub モードでは、メッセージはトピックに発行され、そのトピックを購読しているすべてのコンシューマーに配信されます。以下にトピック プロデューサーのサンプルコードを示します。
package com.activemq.messaging.pubsub;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class TopicPublisher {
private static final String BROKER_URL = "tcp://localhost:61616";
private static final String TOPIC_NAME = "myTopic";
public static void main(String[] args) {
Connection connection = null;
Session session = null;
MessageProducer publisher = null;
try {
// 1. ConnectionFactory の取得
ConnectionFactory factory = new ActiveMQConnectionFactory(BROKER_URL);
// 2. Connection の作成
connection = factory.createConnection();
connection.start(); // 接続の開始
// 3. Session の作成
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 4. Destination (Topic) の作成
Destination topic = session.createTopic(TOPIC_NAME);
// 5. MessageProducer の作成
publisher = session.createProducer(topic);
publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT); // 非永続メッセージ
// 6. メッセージの発行
for (int i = 1; i <= 5; i++) {
TextMessage message = session.createTextMessage("Pub/Sub メッセージ: " + i);
publisher.send(message);
System.out.println("発行: " + message.getText());
}
} catch (JMSException e) {
System.err.println("JMS エラーが発生しました: " + e.getMessage());
e.printStackTrace();
} finally {
// 7. リソースのクリーンアップ
closeResources(publisher, session, connection);
}
}
private static void closeResources(MessageProducer publisher, Session session, Connection connection) {
try {
if (publisher != null) publisher.close();
if (session != null) session.close();
if (connection != null) connection.close();
} catch (JMSException e) {
System.err.println("リソースクローズ中に JMS エラーが発生しました: " + e.getMessage());
}
}
}
トピック コンシューマー 1
トピック コンシューマーは、指定されたトピックを購読し、発行されたメッセージを受信します。以下は、メッセージ リスナーを使用して非同期にメッセージを受信するコンシューマーの例です。
package com.activemq.messaging.pubsub;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
import java.io.IOException;
public class TopicConsumer1 {
private static final String BROKER_URL = "tcp://localhost:61616";
private static final String TOPIC_NAME = "myTopic";
public static void main(String[] args) {
Connection connection = null;
Session session = null;
MessageConsumer consumer = null;
try {
// 1. ConnectionFactory の取得
ConnectionFactory factory = new ActiveMQConnectionFactory(BROKER_URL);
// 2. Connection の作成
connection = factory.createConnection();
// クライアント ID の設定 (永続サブスクリプションの場合に重要)
connection.setClientID("consumer1");
connection.start(); // 接続の開始
// 3. Session の作成
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 4. Destination (Topic) の作成
Destination topic = session.createTopic(TOPIC_NAME);
// 5. MessageConsumer の作成 (永続サブスクリプションを使用する場合は、createDurableSubscriber を使用)
consumer = session.createConsumer(topic);
// 6. メッセージ受信リスナーの設定
System.out.println("トピック コンシューマー 1: メッセージ受信待機中 (リスナー)...");
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
if (message instanceof TextMessage) {
try {
System.out.println("コンシューマー 1 受信: " + ((TextMessage) message).getText());
} catch (JMSException e) {
System.err.println("メッセージ処理中に JMS エラーが発生しました: " + e.getMessage());
}
}
}
});
// アプリケーションを継続させるために、標準入力を待機
System.out.println("Enter キーを押すと終了します...");
System.in.read();
} catch (JMSException e) {
System.err.println("JMS エラーが発生しました: " + e.getMessage());
e.printStackTrace();
} catch (IOException e) {
System.err.println("I/O エラーが発生しました: " + e.getMessage());
} finally {
// 7. リソースのクリーンアップ
closeResources(consumer, session, connection);
}
}
private static void closeResources(MessageConsumer consumer, Session session, Connection connection) {
try {
if (consumer != null) consumer.close();
if (session != null) session.close();
if (connection != null) connection.close();
} catch (JMSException e) {
System.err.println("リソースクローズ中に JMS エラーが発生しました: " + e.getMessage());
}
}
}
トピック コンシューマー 2
別のコンシューマーインスタンスが同じトピックを購読すると、発行されたメッセージは両方のコンシューマーに配信されます。
package com.activemq.messaging.pubsub;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
import java.io.IOException;
public class TopicConsumer2 {
private static final String BROKER_URL = "tcp://localhost:61616";
private static final String TOPIC_NAME = "myTopic";
public static void main(String[] args) {
Connection connection = null;
Session session = null;
MessageConsumer consumer = null;
try {
// 1. ConnectionFactory の取得
ConnectionFactory factory = new ActiveMQConnectionFactory(BROKER_URL);
// 2. Connection の作成
connection = factory.createConnection();
connection.setClientID("consumer2"); // クライアント ID は一意である必要があります
connection.start(); // 接続の開始
// 3. Session の作成
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 4. Destination (Topic) の作成
Destination topic = session.createTopic(TOPIC_NAME);
// 5. MessageConsumer の作成
consumer = session.createConsumer(topic);
// 6. メッセージ受信リスナーの設定
System.out.println("トピック コンシューマー 2: メッセージ受信待機中 (リスナー)...");
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
if (message instanceof TextMessage) {
try {
System.out.println("コンシューマー 2 受信: " + ((TextMessage) message).getText());
} catch (JMSException e) {
System.err.println("メッセージ処理中に JMS エラーが発生しました: " + e.getMessage());
}
}
}
});
// アプリケーションを継続させるために、標準入力を待機
System.out.println("Enter キーを押すと終了します...");
System.in.read();
} catch (JMSException e) {
System.err.println("JMS エラーが発生しました: " + e.getMessage());
e.printStackTrace();
} catch (IOException e) {
System.err.println("I/O エラーが発生しました: " + e.getMessage());
} finally {
// 7. リソースのクリーンアップ
closeResources(consumer, session, connection);
}
}
private static void closeResources(MessageConsumer consumer, Session session, Connection connection) {
try {
if (consumer != null) consumer.close();
if (session != null) session.close();
if (connection != null) connection.close();
} catch (JMSException e) {
System.err.println("リソースクローズ中に JMS エラーが発生しました: " + e.getMessage());
}
}
}