ActiveMQ学習教程(一)——設置と例

8620 ワード

背景：ActiveMQはApacheの製品で、最も流行している、強力なオープンソースメッセージバスです。
JMSの入門学習ミドルウェアとしてActiveMQを選択したのは、以下のような利点があるからです。
1.複数の言語とプロトコルでクライアントを作成する。言語：Java、C、C＋＋、C〓、Ruby、Perl、Python、PHP。アプリケーションプロトコル：OpenWire、Stop REST、WS Notification、XMPP、AMQP
2.JMS 1.1とJ 2 EE 1.4仕様を完全にサポートする（持久化、XAメッセージ、事務）
3.Springへのサポートは、ActiveMQがSpringを使用したシステムに組み込まれやすく、Spring 2.0の特性にも対応しています。
4.JMS 1.1とJ 2 EE 1.4仕様を完全にサポートする（持久化、XAメッセージ、事務）
5.一般的なJ 2 EEサーバ（例えば、Geronimo、JBoss 4、GlassFish、WebLogic）のテストに合格しました。JCA 1.5 resource adatorsの配置により、ActiveMQは任意の互換J 2 EE 1.4ビジネスサーバに自動的に配置することができます。
6.複数の転送プロトコルをサポートする：in-VM、TCP、SSL、NIO、UDP、JGrops、JXTA
7.設計上、高性能なクラスタ、クライアント-サーバ、ポイントマッチングを保証しています。
8.Ajax対応
9.Axisとの整合をサポートする
10.インサートJMS providerを容易に呼び出して、テストを行うことができます。
ActiveMQを習得した後、他のサプライヤーのMQも短時間で速く手に入れることができます。
インストール:
ActiveMQ（本文はMQと略称する）はJDK 1.5以上を要求して、1.6以上のバージョンを推薦します。JDKをインストールしていない友達は、先にインストールしてください。ここで訴えません。
JDKをインストールしてから、
http://activemq.apache.org/download.html MQの最新バージョンをダウンロードします。本教程の使用バージョンは5.5です。
解凍後、MQディレクトリの下に以下のファイルとディレクトリがあります。
activemq-all-5.50.jar:すべてのMQ JARパッケージのセットは、ユーザーシステムの呼び出しに使用されます。
bin:MQの起動スクリプトが含まれています。
conf:MQを含むすべてのプロファイル
data:ログファイルと耐久性メッセージデータ
example:MQの例
lib：MQ運転に必要なすべてのLib
webapps：MQのWebコンソールおよびいくつかの関連DEMO
起動MQ:
binディレクトリのactivemq.batファイルをダブルクリックしてMQを起動します。
最初の例:
JAVAプロジェクトを新設し、activemq-all-5.5.jar、SLFAPI及び対応バージョンLOG 4 JのJARバッグを引用します。
Publisher.java


import java.util.Hashtable;
import java.util.Map;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQMapMessage;

public class Publisher {
	
    protected int MAX_DELTA_PERCENT = 1;
    protected Map<String, Double> LAST_PRICES = new Hashtable<String, Double>();
    protected static int count = 10;
    protected static int total;
    
    protected static String brokerURL = "tcp://localhost:61616";
    protected static transient ConnectionFactory factory;
    protected transient Connection connection;
    protected transient Session session;
    protected transient MessageProducer producer;
    
    public Publisher() throws JMSException {
    	factory = new ActiveMQConnectionFactory(brokerURL);
    	connection = factory.createConnection();
    	try {
        connection.start();
    	} catch (JMSException jmse) {
    		connection.close();
    		throw jmse;
    	}
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        producer = session.createProducer(null);
    }
    
    public void close() throws JMSException {
        if (connection != null) {
            connection.close();
        }
    }
    
    public static void main(String[] args) throws JMSException {
        Publisher publisher = new Publisher();
        while (total < 1000) {
            for (int i = 0; i < count; i++) {
                publisher.sendMessage(args);
            }
            total += count;
            System.out.println("Published '" + count + "' of '" + total + "' price messages");
            try {
              Thread.sleep(1000);
            } catch (InterruptedException x) {
            }
        }
        publisher.close();
    }

    protected void sendMessage(String[] stocks) throws JMSException {
        int idx = 0;
        while (true) {
            idx = (int)Math.round(stocks.length * Math.random());
            if (idx < stocks.length) {
                break;
            }
        }
        String stock = stocks[idx];
        Destination destination = session.createTopic("STOCKS." + stock);
        Message message = createStockMessage(stock, session);
        System.out.println("Sending: " + ((ActiveMQMapMessage)message).getContentMap() + " on destination: " + destination);
        producer.send(destination, message);
    }

    protected Message createStockMessage(String stock, Session session) throws JMSException {
        Double value = LAST_PRICES.get(stock);
        if (value == null) {
            value = new Double(Math.random() * 100);
        }

        // lets mutate the value by some percentage
        double oldPrice = value.doubleValue();
        value = new Double(mutatePrice(oldPrice));
        LAST_PRICES.put(stock, value);
        double price = value.doubleValue();

        double offer = price * 1.001;

        boolean up = (price > oldPrice);

		MapMessage message = session.createMapMessage();
		message.setString("stock", stock);
		message.setDouble("price", price);
		message.setDouble("offer", offer);
		message.setBoolean("up", up);
		return message;
    }

    protected double mutatePrice(double price) {
        double percentChange = (2 * Math.random() * MAX_DELTA_PERCENT) - MAX_DELTA_PERCENT;

        return price * (100 + percentChange) / 100;
    }

}

Consmer.java



import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;

public class Consumer {

    private static String brokerURL = "tcp://localhost:61616";
    private static transient ConnectionFactory factory;
    private transient Connection connection;
    private transient Session session;
    
    public Consumer() throws JMSException {
    	factory = new ActiveMQConnectionFactory(brokerURL);
    	connection = factory.createConnection();
        connection.start();
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    }
    
    public void close() throws JMSException {
        if (connection != null) {
            connection.close();
        }
    }    
    
    public static void main(String[] args) throws JMSException {
    	Consumer consumer = new Consumer();
    	for (String stock : args) {
    		Destination destination = consumer.getSession().createTopic("STOCKS." + stock);
    		MessageConsumer messageConsumer = consumer.getSession().createConsumer(destination);
    		messageConsumer.setMessageListener(new Listener());
    	}
    }
	
	public Session getSession() {
		return session;
	}

}

Listener.java


import java.text.DecimalFormat;

import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageListener;

public class Listener implements MessageListener {

	public void onMessage(Message message) {
		try {
			MapMessage map = (MapMessage)message;
			String stock = map.getString("stock");
			double price = map.getDouble("price");
			double offer = map.getDouble("offer");
			boolean up = map.getBoolean("up");
			DecimalFormat df = new DecimalFormat( "#,###,###,##0.00" );
			System.out.println(stock + "\t" + df.format(price) + "\t" + df.format(offer) + "\t" + (up?"up":"down"));
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}

まずConsmer.javaを実行し、パラメータORCLを入力し、Publisher.javaを実行し、パラメータORCLを入力します。
Publisherがメッセージを送っています。Consmerがメッセージを受信しています。
（ECLIPSEでパラメータを持ってプログラムを実行する方法が分かりませんので、自分でGOOGLEを実行してください。）
はい、MQのインストールと最初の例示的なプログラムの紹介はこれで終わります。
次のセクションでは、JMSのいくつかの概念を紹介して、私たちの後続の学習を支援します。

Spring Bootの公開と実戦（三）日誌フレーム編-日誌システムの集積方法

Resource Bundeleとlocaleの使用