使用 jboss EAP 7 发送 JMS（消息）时出错答案

【问题标题】：Error while sending JMS(message) with jboss EAP 7使用 jboss EAP 7 发送 JMS（消息）时出错
【发布时间】：2019-12-14 20:39:48
【问题描述】：

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

javax.ejb.EJBTransactionRolledbackException：生产者已关闭

我有以下无状态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 是接口类，但是当我执行此代码时，我得到了上面的异常，但是如果我将 ExchangeSenderFacadeCore 中的 sendToExchange() 方法移动到 ExchangeSenderFacadeWrapperBean 类，那么错误将消失。谁能告诉我这种情况的确切原因

【问题讨论】：

标签： java jms ejb stateless-session

【解决方案1】：

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

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

此外，JBOSS 服务器也管理无状态会话 bean 池。无状态会话 bean 仅在您完成其目的后才可在连接池上使用，而不是事先。同时连接（新创建的 JMS 或缓存的）用于在无状态会话 bean 中创建 JMS Session (session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE))，也完成了它的目的，并且在 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);
        }
    }

【讨论】：