Java API 開発でのメッセージ処理に HornetQ を使用する

Java API 開発におけるメッセージ処理に HornetQ を使用する

インターネットの急速な発展に伴い、大量の情報のやり取りが発生し、メッセージキューは高い同時実行性などの問題を解決する重要な手段となっています。 、高可用性、非同期処理。 HornetQ は、JBoss によって開発された JMS プロトコルに基づく、高性能、高可用性のオープンソース メッセージング ミドルウェアです。この記事では、Java API 開発におけるメッセージ処理に HornetQ を使用する方法を紹介します。

1. クイック スタート

1. HornetQ のダウンロード

HornetQ の公式 Web サイト (http://hornetq.apache.org/downloads.html)複数の形式でパッケージをダウンロードします。ここでは HornetQ-2.4.0.Final-bin.tar.gz を選択します。

2. HornetQ のインストール

ダウンロードが完了したら、HornetQ-2.4.0.Final-bin.tar.gz をローカル フォルダーに解凍します。

3. HornetQ の起動

HornetQ の bin ディレクトリに入り、次のコマンドを実行します:

./run.sh

次のログが表示されます この情報は、HornetQ サービスが正常に開始されたことを示します:

11:14:21,867 INFO [ServerImpl] starting HornetQ Server
11:14:21,986 INFO [JournalStorageManager] Using NIO Journal
11:14:22,626 INFO [NettyAcceptor] Netty Acceptor バージョン #{version}
を開始しました 11:14:22,697 INFO [HornetQServerImpl] HornetQ サーバー バージョン #{version} [${name}] が開始されました

#HornetQ コンソールのデプロイ

HornetQ のhornetq-console.war を Tomcat の webapps ディレクトリに置き、Tomcat を起動し、http://localhost:8080/hornetq-console を通じて HornetQ コンソールにアクセスします。

2. HornetQ の使用

メッセージのパブリッシュと受信

HornetQ のパブリッシュおよびサブスクライブ モデルはトピックに基づいており、パブリッシュ側は特定のトピック メッセージを受信すると、複数の受信者が同時にこのトピックをサブスクライブでき、受信者は複数のパブリッシャーによって公開されたメッセージを受信できます。

(1) メッセージの公開側

まず、メッセージを送信するための公開側 (Publisher) を作成します。コードは次のとおりです。

public class Publisher {

    public static void main(String[] args) throws Exception {

        // 初始化连接工厂等配置信息
        ConnectionFactory connectionFactory = HornetQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, new TransportConfiguration(NettyConnectorFactory.class.getName()));
        JMSContext jmsContext = connectionFactory.createContext();

        // 发送消息
        JMSProducer producer = jmsContext.createProducer();
        Destination destination = HornetQJMSClient.createTopic("exampleTopic");
        producer.send(destination, "Hello, HornetQ!");

        // 关闭连接
        jmsContext.close();
    }
}

(2) メッセージの受信側

メッセージを受信して​​出力する別のレシーバー (サブスクライバー) を作成します。コードは次のとおりです:

public class Subscriber {

    public static void main(String[] args) throws Exception {

        // 初始化连接工厂等配置信息
        ConnectionFactory connectionFactory = HornetQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, new TransportConfiguration(NettyConnectorFactory.class.getName()));
        JMSContext jmsContext = connectionFactory.createContext();

        // 创建消费者
        Destination destination = HornetQJMSClient.createTopic("exampleTopic");
        JMSConsumer consumer = jmsContext.createConsumer(destination);

        // 接收消息并打印
        String message = null;
        do {
            message = consumer.receiveBody(String.class, 1000);
            System.out.println("Received message: " + message);
        } while (message != null);

        // 关闭连接
        jmsContext.close();
    }
}

パブリッシャーとレシーバーを実行した後、パブリッシャーによって送信されたメッセージを表示できます。以下の図に示すように、HornetQ コンソール:

メッセージの永続性

HornetQ はメッセージの永続ストレージをサポートしています。つまり、HornetQ がダウンしている場合でも、メッセージは保存できます。永続化されることが保証されています。

(1) 送信者

# 次のように、メッセージの永続性を DeliveryMode.PERSISTENT に設定する必要があります:

public class Publisher {

    public static void main(String[] args) throws Exception {

        ConnectionFactory connectionFactory = HornetQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, new TransportConfiguration(NettyConnectorFactory.class.getName()));
        JMSContext jmsContext = connectionFactory.createContext();

        // 设定持久性
        JMSProducer producer = jmsContext.createProducer();
        destination = HornetQJMSClient.createTopic("exampleTopic");
        producer.setDeliveryMode(DeliveryMode.PERSISTENT);

        // 发送消息
        producer.send(destination, "Hello, HornetQ!");

        jmsContext.close();
    }
}

(2) 受信者

HornetQ はデフォルトでメッセージを永続化しているため、受信側で特定の設定を実行する必要はなく、前のセクションの Subscriber クラスを引き続き使用するだけです。

クラスター モード

