Spring Batch Remote Chunking stellt Nachrichten nicht in die Warteschlange, sondern führt sie lokal aus?

Question

Unten ist meine Konfiguration für Spring Batch Remote Chunking.Meine Schritte werden lokal statt remote ausgeführt.Ich kann keine Nachrichten in Rabbitmq sehen.

<beans:bean id="importExchangesChunkItemWriter"
    class="org.springframework.batch.integration.chunk.ChunkMessageChannelItemWriter"
    scope="step" p:messagingOperations-ref="importExchangesMessagingTemplate"
    p:replyChannel-ref="importExchangesReplyChannel">
</beans:bean>

<beans:bean id="importExchangesChunkHandler"
    class="org.springframework.batch.integration.chunk.RemoteChunkHandlerFactoryBean"
    p:chunkWriter-ref="importExchangesChunkItemWriter" p:step-ref="importExchangesStep">
</beans:bean>

<job id="importExchangesJob" restartable="true">    
     <step id="importExchangesStep" next="importEclsStep">
        <tasklet transaction-manager="transactionManager">
            <chunk reader="importExchangesFileItemReader" writer="importExchangesItemWriter"
                commit-interval="${import.exchanges.commit.interval}" />
        </tasklet>
    </step>    
  </job>



<beans:bean id="passThroughItemProcessor" class="org.springframework.batch.item.support.PassThroughItemProcessor" />

<rabbit:connection-factory id="connectionFactory"
    port="${rabbitmq.port}" host="${rabbitmq.host}" username="${rabbitmq.username}"
    password="${rabbitmq.password}" />

<rabbit:template id="amqpTemplate" connection-factory="connectionFactory" />

<rabbit:admin id="rmqAdmin" connection-factory="connectionFactory" />

<rabbit:queue name="${import.exchanges.queue}" />
<rabbit:queue name="${import.exchanges.reply.queue}" />

<int:channel id="importExchangesChannel" />


<int:channel id="importExchangesReplyChannel" />


<beans:bean id="importExchangesMessagingTemplate"
    class="org.springframework.integration.core.MessagingTemplate"
    p:defaultChannel-ref="importExchangesChannel" p:receiveTimeout="${import.exchanges.reply.timeout}" />

<amqp:outbound-channel-adapter id="importExchangesOutboundAdapter"
    channel="importExchangesChannel" />

<amqp:inbound-channel-adapter id="importExchangesInboundAdapter"
    connection-factory="connectionFactory" channel="importExchangesReplyChannel"
    queue-names="${import.exchanges.reply.queue}" />


<amqp:inbound-channel-adapter id="importExchangesSlaveInboundAdapter"
    connection-factory="connectionFactory" channel="importExchangesChannel"
    queue-names="${import.exchanges.queue}" />



<amqp:outbound-channel-adapter id="importExchangesSlaveOutboundAdapter"
    channel="importExchangesReplyChannel" />

<int:service-activator id="serviceActivatorExchanges"
    input-channel="importExchangesChannel" output-channel="importExchangesReplyChannel"
    ref="chunkProcessorChunkHandlerExchanges" method="handleChunk" />


<beans:bean id="importExchangesItemWriter"
    class="com.st.batch.foundation.ImportExchangesItemWriter" p:symfony-ref="symfony" p:replyTimeout="${import.exchanges.reply.timeout}"/>



<beans:bean id="chunkProcessorExchanges"
    class="org.springframework.batch.core.step.item.SimpleChunkProcessor"
    p:itemWriter-ref="importExchangesItemWriter" p:itemProcessor-ref="passThroughItemProcessor"/>

<beans:bean id="chunkProcessorChunkHandlerExchanges"
    class="org.springframework.batch.integration.chunk.ChunkProcessorChunkHandler"
    p:chunkProcessor-ref="chunkProcessorExchanges" />

Die Konfiguration wurde dahingehend geändert, dass jetzt jeweils eine Nachricht in die Warteschlange gestellt wird und nicht mehrere verarbeitet werden (sollte die Anzahl der Nachrichten verarbeiten, die der Parallelität des Listeners entspricht).

<beans:bean id="simpleThreadScope"
    class="org.springframework.context.support.SimpleThreadScope" />

<util:map id="scopesMap">
    <beans:entry key="thread" value-ref="simpleThreadScope" />
</util:map>

<beans:bean
    class="org.springframework.beans.factory.config.CustomScopeConfigurer"
    p:scopes-ref="scopesMap" />





