apache-camel - Camel 2.21.0 - 如何使用流处理异常

Question

我想记录异常错误并继续下一个记录/拆分，但它不起作用。

我累了OnExcepiton()，doTry()DSL 但它不起作用并转到 ErrorHandler。

onException(IOException.class)
.handled(true).process(exchange -> log.error("error!!"));

from("file:" + rootDir + "/" + account + "/inbox/?move=.done")
.unmarshal(csvDataFormat)
.split(body()).shareUnitOfWork().parallelProcessing().streaming()
.process(fileService)
.end()

日志：

2018-07-18 14:01:59.883 DEBUG 45137 --- [/test1/request/] o.a.camel.processor.MulticastProcessor   : Parallel processing failed due IOException reading next record: java.io.IOException: (line 4) invalid char between encapsulated token and delimiter
2018-07-18 14:01:59.885 ERROR 45137 --- [/test1/request/] o.a.camel.processor.DeadLetterChannel    : Failed delivery for (MessageId: ID-**********-local-1531936914834-0-3 on ExchangeId: ID-*********-local-1531936914834-0-4). On delivery attempt: 0 caught: java.lang.IllegalStateException: IOException reading next record: java.io.IOException: (line 4) invalid char between encapsulated token and delimiter

Answer 1

@Bedla，感谢您的意见，我发现这适用于我的 UseCase，

• 使用onException()仍在发送交换到 DeadLetterChannel，所以不得不使用doTry()
• CasvFormat使用 maps- 我无法修改csvFormatin process，因此必须从文件中读取标头并在每次拆分时将 csv 标头附加到正文中setBody

完整路线定义：

CsvDataFormat csvDataFormat = new CsvDataFormat().setUseMaps(true);

from("file:" + rootDir + "/test/")
                .log(LoggingLevel.INFO,"Start processing ${file:name}")
                .unmarshal().pgp(pgpFileName,pgpUserId,pgpPassword)
                .process(exchange -> { /* just to get csv header */
                    InputStream inputStream = exchange.getIn().getBody(InputStream.class);
                    try(BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream))){
                        String header = bufferedReader.readLine();
                        exchange.getIn().setHeader("CSV_HEADER",header);
                        csvDataFormat.setHeader(header.split(",")); //<- this does not work, so had to add in body below!
                        System.out.println("csvHeader is : " + header);// + " ? " + Arrays.asList(csvDataFormat.getHeader()));
                    }
                })
                .split(body().tokenize("\n")).shareUnitOfWork()
                .parallelProcessing().streaming()
                .setBody(exchange -> exchange.getIn().getHeader("CSV_HEADER") + "\n" + exchange.getIn().getBody())
                .doTry()
                  .unmarshal(csvDataFormat)
                  .process(requestFileService)
                .doCatch(IOException.class)
                  //TODO: custom processing here...
                  .process(exchange -> log.error("caught in dotry: " + exchange.getIn().getBody())).stop()
                .end()//end try/catch
                .choice()
                    .when(simple("${property." + Exchange.SPLIT_COMPLETE + "} == true"))
                    .log(LoggingLevel.INFO, "Finished processing ${file:name}")
                .end();