ActiveMQの送受信サンプルプログラム

3_description.markdown

Apache ActiveMQ

Queue（PTP）とTopic（Pub/Sub）のメッセージ送受信用サンプルプログラム

公式サイト

http://activemq.apache.org/

QueueとTopicの違い

http://activemq.apache.org/how-does-a-queue-compare-to-a-topic.html

build.gradle

apply plugin: 'java'
apply plugin: 'eu.appsatori.fatjar'

repositories {
    jcenter()
}

buildscript {
    repositories {
        jcenter()
    }
    dependencies {
        classpath "eu.appsatori:gradle-fatjar-plugin:0.3"
    }
}

dependencies {
    compile 'org.apache.activemq:activemq-all:5.11.1'
}

ReceiverQueue.java

package com.tasknotes.mq;

import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueReceiver;
import javax.jms.QueueSession;
import javax.jms.TextMessage;

import org.apache.activemq.*;

public class ReceiverQueue {
    public static void main(String[] args) {
        QueueConnection connection = null;
        QueueSession session = null;
        QueueReceiver receiver = null;
        try {
            //Connectionを作成
            QueueConnectionFactory factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_BROKER_URL);
            connection = factory.createQueueConnection();
            connection.start();

            //Receiverの作成
            session = connection.createQueueSession(false,QueueSession.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue("queue_test");
            receiver = session.createReceiver(queue);

            //メッセージの受信
            TextMessage msg = (TextMessage) receiver.receive();
            System.out.println(msg.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        } finally {
            try {
                receiver.close();
                session.close();
                connection.close();
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }
}

ReceiverTopic.java

package com.tasknotes.mq;

import javax.jms.JMSException;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class ReceiverTopic {
    public static void main(String[] args) {
        TopicConnection connection = null;
        TopicSession session = null;
        TopicSubscriber subscriber = null;
        try {
            //Connectionを作成
            TopicConnectionFactory factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_BROKER_URL);
            connection = factory.createTopicConnection();
            connection.start();

            //Subscriberの作成
            session = connection.createTopicSession(false,TopicSession.AUTO_ACKNOWLEDGE);
            Topic topic = session.createTopic("topic_test");
            subscriber= session.createSubscriber(topic);

            //メッセージの受信
            TextMessage msg = (TextMessage) subscriber.receive();
            System.out.println(msg.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        } finally {
            try {
                if (subscriber != null) subscriber.close();
                if (session != null) session.close();
                if (connection != null) connection.close();
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }
}

SenderQueue.java

package com.tasknotes.mq;

import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class SenderQueue {
    public static void main(String[] args){
        QueueConnection connection = null;
        QueueSession session = null;
        QueueSender sender = null;
        try {
            //Connectionを作成
            QueueConnectionFactory factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_BROKER_URL);
            connection = factory.createQueueConnection();
            connection.start();

            //Senderの作成
            session = connection.createQueueSession(false,QueueSession.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue("queue_test");
            sender= session.createSender(queue);

            //メッセージの送信
            TextMessage msg = session.createTextMessage("Hello Message!");
            sender.send(msg);
        } catch (JMSException e) {
            e.printStackTrace();
        } finally {
            try {
                if (sender != null) sender.close();
                if (session != null) session.close();
                if (connection != null) connection.close();
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }
}

SenderTopic.java

package com.tasknotes.mq;

import javax.jms.JMSException;
import javax.jms.QueueSession;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class SenderTopic {
    public static void main(String[] args){
        TopicConnection connection = null;
        TopicSession session = null;
        TopicPublisher publisher = null;
        try {
            //Connectionを作成
            TopicConnectionFactory factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_BROKER_URL);
            connection = factory.createTopicConnection();
            connection.start();

            //Publisherの作成
            session = connection.createTopicSession(false,QueueSession.AUTO_ACKNOWLEDGE);
            Topic topic = session.createTopic("topic_test");
            publisher = session.createPublisher(topic);

            //メッセージの送信
            TextMessage msg = session.createTextMessage("Hello Message!");
            publisher.publish(msg);
        } catch (JMSException e) {
            e.printStackTrace();
        } finally {
            try {
                if (publisher != null) publisher.close();
                if (session != null) session.close();
                if (connection != null) connection.close();
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }
}