收到消息后关闭 JMS 会话和连接答案

【问题标题】：Close JMS session and connection after message received收到消息后关闭 JMS 会话和连接
【发布时间】：2015-07-21 17:17:08
【问题描述】：

我有一个有状态会话 bean，用于发送和接收 JMS 消息。所有连接设置都是手动处理的，因此 bean 包含 javax.jms.connection 和 javax.jms.session 的实例。该 bean 还实现了 MessageListener 以便能够接收消息。

现在，当我发送消息时，我使用 session.createTemporaryQueue() 创建了一个临时队列。我将 message.setJMSReplyTo() 设置为同一个临时队列，最后创建此队列的消费者，并将 MessageListener 设置为实现所有这些的同一个有状态会话 bean。

我可以通过 onMessage() 方法接收消息。但是，我想在收到消息后立即关闭会话和连接，这显然在 onMessage() 方法中是不允许的。

所以问题是： 收到消息后如何关闭会话和连接？我必须手动处理连接设置，不能使用 MDB。

请注意： 这是在 Java EE 环境（GlassFish 4.0）中执行的

编辑：

import javax.ejb.LocalBean;
import javax.ejb.Stateful;
import javax.inject.Inject;
import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import com.sun.messaging.ConnectionConfiguration;
import com.sun.messaging.QueueConnectionFactory;

@LocalBean
@Stateful
public class OpenMqClient implements MessageListener{
    private Connection connection;
    private Session session;
    private MessageConsumer responseConsumer;

    public OpenMqClient(){}

    public void sendMessage(String messageContent, String jmsBrokerUri, String queueName) {
        try{
            String host = System.getProperty("foo", jmsBrokerUri);
            QueueConnectionFactory cf = new QueueConnectionFactory();
            cf.setProperty(ConnectionConfiguration.imqAddressList, host);
            connection = null;
            session = null;

            //Setup connection
            connection = cf.createConnection();
            connection.start();
            session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);

            //Setup queue and producer
            Queue queue = session.createQueue(queueName);
            MessageProducer producer = session.createProducer(queue);


            //Reply destination
            Queue responseQueue = session.createTemporaryQueue();
            responseConsumer = session.createConsumer(responseQueue);
            responseConsumer.setMessageListener(this);

            //Create message
            TextMessage textMessage = session.createTextMessage();
            textMessage.setJMSReplyTo(responseQueue);
            textMessage.setJMSCorrelationID("test0101");
            textMessage.setText(messageContent);

            producer.send(textMessage);
            System.out.println("Message sent");
        } catch (JMSException e) {
            e.printStackTrace();
            System.out.println("JMSException in Sender");
        }
    }

    @Override
    public void onMessage(Message arg0) {
        //On this event I want to close the session and connection, but it's not permitted
    }

}

【问题讨论】：

会话是如何创建的？什么是事务上下文？
打开/关闭会话和连接是一项昂贵的操作，如果您需要任何类型的性能，我会避免这样做。话虽如此，您最好的选择可能是创建一个新的 Runnable 来接收您的会话和连接并生成一个新线程，该线程将在后台关闭您的会话和连接 - 您甚至可以添加时间延迟，以便连接不t 关闭，直到您的 onMessage 可能完成
创建一次会话和连接怎么样。（例如在@PostConstruct 中）然后销毁它一次（使用@PreDestroy）会话bean？
@ringbearer ：这是一个解决方案，但它不是很有效，因为几乎总是很快收到响应。但是，某些响应可能需要更长的时间，因此我不能使用超时。如果有很多用户在发送消息，就会有很多未使用的打开会话（直到会话 bean 被销毁）。
那么你将不得不坚持@Matt 的建议。尝试从 onMessage() 运行更接近的线程

标签： java jms

【解决方案1】：

就我个人而言，我会这样做（请注意，我尚未对此代码进行测试或添加太多错误处理）。

使连接成为静态 - 您可以（可能应该）为所有 bean 重复使用相同的连接，除非您有特定的理由不这样做

在新线程中关闭会话

public class OpenMqClient implements MessageListener {

    private static Connection connection;
    private static final String mutex = "mutex"; 
    private Session session;
    private MessageConsumer responseConsumer;

    public OpenMqClient() {
        if(connection == null) {
            synchronized(mutex) {
                if(connection == null) {
                    String host = System.getProperty("foo", jmsBrokerUri);
                    QueueConnectionFactory cf = new QueueConnectionFactory();
                    cf.setProperty(ConnectionConfiguration.imqAddressList, host);

                    // Setup connection
                    connection = cf.createConnection();
                    connection.start();
                }
            }
        }
    }

    public void sendMessage(String messageContent, String jmsBrokerUri, String queueName) {
        try {
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            // Setup queue and producer
            Queue queue = session.createQueue(queueName);
            MessageProducer producer = session.createProducer(queue);

            // Reply destination
            Queue responseQueue = session.createTemporaryQueue();
            responseConsumer = session.createConsumer(responseQueue);
            responseConsumer.setMessageListener(this);

            // Create message
            TextMessage textMessage = session.createTextMessage();
            textMessage.setJMSReplyTo(responseQueue);
            textMessage.setJMSCorrelationID("test0101");
            textMessage.setText(messageContent);

            producer.send(textMessage);
            System.out.println("Message sent");
        } catch (JMSException e) {
            e.printStackTrace();
            System.out.println("JMSException in Sender");
        }
    }

    @Override
    public void onMessage(Message arg0) {
        // do stuff
        new Thread(
            new Runnable() {
                @Override
                public void run() {
                    if(session != null)
                        try {
                            session.close();
                        } catch (JMSException e) {
                            e.printStackTrace();
                        }               
                }
            }
        ).start();
    }
}

【讨论】：