1

複数のスレッドを使用して jms キューを消費しようとしています。スレッドごとに個別の JMS セッションが必要であり、以下に示すようにコードで行ったことを知っています。しかし、奇妙な例外が発生しています

例外スタック トレースは次のとおりです。

javax.jms.IllegalStateException: Forbidden call on a closed connection.
    at org.objectweb.joram.client.jms.Connection.checkClosed(Connection.java:404)
    at org.objectweb.joram.client.jms.Connection.createSession(Connection.java:530)
    at MessageWorker.run(ReceiveJmsDemoMultiThreaded.java:96)
    at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)

これは私にとってブロックの問題であるため、あなたの助けが必要です

import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;   
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.QueueConnectionFactory;
import javax.jms.Session;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

public class ReceiveJmsDemoMultiThreaded {

public static void main(String[] args) {
    Context context = null;
    ConnectionFactory factory = null;
    Connection connection = null;
    Destination destination = null;

    try {
        context = getInitialContext();
        factory = (QueueConnectionFactory) context.lookup("JQCF");
        destination = (Destination) context.lookup("sampleQueue");
        connection = factory.createConnection();

        final ExecutorService executor = Executors.newCachedThreadPool();
        executor.submit(new MessageWorker(connection, destination) );

        executor.submit(new MessageWorker(connection, destination) );

        executor.submit(new MessageWorker(connection, destination) );

        executor.submit(new MessageWorker(connection, destination) );

        connection.start();

    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        if (context != null) {
            try {
                context.close();
            } catch (NamingException e) {
                e.printStackTrace();
            }
        }

        if (connection != null) {
            try {
                connection.close();
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }
}

private static InitialContext getInitialContext() throws NamingException {
    Properties prop = new Properties();
    prop.put("java.naming.provider.url", "rmi://localhost:1099");
    prop.put("java.naming.factory.initial",
               "org.objectweb.carol.jndi.spi.MultiOrbInitialContextFactory");
    return new InitialContext(prop);
}

}

class MessageWorker extends Thread {
Connection connection = null;
Destination dest = null;
Session session = null;
Destination destination = null;

public MessageWorker(Connection connection, Destination dest) {
    this.connection = connection;
    this.destination = dest;
}
@Override
public void run() {
    try {
        MessageConsumer receiver = null;
        System.out.println("Starting Thread "+currentThread().getName());
        while (true) {
            try {
                System.out.println("Waiting for next msg "+currentThread().getName());
                session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                receiver = session.createConsumer(destination);
                Message msg = receiver.receive();
                if (msg instanceof Message && msg != null) {
                    System.out.println("STARTING consuming "+msg.toString()+" by thread  "+currentThread().getName() );
                    Thread.sleep(2000);//some work here
                    System.out.println("ENDING consuming "+msg.toString()+" by thread  "+currentThread().getName() );
                }
            } catch (JMSException e) {

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

    }
}
}

どうもありがとう

4

1 に答える 1

2

この問題が発生するのは、メイン スレッドでジョブを Executor Service に送信した後、以下を使用して接続が閉じられるためです。

        connection.close();

そのため、スレッドがこの共有接続 (クローズされたばかり) を使用してセッションを作成しようとすると、この例外が発生します。ここで予期しないことは何もありません。テストのためだけに、すべてのスレッドがメッセージの受信を完了するまで、メイン スレッドを長時間スリープ状態にすることができます。このようにして、この例外を受け取っていないことを確認できます。

実際の解決策は、Executor サービスをシャットダウンしawaitTermination()、サブミットされたジョブの完了をメイン スレッドで待機させることです。

于 2013-06-05T13:50:19.280 に答える