使用消息驱动 Bean 出错时将消息重新传递到队列

Question

当发生错误时，我需要向队列返回消息。

我正在使用 Weblogic 11G (EJB3.0)、Java 1.6、javaee-api 6.0 上的消息驱动 Bean，队列位于 Oracle 数据库 (AQJMS) 上

我不想将消息发送到错误队列。

public void onMessage(Message message) {
        try {
            this.processMenssage(message);
        } catch (ServiceInternalException e) {
            // I need return the message to the queue because something wrong (anything) happend
        }
}

我知道消息会一次又一次地恢复，直到我解决问题为止。

MDB 配置了注解

@MessageDriven(
    name = "MyBeanMDB",
    activationConfig = {
            @ActivationConfigProperty(propertyName  = "destinationType", 
                    propertyValue = "javax.jms.Queue"),

            @ActivationConfigProperty(propertyName  = "connectionFactoryJndiName",
            propertyValue = "AqJms3FSCF"), // External JNDI Name

            @ActivationConfigProperty(propertyName  = "destinationJndiName",
            propertyValue = "myque/REQ_JMSQ") // Ext. JNDI Name
    }
    )
public class MyBeanMDB implements MessageListener{....

因为我的驱动程序数据源是 oracle.jdbc.xa.client.OracleXADataSource 我有一个文件 weblogic-ejb-jar.xml

<?xml version = '1.0' encoding = 'windows-1252'?>
<weblogic-ejb-jar xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.bea.com/ns/weblogic/weblogic-ejb-jar http://www.bea.com/ns/weblogic/weblogic-ejb-jar/1.0/weblogic-ejb-jar.xsd"
xmlns="http://www.bea.com/ns/weblogic/weblogic-ejb-jar">
<weblogic-enterprise-bean>
    <ejb-name>MyBeanMDB</ejb-name>
    <message-driven-descriptor>
        <pool>
            <max-beans-in-free-pool>1</max-beans-in-free-pool>
            <initial-beans-in-free-pool>1</initial-beans-in-free-pool>
        </pool>
    </message-driven-descriptor>
</weblogic-enterprise-bean>

我尝试使用 setRollbackOnly()

添加

 @TransactionManagement(value=TransactionManagementType.CONTAINER)
 @TransactionAttribute(TransactionAttributeType.REQUIRED)

和

@Override
public void setMessageDrivenContext(MessageDrivenContext ctx) throws EJBException {
    this.myctx = ctx;
}

内部异常

public void onMessage(Message message) {
        try {
            this.processMenssage(message);
        } catch (ServiceInternalException e) {
            this.myctx.setRollbackOnly(); // the message is not redelivered
        }
}

RollbackOnly 被调用，日志显示：

####<Sep 13, 2015 11:40:33 AM ART> <Info> <EJB> <el01cl03> <AdminServer> <[ACTIVE] ExecuteThread: '0' for queue: 'weblogic.kernel.Default (self-tuning)'> <<anonymous>> <> <39d906d31eb822f1:-27a94574:14fc3fbcefe:-8000-0000000000000b8a> <1442155233307> <BEA-010213> <Message-Driven EJB: MyBeanMDB's transaction was rolled back. The transaction details are: Name=NewJMSMessagePoller.MyBeanMDB,Xid=BEA1-2BC2680CDDBA895EF953(294629023),Status=Rolled back. [Reason=weblogic.transaction.internal.AppSetRollbackOnlyException: setRollbackOnly called on transaction],numRepliesOwedMe=0,numRepliesOwedOthers=0,seconds since begin=0,seconds left=60,XAServerResourceInfo[aqjmsuserDS_e2eSOADomain]=(ServerResourceInfo[aqjmsuserDS_e2eSOADomain]=(state=rolledback,assigned=AdminServer),xar=aqjmsuserDS,re-Registered = false),SCInfo[e2eSOADomain+AdminServer]=(state=rolledback),properties=({weblogic.transaction.name=NewJMSMessagePoller.MyBeanMDB}),local properties=({weblogic.jdbc.jta.aqjmsuserDS=[ No XAConnection is attached to this TxInfo ]}),OwnerTransactionManager=ServerTM[ServerCoordinatorDescriptor=(CoordinatorURL=AdminServer+10.10.10.150:7001+e2eSOADomain+t3+, XAResources={eis/tibjms/Queue, NIICommonDS-rac0_e2eSOADomain, eis/activemq/Queue, 10gDataSource_e2eSOADomain, eis/fioranomq/Topic, eis/jbossmq/Queue, eis/Apps/Apps, eis/aqjms/Topic, eis/webspheremq/Queue, eis/AQ/aqSample, eis/tibjms/Topic, eis/aqjms/Queue, eis/aqjms/colasjmsuser3, ITM_e2eSOADomain, eis/sunmq/Queue, WSATGatewayRM_AdminServer_e2eSOADomain, NIICommonDS_e2eSOADomain, eis/jms/ReprocessJMSQueue, eis/tibjmsDirect/Queue, eis/wls/Queue, aqjmsuserDS_e2eSOADomain, eis/tibjmsDirect/Topic, eis/wls/Topic, eis/pramati/Queue, NIICommonDS-rac1_e2eSOADomain, eis/jms/ReprocessJMS11gQueueCF},NonXAResources={})],CoordinatorURL=AdminServer+10.10.10.150:7001+e2eSOADomain+t3+).>

但之后队列中的状态（AQJMS）是3（已处理）

也许问题是：

