java - 使用 jboss EAP 7 发送 JMS(消息)时出错

Question

通过 ActiveMQ 使用 Wildfly 和 JMS 我遇到以下异常。

javax.ejb.EJBTransactionRolledbackException: Producer is closed

我有以下无状态 bean

@Stateless(name = "ExchangeSenderFacadeBean")
@Local({ExchangeSenderFacadeLocalI.class})
public class ExchangeSenderFacadeWrapperBean implements ExchangeSenderFacadeLocalI {
    @Resource(lookup = "java:/JmsXA")     // inject ConnectionFactory (more)
    protected ConnectionFactory factory;

    @EJB(beanName = "BeanRegistryLoader")
    protected BeanRegistryLoader omsRegistryBean;

    protected BeanRegistryCore beanRegistryCore;

    @Resource(lookup = "java:/jms/queue/ToExchange")
    protected Queue target;

    private ExchangeSenderFacadeCoreI exchangeSenderFacadeCore;


    @Override
    public void sendToExchange(ExchangeMessage exchangeMessage) {
        exchangeSenderFacadeCore.sendToExchange(exchangeMessage);

    }

    @PostConstruct
    public void init() {
        beanRegistryCore = omsRegistryBean.registry();
        if (exchangeSenderFacadeCore == null) {
            exchangeSenderFacadeCore = ((BeanRegistryCore) omsRegistryBean.registry()).getExchangeSenderFacadeCoreI();
            exchangeSenderFacadeCore.setBeanRegistryCore(omsRegistryBean.registry());
            exchangeSenderFacadeCore.setFactory(factory);
            exchangeSenderFacadeCore.setTargetQueue(target);
        }
    }

}

我使用简单的 java 类创建一个方法，该方法生成一条消息并将其发送到目的地，如下所示

public class ExchangeSenderFacadeCore implements ExchangeSenderFacadeCoreI {
    private static final OMSLogHandlerI logger = new Log4j2HndlAdaptor("ExchangeSenderFacadeCore");
    private BeanRegistryCore beanRegistryCore;
    private ConnectionFactory factory;
    private Connection connection = null;
    private Session session = null;
    private long ttl = 900000;
    protected Queue targetQueue;

    public ExchangeSenderFacadeCore() {
        if (System.getProperty(OMSConst.SYS_PROPERTY_JMS_TTL) != null && System.getProperty(OMSConst.SYS_PROPERTY_JMS_TTL).length() > 0) {
            ttl = Long.parseLong(System.getProperty(OMSConst.SYS_PROPERTY_JMS_TTL));
        }
        logger.info("LN:103", "==JMS Topic TTL:" + ttl);
    }

    @Override
    public void processSendToExchange(ExchangeMessage exchangeMessage) {
        sendToExchange(exchangeMessage);
    }

    public boolean isParallelRunEnabled() {
        Object isParallelRun = beanRegistryCore.getCacheAdaptorI().cacheGet(OMSConst.DEFAULT_TENANCY_CODE, OMSConst.APP_PARAM_IS_PARALLEL_RUN, CACHE_NAMES.SYS_PARAMS_CACHE_CORE);
        if (isParallelRun != null && String.valueOf(isParallelRun).equals(OMSConst.STRING_1)) {
            return true;
        }
        return false;
    }

    @Override
    public void sendToExchange(ExchangeMessage exchangeMessage) {
        MessageProducer producer = null;
        try {
            if (isParallelRunEnabled()) {
                logger.info("LN:66", "== Message send to exchange skipped,due to parallel run enabled");
                return;
            }
            if (connection == null) {
                connection = factory.createConnection();
            }
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            producer = session.createProducer(targetQueue);
            producer.setDisableMessageID(true);
            Message message = beanRegistryCore.getJmsExchangeMsgTransformerI().transformToJMSMessage(session, exchangeMessage);
            producer.send(message);
            producer.setTimeToLive(ttl);//default 15min
            logger.elkLog("78", "-1", LogEventsEnum.SENT_TO_EXCHANGE, exchangeMessage.toString());
        } catch (Exception e) {
            logger.error("LN:80", " Error when sending order to exchange:", e);
            throw new OMSCoreRuntimeException(e.getMessage(), e);
        } finally {
            try {
                if (producer != null)
                    producer.close();
            } catch (JMSException e) {
                logger.error("LN:87", "JMS producer close error:", e);
            }
            try {
                if (session != null)
                    session.close();
            } catch (JMSException e) {
                logger.error("LN:93", "JMS session close error:", e);
            }
        }
    }