3. HornetQ は高可用性を備えており、メッセージの信頼性と高い同時実行性を確保するためにクラスター モードで実行できます。 HornetQ クラスターモードを実装する手順は次のとおりです。

(1) HornetQ ディレクトリをコピーして新しいフォルダーを作成します

HornetQ ディレクトリをコピーして名前を HornetQ2 に変更し、新しいフォルダーを作成しますクラスターフォルダーという名前のファイルを作成し、HornetQ2 ディレクトリの下にあるすべてのデータディレクトリ、ログディレクトリ、tmp ディレクトリ、およびその他のフォルダーをクラスターフォルダーにコピーします。

(2) 設定ファイルを変更する

HornetQ ディレクトリの下のexamples/configs/clustered 設定フォルダーで、hq-configuration.xml ファイルとserver0 フォルダーとserver1 フォルダーを HornetQ2 ディレクトリにコピーします。を変更し、server0 フォルダー内のhornetq-configuration.xml ファイルを次のように変更します:

(a) ノード名をserver0

に変更します (b) [サーバーの変更] でクラスター接続を変更します。ユーザー名とサーバーパスワードを「guest」

(c) コネクタアドレスをローカル IP アドレス (192.168.1.1

など) に変更します。(d) jms-configuration -ha での使用を変更します。は true に設定されます

以下に示すように:

<configuration xmlns="urn:hornetq"
               xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
               xsi:schemaLocation="urn:hornetq /schema/hornetq-configuration.xsd">
    <cluster-password>guest</cluster-password>
    <paging-directory>${data.dir:../data}/paging</paging-directory>
    <bindings-directory>${data.dir:../data}/bindings</bindings-directory>
    <journal-directory>${data.dir:../data}/journal</journal-directory>
    <large-messages-directory>${data.dir:../data}/large-messages</large-messages-directory>
    <journal-type>NIO</journal-type>
    <journal-datasync>true</journal-datasync>
    <journal-min-files>2</journal-min-files>
    <journal-pool-files>10</journal-pool-files>
    <journal-file-size>10240</journal-file-size>
    <journal-buffer-timeout>28000</journal-buffer-timeout>
    <journal-max-io>1</journal-max-io>
    <disk-scan-period>5000</disk-scan-period>
    <max-disk-usage>90</max-disk-usage>
    <critical-analyzer>true</critical-analyzer>
    <critical-analyzer-timeout>120000</critical-analyzer-timeout>
    <critical-analyzer-check-period>60000</critical-analyzer-check-period>
    <critical-analyzer-policy>HALT</critical-analyzer-policy>
    <page-sync-timeout>1628000</page-sync-timeout>
    <global-max-size>100Mb</global-max-size>
    <connectors>
        <connector name="netty">
            <factory-class>org.hornetq.core.remoting.impl.netty.NettyConnectorFactory</factory-class>
            <param key="host" value="192.168.1.1"/>
            <param key="port" value="5445"/>
        </connector>
    </connectors>
    <acceptors>
        <acceptor name="netty">
            <factory-class>org.hornetq.core.remoting.impl.netty.NettyAcceptorFactory</factory-class>
            <param key="host" value="192.168.1.1"/>
            <param key="port" value="5545"/>
        </acceptor>
    </acceptors>
    <cluster-connections>
        <cluster-connection name="my-cluster">
            <address>jms</address>
            <connector-ref>netty</connector-ref>
            <retry-interval>500</retry-interval>
            <use-duplicate-detection>true</use-duplicate-detection>
            <forward-when-no-consumers>true</forward-when-no-consumers>
            <max-hops>1</max-hops>
            <discovery-group-ref discovery-group-name="my-discovery-group"/>
            <static-connectors>
                <connector-ref>netty</connector-ref>
            </static-connectors>
        </cluster-connection>
    </cluster-connections>
    <ha-policy>
        <replication>
            <slave>
                <allow-failback>true</allow-failback>
                <failback-delay>5000</failback-delay>
                <max-saved-replicated-journals-size>1000000</max-saved-replicated-journals-size>
                <restart-backup>true</restart-backup>
            </slave>
        </replication>
    </ha-policy>
</configuration>

次に、server1 フォルダー内のhornetq-configuration.xml ファイルを同様に変更し、server0 をserver1 に変更します。

(3) HornetQ の起動

HornetQ および HornetQ2 の bin ディレクトリで run.sh コマンドを実行して、HornetQ プロセスを開始します。このとき、2 つの HornetQ ノードはクラスターを形成します。 HornetQ を通じて使用できます。 コンソールを確認してください。

3. 概要

この記事の導入部を通じて、HornetQ の基本的な使用方法とクラスターモードの設定方法について学びました。 HornetQ を使用すると、メッセージ対話の問題を簡単に解決し、システムの堅牢性と同時実行性を向上させることができます。同時に、HornetQ は複数のメッセージング モード、リッチメッセージ永続メカニズム、拡張プラグイン、および実際のニーズに応じて選択および設定できるその他の機能もサポートしています。

以上がJava API 開発でのメッセージ処理に HornetQ を使用するの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。