JMS Poller Transactional

Question

Ich benutze Spring Integration 4.1.5 und versuche, etwas mit Transaktionen zu tun, aber leider nicht in der Lage, und konnte keine Arbeitsbeispiele finden. Ich versuche einen JMS-Poller einzurichten, der nach Nachrichten sucht. Sobald eine Nachricht empfangen wird, fügt ein Dienstaktivator eine Zeile in die Datenbank ein und die Nachricht wird an einen anderen Dienstaktivator übergeben. Ich möchte die ersten beiden Stücke machen, die Nachrichtenübernahme & Datenbank einfügen Transaktions. Ich möchte nicht, dass der Rest des Flusses transaktional ist. Ich benutze Weblogic als den Anwendungscontainer, also benutze den WebLogicJtaTransactionManager.

Das Problem, in das ich hineinrenne, ist, dass ich die ersten beiden Stücke nicht transaktional machen kann. Es ist entweder alles oder nichts. Ich habe viele Methoden ausprobiert, aber ich denke, dass es am besten ist, eine Ratgeberkette zu benutzen. Ich wäre in der Lage zu kontrollieren, welche Methoden Teil der Transaktion sein würden.

Ich habe Beispiele mit einem Message Driven Listener gesehen, aber ich benutze Weblogic und würde die Work Manager verwenden und ich glaube ich muss einen Poller verwenden, um Arbeitsmanager zu nutzen (wenn das nicht der Fall ist, Ich denke, das ist eine andere Frage für die Zukunft!)

Ich habe die XML und vereinfacht es, aber neben dem Bearbeiten des Paketnamens, erzeugt der Kontext das Problem.

<beans xmlns="http://www.springframework.org/schema/beans" 
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:int="http://www.springframework.org/schema/integration" 
    xmlns:file="http://www.springframework.org/schema/integration/file" 
    xmlns:context="http://www.springframework.org/schema/context" 
    xmlns:int-sftp="http://www.springframework.org/schema/integration/sftp" 
    xmlns:int-xml="http://www.springframework.org/schema/integration/xml" 
    xmlns:int-jms="http://www.springframework.org/schema/integration/jms" 
    xmlns:tx="http://www.springframework.org/schema/tx" xmlns:aop="http://www.springframework.org/schema/aop" 
    xsi:schemaLocation="http://www.springframework.org/schema/beans 
    http://www.springframework.org/schema/beans/spring-beans.xsd 
    http://www.springframework.org/schema/integration 
    http://www.springframework.org/schema/integration/spring-integration.xsd 
    http://www.springframework.org/schema/integration/file 
    http://www.springframework.org/schema/integration/file/spring-integration-file.xsd 
    http://www.springframework.org/schema/integration/sftp 
    http://www.springframework.org/schema/integration/sftp/spring-integration-sftp.xsd 
    http://www.springframework.org/schema/context 
    http://www.springframework.org/schema/context/spring-context.xsd 
    http://www.springframework.org/schema/integration/xml 
    http://www.springframework.org/schema/integration/xml/spring-integration-xml.xsd 
    http://www.springframework.org/schema/integration/jms 
    http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd 
    http://www.springframework.org/schema/tx 
    http://www.springframework.org/schema/tx/spring-tx.xsd 
    http://www.springframework.org/schema/aop 
    http://www.springframework.org/schema/aop/spring-aop.xsd"> 


    <bean id="jtaTransactionManager" 
     class="org.springframework.transaction.jta.WebLogicJtaTransactionManager"> 
     <property name="transactionManagerName" value="javax.transaction.TransactionManager" /> 
    </bean> 

    <bean id="insertMessageToDb" class="com.ReadMsgFromAxway" /> 
    <bean id="serviceActivator" class="com.CreateTMDFile" /> 

    <int-jms:inbound-channel-adapter id="jmsDefaultReceiver" 
     connection-factory="inboundDefaultAdaptorConnectionFactory" 
     extract-payload="false" destination="inboundAdaptorDefaultListenerQueue" 
     channel="inboundJMS" acknowledge="transacted"> 
     <int:poller id="poller" 
      max-messages-per-poll="100" fixed-rate="10"> 
      <int:advice-chain> 
       <ref bean="txAdvice" /> 
      </int:advice-chain> 
     </int:poller> 
    </int-jms:inbound-channel-adapter> 


    <tx:advice id="txAdvice" transaction-manager="jtaTransactionManager"> 
     <tx:attributes> 
      <tx:method name="processMessage" propagation="REQUIRED"/> 
     </tx:attributes> 
    </tx:advice> 

    <aop:config> 
     <aop:pointcut id="txOperation" 
      expression="execution(* axway.ReadMsgFromAxway.processMessage(..))"/> 
     <aop:advisor advice-ref="txAdvice" pointcut-ref="txOperation" /> 
    </aop:config> 

    <int:service-activator input-channel="inboundJMS" 
     output-channel="serviceActivatorChannel" ref="insertMessageToDb" method="processMessage" /> 

    <int:chain input-channel="serviceActivatorChannel" output-channel="nullChannel"> 
     <int:service-activator ref="serviceActivator" /> 
    </int:chain> 