 aqjmsuserDS=[ No XAConnection is attached to this TxInfo ]

Answer 1

问题是队列的创建，队列是这样创建的

BEGIN DBMS_AQADM.CREATE_QUEUE_TABLE(
 Queue_table        => '"JMSUSER"."SALES_JMSQTAB"',
 Queue_payload_type => 'SYS.AQ$_JMS_MESSAGE',
 storage_clause     => 'PCTFREE 10 PCTUSED 40 INITRANS 1 MAXTRANS 255 TABLESPACE USERS',
 Sort_list          => 'PRIORITY,ENQ_TIME',
 Compatible         => '8.1.3');
END;

BEGIN DBMS_AQADM.CREATE_QUEUE(
 Queue_name          => 'JMSUSER.SALES',
 Queue_table         => 'JMSUSER.SALES_JMSQTAB',
 Queue_type          =>  0,
 Max_retries         =>  0,
 Retry_delay         =>  0,
 Retention_time      =>  86400,
 dependency_tracking =>  FALSE);
END;

问题在于字段 MAX_RETRIES，当为 0 时不允许重试（回滚）。

创建队列的更好方法是：

BEGIN DBMS_AQADM.CREATE_QUEUE_TABLE(
 Queue_table        => 'JMSUSER.SALES_JMSQTAB',
 Queue_payload_type => 'SYS.AQ$_JMS_MESSAGE',
 storage_clause     => 'PCTFREE 10 PCTUSED 40 INITRANS 1 MAXTRANS 255 TABLESPACE USERS',
 Sort_list          => 'PRIORITY,ENQ_TIME',
 Compatible         => '8.1.3');
END;

BEGIN DBMS_AQADM.CREATE_QUEUE(
 Queue_name          => 'JMSUSER.SALES_JMSQTAB',
 Queue_table         => 'JMSUSER.SALES_QTAB',
 Queue_type          =>  0,
 Max_retries         =>  2147483647,
 Retry_delay         =>  5,
 Retention_time      =>  259200,
 dependency_tracking =>  FALSE);
END;

Answer 2

您能否检查以下处理 java.lang.Exception 的代码帮助 同样，它会捕获非 ServiceInternalExceptions。还添加一个 logger 以实际发现此异常已被捕获：

    public void onMessage(Message message) {
     try {
        this.processMenssage(message);
     } catch (ServiceInternalException e) {
        Logger.error(e);
        this.myctx.setRollbackOnly(); 
     } catch (Exception e) {
        Logger.error(e);
        this.myctx.setRollbackOnly();
     }
  }