Как заставить постоянные данные во время выполнения верблюжьих процессоров, а не в конце маршрута?

Question

Когда я переключаю точку останова рядом с repository.saveAndFlush и в режиме отладки, я вижу, что он возвращает новый клиентский объект с новым Id, но когда я проверяю базу данных, я не нахожу этого клиента.Однако, если я делаю резюме (F8 с затмением), то я перепроверяю БД и нахожу клиента.

Итак, как заставить сохраняться данные во время выполнения верблюжьих процессоров, а не конца маршрута?

   @Component
    public class myRoute extends RouteBuilder {

    @Autowired Processor validationDatasProcessor;
    @Autowired Processor clientProcessor;
    @Autowired Processor endCientProcessor;

    @Override
    public void configure() throws Exception {
        from("queueIn")
        .id("route_processing").messageHistory().transacted()
        .log(LoggingLevel.DEBUG, log, "reception").pipeline()
        .process(validationDatasProcessor)
        .id(validationDatasProcessor.getClass().getSimpleName().toLowerCase())
        .process(clientProcessor)
        .id(clientProcessor.getClass().getSimpleName().toLowerCase())
        .process(endCientProcessor).id(endCientProcessor.getClass().getSimpleName().toLowerCase())
        .to("outputQueue")
        .end();
     }
    }

Процессоры:

@Component
public class ValidationDatasProcessor implements Processor {

    @Autowired ObjectMapper objectMapper;

    @Autowired ClientRepository clientRepository;

    @Override
    public void process(Exchange exchange) throws Exception {
        String clientString = exchange.getIn().getBody(String.class);
        Client client = objectMapper.readValue(clientString, Client.class);
        clientRepository.saveAndFlush(client)
        exchange.setOut(generateOutMessage(client, exchange.getContext()));
    }

    Message generateOutMessage(Client client, CamelContext camelContext) throws JsonProcessingException {
        DefaultMessage outMessage = new DefaultMessage(camelContext);
        outMessage.setBody(objectMapper.writeValueAsString(client), String.class);
        return outMessage;
    }
}

@Component
public class ClientProcessor implements Processor {

    @Autowired ObjectMapper objectMapper;

    @Autowired ClientRepository clientRepository;

    ....

    @Override
    public void process(Exchange exchange) throws Exception {
        String clientString = exchange.getIn().getBody(String.class);
        Client client = objectMapper.readValue(clientString, Client.class);
        client.setAccessDate(LocalDateTime.now);
        clientRepository.saveAndFlush(client)
        exchange.setOut(generateOutMessage(client, exchange.getContext()));
    }

    Message generateOutMessage(Client client, CamelContext camelContext) throws JsonProcessingException {
        DefaultMessage outMessage = new DefaultMessage(camelContext);
        outMessage.setBody(objectMapper.writeValueAsString(client), String.class);
        return outMessage;
    }
}

Answer 1

Весь ваш маршрут является транзакционным; Это означает, что весь маршрут находится в рамках транзакции. commit будет выполнено только после выполнения всего маршрута.

, если вы хотите запустить процессор за пределами транзакции, разделите маршрут и используйте конечную точку seda. seda являются асинхронными и запускают новые потоки. Они не будут участвовать в активной границе транзакции.

Фиксация частей в середине объема транзакции не кажется хорошей идеей. Возможно, ваш раунд должен быть разбит на несколько фрагментов.

Эта документация может помочь вам лучше понять их.