EJB老应用中Singleton Bean的SQS消息监听器使用EntityManager触发JBAS011469事务异常的解决方案咨询

我来帮你拆解下这个问题的核心，然后给出几个落地可行的解决方案：

问题根源 #

你碰到的javax.persistence.TransactionRequiredException，本质是因为JMS监听器的onMessage方法运行在EJB容器的事务上下文之外：

• 你的SqsMessageReceiver是@Dependent作用域Bean，被注入到@Singleton的SqsMessageManager后，生命周期完全绑定到Singleton实例；
• 当SQS消息触发onMessage时，是AWS的JMS客户端线程在执行代码，这个线程不受EJB容器的事务管理，所以EntityManager执行需要事务的操作（比如持久化、更新）时就会报错；
• 而你提到的MessageDrivenBean（MDB）能正常工作，是因为EJB容器会自动为MDB的onMessage方法创建事务上下文，这是MDB专为消息处理设计的核心特性。

解决方案 #

方案1：将SqsMessageReceiver改为Message-Driven Bean（推荐） #

这是最贴合EJB规范的做法，直接依托容器的事务管理能力，无需手动处理线程和事务：

@MessageDriven(
    activationConfig = {
        @ActivationConfigProperty(
            propertyName = "destinationType",
            propertyValue = "javax.jms.Queue"
        ),
        @ActivationConfigProperty(
            propertyName = "destination",
            propertyValue = "nonprod-mbo-sqs-api-hybridcollection-dev-mwcfgsend"
        ),
        // 根据你的AWS SQS配置补充连接工厂等属性
        @ActivationConfigProperty(
            propertyName = "connectionFactoryLookup",
            propertyValue = "java:jboss/yourSQSConnectionFactory"
        )
    }
)
public class SqsMessageReceiverMDB implements MessageListener {

    private static final LoggerAbstract LOG;
    @PersistenceContext(unitName = "meterMgmtPU")
    private EntityManager em;

    static {
        LOG = LoggerAbstract.getInstance();
    }

    private static final String CLASS_NAME = SqsMessageReceiverMDB.class.getName();

    @Override
    public void onMessage(Message message) {
        LOG.info(CLASS_NAME, "Message received - start(), date: "+new Date());
        try {
            if (message instanceof TextMessage) {
                TextMessage txtMsg = (TextMessage) message;
                LOG.debug(CLASS_NAME, "Received Command " + txtMsg);
                String onMessageRequestXML = txtMsg.getText();
                // 直接使用EntityManager，容器自动维护事务
                HouseholdUpdateService hus = new HouseholdUpdateService(em);
                // 执行业务逻辑
            }
        } catch (Exception e) {
            LOG.error( "Failed to Process the request", CLASS_NAME, PrintStackTrace.printStackTrace(e), Propertyloader.getProperty(MMConstants.CCS_ERROR_CODE), "Error occured while processing Request - " + e.getMessage());
            // 抛出异常时，容器会自动回滚事务
        }
        LOG.info(CLASS_NAME, new Date() + " Message received - end() ");
    }
}

同时你可以移除SqsMessageManager中手动注册监听器的代码，MDB会由容器自动绑定到SQS队列并启动监听。

方案2：手动管理事务（适合无法重构为MDB的场景） #

如果必须保留当前的监听器注册方式，可以通过UserTransaction手动控制事务的开启和提交：

@Dependent
public class SqsMessageReceiver implements MessageListener {
    private static final LoggerAbstract LOG;
    @PersistenceContext(unitName = "meterMgmtPU")
    private EntityManager em;
    @Resource
    private UserTransaction ut; // 注入容器的事务管理器

    static {
        LOG = LoggerAbstract.getInstance();
    }

    private static final String CLASS_NAME = SqsMessageReceiver.class.getName();

    @Override
    public void onMessage(Message message) {
        LOG.info(CLASS_NAME, "Message received - start(), date: "+new Date());
        try {
            // 手动开启事务
            ut.begin();
            
            if (message instanceof TextMessage) {
                TextMessage txtMsg = (TextMessage) message;
                String onMessageRequestXML = txtMsg.getText();
                HouseholdUpdateService hus = new HouseholdUpdateService(em);
                // 执行业务逻辑
            }
            
            // 提交事务
            ut.commit();
        } catch (Exception e) {
            LOG.error( "Failed to Process the request", CLASS_NAME, PrintStackTrace.printStackTrace(e), Propertyloader.getProperty(MMConstants.CCS_ERROR_CODE), "Error occured while processing Request - " + e.getMessage());
            // 异常时回滚事务
            try {
                if (ut.getStatus() == Status.STATUS_ACTIVE) {
                    ut.rollback();
                }
            } catch (SystemException ex) {
                LOG.error("Failed to rollback transaction", CLASS_NAME, PrintStackTrace.printStackTrace(ex), "66310", "Rollback error: " + ex.getMessage());
            }
        }
        LOG.info(CLASS_NAME, new Date() + " Message received - end() ");
    }
}

方案3：将业务逻辑委托给事务性EJB #

把涉及数据库操作的逻辑封装到@Stateless的EJB中，利用EJB的自动事务管理：

// 新建事务性业务Bean
@Stateless
public class HouseholdUpdateServiceBean {
    @PersistenceContext(unitName = "meterMgmtPU")
    private EntityManager em;

    public void processHouseholdUpdate(String requestXML) {
        // 这里的操作会自动在容器事务中执行
        HouseholdUpdateService hus = new HouseholdUpdateService(em);
        // 执行具体业务逻辑
    }
}

然后在SqsMessageReceiver中注入并调用这个Bean：

@Dependent
public class SqsMessageReceiver implements MessageListener {
    private static final LoggerAbstract LOG;
    @Inject
    private HouseholdUpdateServiceBean serviceBean; // 注入事务性Bean

    static {
        LOG = LoggerAbstract.getInstance();
    }

    private static final String CLASS_NAME = SqsMessageReceiver.class.getName();

    @Override
    public void onMessage(Message message) {
        LOG.info(CLASS_NAME, "Message received - start(), date: "+new Date());
        try {
            if (message instanceof TextMessage) {
                TextMessage txtMsg = (TextMessage) message;
                String onMessageRequestXML = txtMsg.getText();
                // 调用EJB方法，容器自动管理事务
                serviceBean.processHouseholdUpdate(onMessageRequestXML);
            }
        } catch (Exception e) {
            LOG.error( "Failed to Process the request", CLASS_NAME, PrintStackTrace.printStackTrace(e), Propertyloader.getProperty(MMConstants.CCS_ERROR_CODE), "Error occured while processing Request - " + e.getMessage());
        }
        LOG.info(CLASS_NAME, new Date() + " Message received - end() ");
    }
}

总结 #

优先推荐方案1，它完全遵循EJB的设计思想，让容器接管消息监听和事务管理，减少手动代码的维护成本。如果受限于现有架构无法改用MDB，方案3是次优选择，把事务逻辑封装到EJB中，代码更整洁易维护。方案2适合临时快速修复，但手动管理事务容易出现遗漏，需要格外注意异常处理和回滚逻辑。

内容的提问来源于stack exchange，提问作者Vape