2

私はjmsが初めてです。目標は、リスナー インスタンスを複数のコンシューマーにアタッチし、各コンシューマーが独自のセッションを使用して別のスレッドで実行することにより、非同期リスナーの onMessage メソッドでキューからメッセージを同時に処理することです。同時処理。

1) 複数のコンシューマーを作成することにより、単一のキューから同時にメッセージを処理することは可能ですか? 2)私は以下のコードを思いつきましたが、以下のコードが私が達成したいことに対して正しいように見えるかどうかについてあなたの考えを聞きたいです.

public class QueueConsumer implements Runnable, MessageListener {

public static void main(String[] args) {




    QueueConsumer consumer1 = new QueueConsumer();
    QueueConsumer consumer2 = new QueueConsumer();
    try {
        consumer1.init("oms", "US.Q.CHECKOUT-ORDER.1.0.JSON");
        consumer2.init("oms","US.Q.CHECKOUT-ORDER.1.0.JSON");
    } catch (JMSException ex) {
        ex.printStackTrace();
        System.exit(-1);
    }


    Thread newThread1 = new Thread(consumer1);
    Thread newThread2 = new Thread(consumer1);
    newThread1.start();
    newThread2.start();



}


private static String connectionFactoryName = null;
private static String queueName = null;


private static ConnectionFactory qcf = null;
private static Connection queueConnection = null;


private Session ses = null;
private Destination queue = null;
private MessageConsumer msgConsumer = null;

public static final Logger logger = LoggerFactory
        .getLogger(QueueConsumer.class);

public QueueConsumer() {
    super();
}

public void onMessage(Message msg) {
    if (msg instanceof TextMessage) {
        try {

            //process message

        } catch (JMSException ex) {
            ex.printStackTrace();

        }
    }

}

public void run() {

    try {
        queueConnection.start();
    } catch (JMSException e) {

        e.printStackTrace();

        System.exit(-1);
    }
    while (!Thread.currentThread().isInterrupted()) {
        synchronized (this) {
            try {
                wait();
            } catch (InterruptedException ex) {
                break;
            }
        }
    }

}



public void init(String factoryName, String queue2) throws JMSException {
    try {

        qcf = new JMSConnectionFactory(factoryName);


        queueConnection = qcf.createConnection();


        ses = queueConnection.createSession(false,
                Session.CLIENT_ACKNOWLEDGE);
        queue = ses.createQueue(queue2);
        logger.info("Subscribing to destination: " + queue2);

        msgConsumer = ses.createConsumer(queue);


        msgConsumer.setMessageListener(this);

        System.out.println("Listening on queue " + queue2);

    } catch (Exception e) {
        e.printStackTrace();
        System.exit(-1);
    }

}

private static void setConnectionFactoryName(String name) {
    connectionFactoryName = name;
}

private static String getQueueName() {
    return queueName;
}

private static void setQueueName(String name) {
    queueName = name;
}

}

4

3 に答える 3

2
  1. そのとおり
  2. 簡単に見ただけで、間違ったコンシューマーを 2 番目のスレッドに渡していることに気付きました。

    Thread newThread2 = new Thread(consumer1); // has to pass consumer2
    

    これに加えて、 などの一部の変数ConnectionFactoryは静的であり、複数回初期化/オーバーライドされます。複数のセッションやコンシューマーを作成できる接続が 1 つだけ必要です。

于 2013-06-13T07:01:09.020 に答える
0

あなたが提供したコード例に関連して、デプロイされたアプリケーションで低レベルのスレッドを作成することはオラクルによって推奨されていません。Weblogic の例 : WebLogic Server でのスレッドの使用

于 2015-04-08T15:50:01.347 に答える