<int:channel id="importExchangesChannel" />
<int:channel id="importExchangesReplyChannel" scope="thread">
    <int:queue />
</int:channel>

<beans:bean id="importExchangesMessagingTemplate"
    class="org.springframework.integration.core.MessagingTemplate"
    p:defaultChannel-ref="importExchangesChannel" p:receiveTimeout="${import.exchanges.reply.timeout}" />

<amqp:outbound-channel-adapter
    amqp-template="amqpTemplate" channel="importExchangesChannel"
    exchange-name="${import.exchanges.exchange}" routing-key="${import.exchanges.routing.key}" />

<rabbit:listener-container
    connection-factory="rabbitConnectionFactory" concurrency="${import.exchanges.listener.concurrency}"
    requeue-rejected="false" prefetch="1">
    <rabbit:listener queues="${import.exchanges.queue}"
        ref="importExchangesChunkHandler" method="handleChunk" />
</rabbit:listener-container>


<int:channel id="importEclsChannel" />
<int:channel id="importEclsReplyChannel" scope="thread">
    <int:queue />
</int:channel>

<beans:bean id="importEclsMessagingTemplate"
    class="org.springframework.integration.core.MessagingTemplate"
    p:defaultChannel-ref="importEclsChannel" p:receiveTimeout="${import.ecls.reply.timeout}" />

<amqp:outbound-channel-adapter
    amqp-template="amqpTemplate" channel="importEclsChannel"
    exchange-name="${import.ecls.exchange}" routing-key="${import.ecls.routing.key}" />

<rabbit:listener-container
    connection-factory="rabbitConnectionFactory" concurrency="${import.ecls.listener.concurrency}"
    requeue-rejected="false" prefetch="1">
    <rabbit:listener queues="${import.ecls.queue}"
        ref="importEclsChunkHandler" method="handleChunk" />
</rabbit:listener-container>



<beans:bean id="importExchangesItemWriter"
    class="com.st.batch.foundation.ImportExchangesItemWriter"
    p:symfony-ref="symfony" p:replyTimeout="${import.exchanges.reply.timeout}" />


<beans:bean id="importExchangesChunkItemWriter"
    class="org.springframework.batch.integration.chunk.ChunkMessageChannelItemWriter"
    scope="step" p:messagingOperations-ref="importExchangesMessagingTemplate"
    p:replyChannel-ref="importExchangesReplyChannel">
</beans:bean>

<beans:bean id="importExchangesChunkHandler"
    class="org.springframework.batch.integration.chunk.RemoteChunkHandlerFactoryBean"
    p:chunkWriter-ref="importExchangesChunkItemWriter" p:step-ref="importExchangesStep">
</beans:bean>

<rabbit:queue name="${import.exchanges.queue}" />
<rabbit:queue name="${import.exchanges.reply.queue}" />

<rabbit:direct-exchange name="${import.exchanges.exchange}">
    <rabbit:bindings>
        <rabbit:binding queue="${import.exchanges.queue}"
            key="${import.exchanges.routing.key}" />
    </rabbit:bindings>
</rabbit:direct-exchange>

Ich kann jeweils nur eine Nachricht in der Warteschlange sehen.Ich sollte Nachrichten senden = ${import.exchanges.commit.interval} und alle sollten von gleichzeitigen Zuhörern abgeholt und parallel verarbeitet werden.

Answer 1

Ich bin mir nicht sicher, was Sie mit „lokal ausführen“ meinen, aber Sie haben keine Routing-Informationen zu den ausgehenden Adaptern.Wenn Rabbit nicht weiß, wie man Nachrichten weiterleitet, verwirft er sie einfach.

Sie müssen hinzufügen routing-key="${import.exchanges.queue}" Und routing-key="${import.exchanges.reply.queue}" zu den Adaptern.Dabei wird der Standardaustausch („“) verwendet, bei dem die Warteschlangen über ihre Namen gebunden sind.

Außerdem können Sie nicht auf beiden Seiten denselben Kanalnamen verwenden (importExchangesChannel).Auf diese Weise werden sowohl der Ausgangsadapter als auch der Dienstaktivator abonniert und Nachrichten werden im Round-Robin-Verfahren verteilt.

Einige Blöcke werden also lokal ausgeführt.andere werden aufgrund des Problems mit dem Routing-Schlüssel gelöscht.

Sie müssen den Routing-Schlüssel festlegen und auf der Serviceseite einen anderen Kanal verwenden.