</beans> 

ReadMsgFromAxway.java

public Message<File> processMessage(Message<?> message) { 
//Insert into DB 
     trackerProcess.insertUpdateMessageTracker(message, "axwayChannel", 
       "axwayChannel", currentStatusID, null, null); 
     count++; 
     int mod = count % 2; 
     if (mod != 0) { 
      // pass every 2 
      String hello = "hey"; 
     } else { 
      throw new RuntimeException("Testing transactional"); 
     } 

     Message<File> springMessage = MessageBuilder.createMessage(payloadFile, 
       messageHeaders); 
     return springMessage; 
    } 

Die XML wie es ist nichts tun, ob die Laufzeit Ausnahme ausgelöst wird, oder eine Ausnahme wird bei dem nächsten Service Aktivatorkomponente geworfen.

Wenn ich ändern die Beratung Attribute

 <tx:method name="*" propagation="REQUIRED"/> 

Dann werden die Ausnahmen in der 1. und 2. Service-Aktivator das Rollback verursacht.

Seltsamer wenn ich dies tun

 <tx:method name="processMessage" propagation="REQUIRED"/> 
     <tx:method name="*" propagation="NEVER"/> 

dann, ob die Laufzeitausnahme in dem ersten Dienst-Aktivator oder dem zweiten Aktivator ausgelöst wird, wird die Nachricht zurückgerollt. Ich würde denken, der Pointcut würde begrenzen, welche Klasse die Transaktion verursachen würde, aber vielleicht verstehe ich etwas falsch.

Ein weiterer Hinweis - Diese Anwendung befindet sich in einer Ear-Datei mit mehreren anderen Kriegen. Dieser Kontext startet den gesamten eingehenden Prozess und stellt über eine JMS-Warteschlange eine Verbindung zu einem anderen Krieg her, der Geschäftslogik enthält. Im Szenario der Verwendung von Methodenname = "*" sah ich die Ausnahmen im Business-Logik-Krieg, die bewirken, dass die JMS-Nachricht der ursprünglichen eingehenden Nachricht ebenfalls zurückgerollt wird. Ich hatte den Eindruck, dass der 2. Krieg seine Verarbeitung in einem anderen Thread durchführen würde, da er eine Nachricht über eine Warteschlange erhält und somit nicht Teil der Transaktion ist. Könnte das ein Nebeneffekt von JTA sein, der Container-verwaltet wird?

Danke!

Answer 1

Ich würde Ihnen empfehlen, Dave Syers article über Transaktion zu lesen, und dort können Sie einen Link in der "Mehr wie das" finden.

Im Moment sieht es so aus, als ob Sie Transaktionen und AOP überhaupt nicht verstehen. Sie sollten der AOP-Unterstützung in Spring Framework mehr Aufmerksamkeit schenken.

Im Allgemeinen sind deklarative Transaktionen (@Transactional auf Methode) die besonderen Fälle von AOP-Ratschlägen. Das Hauptkonzept hinter jedem AOP ist die Aufrufstapelgrenze, die durch den Methodenaufruf hervorgebracht wird, für den wir einen Hinweis angeben.

Wenn jedoch eine solche Methode im Zielobjekt nicht vorhanden ist, wird sie nicht an einen AOP-Proxy weitergeleitet. Wie in Ihrem Fall für die processMessage, wenn Sie die txAdvice für ein internes Objekt um die org.springframework.messaging.Message.MessageSource als Vertrag der JmsDestinationPollingSource für diese <int-jms:inbound-channel-adapter> gelten.

Stattdessen können Sie <transactional> Konfiguration der <poller> verwenden. Und richtig: Der gesamte Donwstream-Fluss wird durch die Transaktion abgedeckt.

Um Transaktion in diesem Fall für die internen

Callable<Boolean> pollingTask = new Callable<Boolean>() { 

     @Override 
     public Boolean call() throws Exception { 
      return doPoll(); 
     } 
    }; 

um und handleMessage(), Sie sollten nur den Methodenaufruf beenden, wie in regelmäßigem @Transactional Fall zu beenden. Zu diesem Zweck sollten wir die nächste Nachrichtenbehandlung auf den anderen Thread scheissen. Nur weil standardmäßig alle <channel> s sind DirectChannel und mit dem aktuellen Call-Stack gebunden.

Für diesen Zweck können Sie einfach eine ExecutorChannel (<int:dispatcher task-executor="threadPoolExecutor"/>) für Ihre serviceActivatorChannel verwenden.

Von dort brauchen Sie die restliche AOP-Konfiguration nicht.

Nicht sicher über Ihre zweite Frage für eine andere Web-App. Sieht so aus, als hätte es eine gewisse Logik, um seine Arbeit im Falle von Ungereimtheiten von Ihrer Seite zu wiederholen. Aber irgendwie sieht das nach einer anderen Frage aus.