camel系列-异常处理示例

发表于 更新于

本篇以实际例子来介绍 camel 的异常处理

抛异常

在路由运行期间使用 throwException 方法直接抛异常

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public class NewExceptionTest extends CamelTestSupport {

    @Override
    protected RouteBuilder createRouteBuilder() throws Exception {
        return new RouteBuilder() {
            @Override
            public void configure() throws Exception {
                from("direct:start")
                    .throwException(new RuntimeException("Forbidden"))
                    .end()
                    .to("mock:done");
            }
        };
    }

    @Test
    public void testNoError() throws Exception {
        getMockEndpoint("mock:done").expectedMessageCount(1);
        template.sendBodyAndHeader("direct:start", "Hello Camel", "name", "Camel");
        assertMockEndpointsSatisfied();
    }
}

运行日志结果

1
2
3
4
5
6
7
8
org.apache.camel.CamelExecutionException: Exception occurred during execution on the exchange: Exchange[ID-MacBook-Pro-local-1649419373115-0-1]

at org.apache.camel.util.ObjectHelper.wrapCamelExecutionException(ObjectHelper.java:1842)
at org.apache.camel.util.ExchangeHelper.extractResultBody(ExchangeHelper.java:745)
at org.apache.camel.impl.DefaultProducerTemplate.extractResultBody(DefaultProducerTemplate.java:515)
at org.apache.camel.impl.DefaultProducerTemplate.extractResultBody(DefaultProducerTemplate.java:511)
at org.apache.camel.impl.DefaultProducerTemplate.sendBodyAndHeader(DefaultProducerTemplate.java:189)
at org.apache.camel.impl.DefaultProducerTemplate.sendBodyAndHeader(DefaultProducerTemplate.java:183)

捕获并处理异常

这里分为 3 个步骤

  1. 使用 onException 方法捕获异常，其类似 try catch 的 catch 动作
  2. 使用 handled 方法将异常标记为已处理
  3. 自定义处理异常，处理时，可以在 exchange 中使用Exchange.EXCEPTION_CAUGHT属性字段捕获当前的错误异常
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
	return new RouteBuilder() {
		@Override
		public void configure() throws Exception {

			onException(Exception.class)
				.handled(true)
				.process((exchange -> {
					Throwable caused = exchange.getProperty(Exchange.EXCEPTION_CAUGHT, Throwable.class);
					log.error("error:"+caused.getMessage());
				}));

			from("direct:start")
				.throwException(new RuntimeException("Forbidden"))
				.end()
				.to("mock:done");
		}
	};
}

日志输出

1
2
2022-04-08 20:20:46,176 [main           ] ERROR NewExceptionTest2$1            - error:Forbidden
2022-04-08 20:20:46,177 [main           ] INFO  MockEndpoint                   - Asserting: mock://done is satisfied

捕获异常并继续

有些场景，异常捕获之后并不影响正常流程，路由的后续流程可以继续运行，使用continued方法设置为 true，表示忽略该异常并继续运行，实际上有点类似该异常并没有发生，如下代码示例

1
2
3
4
5
6
onException(IllegalArgumentException.class).continued(true);

from("direct:start")
    .to("mock:start")
    .throwException(new IllegalArgumentException("Forced"))
  .to("mock:result");

异常返回

当异常被捕获并处理完以后，但由于并没有返回结果，原路由依然在等待结果，所以可以在异常处理完毕之后，使用 to 方法，返回自定义的响应，这样该路由整体流程是完整的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            onException(Exception.class)
                .handled(true)
                .process((exchange -> {
                    Throwable caused = exchange.getProperty(Exchange.EXCEPTION_CAUGHT, Throwable.class);
                    log.error("error:"+caused.getMessage());
                })).to("mock:done");

            from("direct:start")
                .throwException(new RuntimeException("Forbidden"))
                .end()
                .to("mock:done");
        }
    };
}

