java - Apache Camel : ValidatingProcessor : XSD validation

Question

---

我正在尝试创建一个包并在 ServiceMix 中运行它。我遇到了问题，如有任何帮助，我们将不胜感激。

我正在尝试做的事情:

从 CXF 端点生成 SOAP 消息，并使用 Apache ServiceMix 5.0.0 (Camel 2.12.3) 根据 XSD 模式验证它们。

我做了什么:

实现了从 CXF 端点生成消息然后验证它们的路由。

路由的configure方法:

private RouteBuilder getInputRoute() {

    SoapValidatingProcessor soapValidatingProcessor = new SoapValidatingProcessor();

    inputRoute = new RouteBuilder() {

      @Override
      public void configure() throws Exception {

        from(cxfEndpointInId())
          .convertBodyTo(java.lang.String.class, "UTF-8")
          .onException(org.apache.camel.ValidationException.class)
            .log(LoggingLevel.INFO, LOG_NAME, "Invalid message received!")
            .handled(true)
            .stop()
            .end()
          .bean(soapValidatingProcessor).id("SoapHeaderValidatingProcessor")
          .to("browse:foo")

}

创建了一个验证处理器:

public class SoapValidatingProcessor {

  private final String SCHEMA = "schema.xsd";  

  public SoapValidatingProcessor() {

    validatingProcessor = new ValidatingProcessor();
    validatingProcessor.setFailOnNullHeader(false);
    validatingProcessor.setFailOnNullBody(false);

  }

  @Handler
  public void validate(Exchange exchange) throws Exception {

    Resource validationSchema = context.getApplicationContext().getResource(SCHEMA);
    validatingProcessor.setSchemaUrl(validationSchema.getURL());
    validatingProcessor.loadSchema();

    /* Creating a new SchemaFactory instance */
    SchemaFactory xmlSchema = SchemaFactory
        .newInstance(XMLConstants.W3C_XML_SCHEMA_NS_URI);   

    validatingProcessor.setSchemaFactory(xmlSchema);    
    validatingProcessor.process(exchange);

  }

}

问题:

Apache Camel ValidatingProcessor [1] 在多重威胁中表现不佳。我正在大约发送 SOAP 消息。 20 毫秒间隔，我收到以下异常。如果我将发送间隔增加到 200 毫秒以上，一切都很好。

2014-07-11 17:11:47,404 | WARN  | ult-workqueue-18 | PhaseInterceptorChain            | ?                                   ? | 129 - org.ap
ache.cxf.cxf-api - 2.7.10 | Application {http://ws.service}WSImplService#{http://foo.bar}Update has thrown exception, unwinding now
org.apache.cxf.interceptor.Fault: FWK005 parse may not be called while parsing.
        at org.apache.camel.component.cxf.CxfConsumer$1.checkFailure(CxfConsumer.java:228)[181:org.apache.camel.camel-cxf:2.12.3]
        at org.apache.camel.component.cxf.CxfConsumer$1.setResponseBack(CxfConsumer.java:206)[181:org.apache.camel.camel-cxf:2.12.3]
        at org.apache.camel.component.cxf.CxfConsumer$1.syncInvoke(CxfConsumer.java:140)[181:org.apache.camel.camel-cxf:2.12.3]
        at org.apache.camel.component.cxf.CxfConsumer$1.invoke(CxfConsumer.java:75)[181:org.apache.camel.camel-cxf:2.12.3]
        at org.apache.cxf.interceptor.ServiceInvokerInterceptor$1.run(ServiceInvokerInterceptor.java:58)[129:org.apache.cxf.cxf-api:2.7.10]
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)[:1.7.0_45]
        at java.util.concurrent.FutureTask.run(FutureTask.java:262)[:1.7.0_45]
        at org.apache.cxf.workqueue.SynchronousExecutor.execute(SynchronousExecutor.java:37)[129:org.apache.cxf.cxf-api:2.7.10]
        at org.apache.cxf.interceptor.ServiceInvokerInterceptor.handleMessage(ServiceInvokerInterceptor.java:107)[129:org.apache.cxf.cxf-api
:2.7.10]
        at org.apache.cxf.phase.PhaseInterceptorChain.doIntercept(PhaseInterceptorChain.java:272)[129:org.apache.cxf.cxf-api:2.7.10]
        at org.apache.cxf.phase.PhaseInterceptorChain.resume(PhaseInterceptorChain.java:242)[129:org.apache.cxf.cxf-api:2.7.10]
        at org.apache.cxf.interceptor.OneWayProcessorInterceptor$1.run(OneWayProcessorInterceptor.java:144)[129:org.apache.cxf.cxf-api:2.7.1
0]
        at org.apache.cxf.workqueue.AutomaticWorkQueueImpl$3.run(AutomaticWorkQueueImpl.java:428)[129:org.apache.cxf.cxf-api:2.7.10]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)[:1.7.0_45]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)[:1.7.0_45]
        at org.apache.cxf.workqueue.AutomaticWorkQueueImpl$AWQThreadFactory$1.run(AutomaticWorkQueueImpl.java:353)[129:org.apache.cxf.cxf-ap
i:2.7.10]
        at java.lang.Thread.run(Thread.java:744)[:1.7.0_45]
Caused by: org.xml.sax.SAXException: FWK005 parse may not be called while parsing.
        at org.apache.xerces.jaxp.validation.Util.toSAXException(Unknown Source)[:]

如果 validate 方法是 synchronized，我不会得到该异常的唯一方法，但我不希望那样。

我发现了什么:

我在这里 [2] 看到错误已解决。我尝试为每条消息创建一个新的 SchemaFactory [3] 实例，但仍然没有成功。任何想法为什么我仍然收到错误？我做错了什么？

[1] http://camel.apache.org/maven/camel-2.8.0/camel-core/apidocs/org/apache/camel/processor/validation/ValidatingProcessor.html

[2] https://issues.apache.org/jira/browse/CAMEL-6630

[3] http://docs.oracle.com/javase/6/docs/api/javax/xml/validation/SchemaFactory.html

谢谢!

Answer 1

您的代码不是线程安全的。在您的 SoapValidatingProcessor 中，您正在使用共享的 validatingProcessor，同时修改其状态/配置(通过每次加载不同的模式)。

但是，您的示例中的架构没有改变。在这种情况下，您应该只配置处理器一次(通过在启动时加载架构)。

您的处理程序应该只调用不修改处理器状态的方法:

@Handler
public void validate(Exchange exchange) throws Exception {
    validatingProcessor.process(exchange);
}

并且状态应该在处理任何交换之前配置一次，例如:

public SoapValidatingProcessor(CamelContext context) {
    validatingProcessor = new ValidatingProcessor();
    validatingProcessor.setFailOnNullHeader(false);
    validatingProcessor.setFailOnNullBody(false);

    Resource validationSchema = context.getApplicationContext().getResource(SCHEMA);
    validatingProcessor.setSchemaUrl(validationSchema.getURL());
    validatingProcessor.loadSchema();

    /* Creating a new SchemaFactory instance */
    SchemaFactory xmlSchema = SchemaFactory
        .newInstance(XMLConstants.W3C_XML_SCHEMA_NS_URI);   

    validatingProcessor.setSchemaFactory(xmlSchema);    
}

事实上，删除 SoapValidatingProcessor 类并将所有 validatingProcessor 配置移动到 RouteBuilder.configure() 可能会更好方法，然后直接在路由中使用 validatingProcessor。