java - 使用消息驱动 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;