处理和继续有什么区别？

  • 如果 handled 为真，那么抛出的异常会被处理，Camel 不会继续在原来的路由中路由，而是抛异常。但是，您可以在其中配置将使用的路由 onException。如果您需要创建一些自定义响应消息返回给调用者，或者因为抛出异常而进行任何其他处理，您可以使用此路由。
  • 如果 continue 为真，那么 Camel 将捕获异常，实际上只是忽略它并继续在原始路由中路由。但是，如果您在其中配置了 onException 路由，它将首先路由该路由，然后再继续在原始路由中路由。

失败重试

可以在捕获异常后采取失败重试策略，maximumRedeliveries 表示要重试的次数，如果是 2，那么该方法一直失败的话，则是一共执行 3 次，即一共执行 maximumRedeliveries+1 次，如下代码清单示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            onException(Exception.class)
                .maximumRedeliveries(2);

            from("direct:start")
                .process((exchange -> {
                    index++;
                    log.info("doProcess:"+index);
                    if(index<3)
                    {
                        throw new RuntimeException("Forbidden");
                    }
                    log.info("doProcess end");
                }))
                .end()
                .to("mock:done");
        }
    };
}

日志输出结果

1
2
3
4
2022-04-08 20:40:36,303 [main           ] INFO  NewExceptionTest3$1            - doProcess:1
2022-04-08 20:40:37,316 [main           ] INFO  NewExceptionTest3$1            - doProcess:2
2022-04-08 20:40:38,324 [main           ] INFO  NewExceptionTest3$1            - doProcess:3
2022-04-08 20:40:38,324 [main           ] INFO  NewExceptionTest3$1            - doProcess end

失败延迟重试

通过设置 redeliveryDelay 方法，支持延迟重试，比如网络不好的场景，不需要太频繁的重试，在上面例子基础上添加 redeliveryDelay 方法，设置为每 5 秒重试一次

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {

            onException(Exception.class)
                .maximumRedeliveries(2)
                .redeliveryDelay(5000);

            from("direct:start")
                .process((exchange -> {
                    index++;
                    log.info("doProcess:"+index);
                    if(index<3)
                    {
                        throw new RuntimeException("Forbidden");
                    }
                    log.info("doProcess end");
                }))
                .end()
                .to("mock:done");
        }
    };
}

日志输出结果

1
2
3
4
2022-04-08 20:49:54,816 [main           ] INFO  NewExceptionTest3$1            - doProcess:1
2022-04-08 20:49:59,830 [main           ] INFO  NewExceptionTest3$1            - doProcess:2
2022-04-08 20:50:04,835 [main           ] INFO  NewExceptionTest3$1            - doProcess:3
2022-04-08 20:50:04,835 [main           ] INFO  NewExceptionTest3$1            - doProcess end

重试终止条件

retryWhile 方法可以传入一个谓语，即满足该条件以后才会发起重试，当返回 false，重试将会终止如下示例，将重试次数放在 retryWhile 中检查

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {

            onException(Exception.class)
                .maximumRedeliveries(5)
                .retryWhile((exchange -> {
                    if(index>2)
                    {
                        return false;
                    }
                    return true;
                }));
            from("direct:start")
                .process((exchange -> {
                    index++;
                    log.info("doProcess:"+index);
                    throw new RuntimeException("Forbidden");
                }))
                .end()
                .to("mock:done");
        }
    };
}

TRY … CATCH … FINALLY

以传统TRY … CATCH … FINALLY 的方式捕获异常

1
2
3
4
5
6
7
8
9
from("direct:start")
    .doTry()
        .process(new ProcessorFail())
        .to("mock:result")
    .doCatch(IOException.class, IllegalStateException.class)
        .to("mock:catch")
    .doFinally()
        .to("mock:finally")
    .end();

全局异常捕获

使用 errorHandler 方法设置全局异常

1
2
3
4
errorHandler(defaultErrorHandler()
    .maximumRedeliveries(3)
    .redeliveryDelay(5000)
    .onExceptionOccurred(myProcessor));

参考：