    @Override
    public void processSendToExchangeSync(ExchangeMessage exchangeMessage) {

    }

    @Override
    public BeanRegistryCore getBeanRegistryCore() {
        return beanRegistryCore;
    }

    @Override
    public void setBeanRegistryCore(BeanRegistryCore beanRegistryCore) {
        this.beanRegistryCore = beanRegistryCore;
    }

    @Override
    public ConnectionFactory getFactory() {
        return factory;
    }

    @Override
    public void setFactory(ConnectionFactory factory) {
        this.factory = factory;
    }

    @Override
    public Queue getTargetQueue() {
        return targetQueue;
    }

    @Override
    public void setTargetQueue(Queue targetQueue) {
        this.targetQueue = targetQueue;
    }
}

ExchangeSenderFacadeCoreI 是接口(interface)类，但是当我执行此代码时，我得到上述异常，但是如果我将 ExchangeSenderFacadeCore 中的 sendToExchange() 方法移动到 ExchangeSenderFacadeWrapperBean 类，那么错误就会消失。谁能告诉我这种情况的确切原因

Answer 1

经过深入研究这个问题，我发现了这个 https://developer.jboss.org/wiki/ShouldICacheJMSConnectionsAndJMSSessions文章发布在 JBOSS 开发者线程之一上。这清楚地解释了缓存连接和其他 JMS 相关资源是 JMS 代码在 JEE 应用程序服务器中运行的反模式的原因。

简而言之，JCA 层池化 JMS 连接和 JMS session 。所以当你调用createConnection()时或createSession() ，然后，在大多数情况下，它并没有真正调用实际的 JMS 实现来实际创建新的 JMS 连接或 JMS session ，它只是从其自己的内部缓存中返回一个。

此外，JBOSS 服务器也管理无状态 session Bean 池。无状态 session bean 仅在您完成其用途之后才可在连接池上使用，但在此之前则不然。用于创建 JMS Session (session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)) 的同时连接(新创建的 JMS 或缓存)在无状态 session bean 中，也完成了其目的并且在 JCA 层连接池上也可用。因此，如下所示在无状态 EJB 类中调用缓存连接不会给您带来异常，即使 Oracle 不推荐这样做。

public void sendToExchange(ExchangeMessage exchangeMessage) {
        MessageProducer producer = null;
        try {
            if (isParallelRunEnabled()) {
                logger.info("LN:66", "== Message send to exchange skipped,due to parallel run enabled");
                return;
            }
            if (connection == null) {
                connection = factory.createConnection();
            }
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            producer = session.createProducer(targetQueue);
            producer.setDisableMessageID(true);
            Message message = beanRegistryCore.getJmsExchangeMsgTransformerI().transformToJMSMessage(session, exchangeMessage);
            producer.send(message);
            producer.setTimeToLive(ttl);//default 15min
            logger.elkLog("78", "-1", LogEventsEnum.SENT_TO_EXCHANGE, exchangeMessage.toString());
        } catch (Exception e) {
            logger.error("LN:80", " Error when sending order to exchange:", e);
            throw new OMSCoreRuntimeException(e.getMessage(), e);
        } finally {
            try {
                if (producer != null)
                    producer.close();
            } catch (JMSException e) {
                logger.error("LN:87", "JMS producer close error:", e);
            }
            try {
                if (session != null)
                    session.close();
            } catch (JMSException e) {
                logger.error("LN:93", "JMS session close error:", e);
            }
        }
    }

但在这种情况下，由于同一个 POJO 类实例可以在多个场合使用，如下所示。它不保证连接被释放并在JCA层连接池中可用并给出异常。

   @PostConstruct
    public void init() {
        beanRegistryCore = omsRegistryBean.registry();
        if (exchangeSenderFacadeCore == null) {
            exchangeSenderFacadeCore = ((BeanRegistryCore) omsRegistryBean.registry()).getExchangeSenderFacadeCoreI();
            exchangeSenderFacadeCore.setBeanRegistryCore(omsRegistryBean.registry());
            exchangeSenderFacadeCore.setFactory(factory);
            exchangeSenderFacadeCore.setTargetQueue(target);
        }
    }