java - Spring集成DSL lastTime没有更新错误

Question

我在 Spring 集成 DSL 方面完全是新手，我想做的是从 mongo 获取所有 RSS 提要，将它们全部注册到流上下文中，然后处理从提要中获取的所有文章。目前我有一个 for-each 循环，它调用一个包含以下内容的函数:

IntegrationFlow newFlow = IntegrationFlows
            .from(Feed.inboundAdapter(new URL(url), source), e -> e.id(org + "-" + source + "-feed").poller(Pollers.fixedDelay(poll).maxMessagesPerPoll(maxMessages).errorChannel("feedErrors")))
            .enrichHeaders(h -> h.header("sourceId", sourceId))
            .enrichHeaders(h -> h.header("sourceName", sourceName))
            .enrichHeaders(h -> h.header("source", source))
            .enrichHeaders(h -> h.header("categories", categories))
            .enrichHeaders(h -> h.header("org", org))
            .channel(MessageChannels.executor(taskExecutor))
            .handle("enricher", "enhance")
            .channel(news())
            .get();
        
this.flowContext.registration(newFlow).id(org + "-" + source + "-flow").register();

除了异常(exception)，它做得非常好，很有趣。当提要当前不可用、已重命名或来自提要本身的文章格式错误时，将引发异常并进入流定义中指示的“feedErrors” channel 。该部分如下所示:

@Component
public class RssFeedErrorHandler {

    private final Logger logger = LoggerFactory.getLogger(RssFeedErrorHandler.class);

    @Bean
    public IntegrationFlow errorSender() {
        return IntegrationFlows.from("feedErrors")
            .handle("rssFeedErrorHandler", "handleError")
            .get();
    }

    public void handleError(Message<MessagingException> message) {
        logger.error(message.toString());
    }
}

异常示例:

AdviceMessage [payload=org.springframework.messaging.MessagingException: Failed to retrieve feed for 'FeedEntryMessageSource{feedUrl=https://news.google.com/rss/topics/CAAqIQgKIhtDQkFTRGdvSUwyMHZNRFZ4ZERBU0FtVnVLQUFQAQ?hl=en-GB&gl=GB&ceid=GB%3Aen, feedResource=null, metadataKey='google-news-politics.https://news.google.com/rss/topics/CAAqIQgKIhtDQkFTRGdvSUwyMHZNRFZ4ZERBU0FtVnVLQUFQAQ?hl=en-GB&gl=GB&ceid=GB%3Aen', lastTime=1613665200000}'; nested exception is java.io.FileNotFoundException: https://news.google.com/rss/topics/CAAqIQgKIhtDQkFTRGdvSUwyMHZNRFZ4ZERBU0FtVnVLQUFQAQ?hl=en-GB&gl=GB&ceid=GB%3Aen was successful, headers={id=101f3da9-42de-eb03-0da2-015952fc0a23, timestamp=1613667428995}, inputMessage=ErrorMessage [payload=org.springframework.messaging.MessagingException: Failed to retrieve feed for 'FeedEntryMessageSource{feedUrl=https://news.google.com/rss/topics/CAAqIQgKIhtDQkFTRGdvSUwyMHZNRFZ4ZERBU0FtVnVLQUFQAQ?hl=en-GB&gl=GB&ceid=GB%3Aen, feedResource=null, metadataKey='google-news-politics.https://news.google.com/rss/topics/CAAqIQgKIhtDQkFTRGdvSUwyMHZNRFZ4ZERBU0FtVnVLQUFQAQ?hl=en-GB&gl=GB&ceid=GB%3Aen', lastTime=1613665200000}'; nested exception is java.io.FileNotFoundException: https://news.google.com/rss/topics/CAAqIQgKIhtDQkFTRGdvSUwyMHZNRFZ4ZERBU0FtVnVLQUFQAQ?hl=en-GB&gl=GB&ceid=GB%3Aen, headers={id=8a470a5f-4027-371c-44fd-1250703077b0, timestamp=1613667428995}]]
AdviceMessage [payload=org.springframework.messaging.MessagingException: Failed to retrieve feed for 'FeedEntryMessageSource{feedUrl=https://www.startupbootcamp.org/feed/, feedResource=null, metadataKey='startupbootcamp-startups.https://www.startupbootcamp.org/feed/', lastTime=1613015630000}'; nested exception is java.net.ConnectException: Operation timed out (Connection timed out) was successful, headers={id=6b5a1a00-905a-68b6-128b-4cdd8eb06737, timestamp=1613664837150}, inputMessage=ErrorMessage [payload=org.springframework.messaging.MessagingException: Failed to retrieve feed for 'FeedEntryMessageSource{feedUrl=https://www.startupbootcamp.org/feed/, feedResource=null, metadataKey='startupbootcamp-startups.https://www.startupbootcamp.org/feed/', lastTime=1613015630000}'; nested exception is java.net.ConnectException: Operation timed out (Connection timed out), headers={id=70ef8ae7-1b75-9e38-e631-c88777741609, timestamp=1613664837150}]]
AdviceMessage [payload=org.springframework.messaging.MessagingException: Failed to retrieve feed for 'FeedEntryMessageSource{feedUrl=https://timesofindia.indiatimes.com/rssfeeds/1081479906.cms, feedResource=null, metadataKey='india-times-entertainment.https://timesofindia.indiatimes.com/rssfeeds/1081479906.cms', lastTime=1613715922000}'; nested exception is com.rometools.rome.io.ParsingFeedException: Invalid XML: Error on line 1: Character reference "&#55357" is an invalid XML character. was successful, headers={id=fb936878-6450-d79a-851a-ca1ceeef8182, timestamp=1613705658735}, inputMessage=ErrorMessage [payload=org.springframework.messaging.MessagingException: Failed to retrieve feed for 'FeedEntryMessageSource{feedUrl=https://timesofindia.indiatimes.com/rssfeeds/1081479906.cms, feedResource=null, metadataKey='india-times-entertainment.https://timesofindia.indiatimes.com/rssfeeds/1081479906.cms', lastTime=1613715922000}'; nested exception is com.rometools.rome.io.ParsingFeedException: Invalid XML: Error on line 1: Character reference "&#55357" is an invalid XML character., headers={id=24f2a11a-80a6-6c13-dbe1-9548747174aa, timestamp=1613705658734}]]
AdviceMessage [payload=org.springframework.messaging.MessagingException: Failed to retrieve feed for 'FeedEntryMessageSource{feedUrl=https://www.jeffbullas.com/feed/, feedResource=null, metadataKey='jeff-bullas-marketing.https://www.jeffbullas.com/feed/', lastTime=1613574000000}'; nested exception is java.io.IOException: Server returned HTTP response code: 500 for URL: https://www.jeffbullas.com/feed/ was successful, headers={id=e6998e35-afdc-b476-c2d9-2f4b0cc46c6b, timestamp=1613710313323}, inputMessage=ErrorMessage [payload=org.springframework.messaging.MessagingException: Failed to retrieve feed for 'FeedEntryMessageSource{feedUrl=https://www.jeffbullas.com/feed/, feedResource=null, metadataKey='jeff-bullas-marketing.https://www.jeffbullas.com/feed/', lastTime=1613574000000}'; nested exception is java.io.IOException: Server returned HTTP response code: 500 for URL: https://www.jeffbullas.com/feed/, headers={id=420f9b5d-1c58-cd0a-c4e2-5d640578f1c9, timestamp=1613710313322}]]

这是两个完整的堆栈:

org.springframework.messaging.MessagingException: Failed to retrieve feed for 'FeedEntryMessageSource{feedUrl=https://timesofindia.indiatimes.com/rssfeeds/1081479906.cms, feedResource=null, metadataKey='india-times-entertainment.https://timesofindia.indiatimes.com/rssfeeds/1081479906.cms', lastTime=-1}'; nested exception is com.rometools.rome.io.ParsingFeedException: Invalid XML: Error on line 1: Character reference "&#55357" is an invalid XML character.
        at org.springframework.integration.feed.inbound.FeedEntryMessageSource.getFeed(FeedEntryMessageSource.java:239)
        at org.springframework.integration.feed.inbound.FeedEntryMessageSource.populateEntryList(FeedEntryMessageSource.java:202)
        at org.springframework.integration.feed.inbound.FeedEntryMessageSource.doReceive(FeedEntryMessageSource.java:177)
        at org.springframework.integration.feed.inbound.FeedEntryMessageSource.doReceive(FeedEntryMessageSource.java:58)
        at org.springframework.integration.endpoint.AbstractMessageSource.receive(AbstractMessageSource.java:167)
        at org.springframework.integration.endpoint.SourcePollingChannelAdapter.receiveMessage(SourcePollingChannelAdapter.java:250)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:359)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint.pollForMessage(AbstractPollingEndpoint.java:328)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$null$1(AbstractPollingEndpoint.java:275)
        at org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute$0(ErrorHandlingTaskExecutor.java:57)
        at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
        at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:55)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$2(AbstractPollingEndpoint.java:272)
        at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
        at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:93)
        at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
        at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
        at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
        at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: com.rometools.rome.io.ParsingFeedException: Invalid XML: Error on line 1: Character reference "&#55357" is an invalid XML character.
        at com.rometools.rome.io.WireFeedInput.build(WireFeedInput.java:236)
        at com.rometools.rome.io.SyndFeedInput.build(SyndFeedInput.java:150)
        at org.springframework.integration.feed.inbound.FeedEntryMessageSource.getFeed(FeedEntryMessageSource.java:226)
        ... 20 more
Caused by: org.jdom2.input.JDOMParseException: Error on line 1: Character reference "&#55357" is an invalid XML character.
        at org.jdom2.input.sax.SAXBuilderEngine.build(SAXBuilderEngine.java:232)
        at org.jdom2.input.sax.SAXBuilderEngine.build(SAXBuilderEngine.java:303)
        at org.jdom2.input.SAXBuilder.build(SAXBuilder.java:1196)
        at com.rometools.rome.io.WireFeedInput.build(WireFeedInput.java:233)
        ... 22 more
Caused by: org.xml.sax.SAXParseException; lineNumber: 1; columnNumber: 15681; Character reference "&#55357" is an invalid XML character.
        at org.apache.xerces.util.ErrorHandlerWrapper.createSAXParseException(Unknown Source)
        at org.apache.xerces.util.ErrorHandlerWrapper.fatalError(Unknown Source)
        at org.apache.xerces.impl.XMLErrorReporter.reportError(Unknown Source)
        at org.apache.xerces.impl.XMLErrorReporter.reportError(Unknown Source)
        at org.apache.xerces.impl.XMLErrorReporter.reportError(Unknown Source)
        at org.apache.xerces.impl.XMLScanner.reportFatalError(Unknown Source)
        at org.apache.xerces.impl.XMLScanner.scanCharReferenceValue(Unknown Source)
        at org.apache.xerces.impl.XMLDocumentFragmentScannerImpl.scanCharReference(Unknown Source)
        at org.apache.xerces.impl.XMLDocumentFragmentScannerImpl$FragmentContentDispatcher.dispatch(Unknown Source)
        at org.apache.xerces.impl.XMLDocumentFragmentScannerImpl.scanDocument(Unknown Source)
        at org.apache.xerces.parsers.XML11Configuration.parse(Unknown Source)
        at org.apache.xerces.parsers.XML11Configuration.parse(Unknown Source)
        at org.apache.xerces.parsers.XMLParser.parse(Unknown Source)
        at org.apache.xerces.parsers.AbstractSAXParser.parse(Unknown Source)
        at org.apache.xerces.jaxp.SAXParserImpl$JAXPSAXParser.parse(Unknown Source)
        at org.jdom2.input.sax.SAXBuilderEngine.build(SAXBuilderEngine.java:217)
        ... 25 more


org.springframework.messaging.MessagingException: Failed to retrieve feed for 'FeedEntryMessageSource{feedUrl=http://www.independent.co.uk/news/uk/politics/rss, feedResource=null, metadataKey='independent-politics.http://www.independent.co.uk/news/uk/politics/rss', lastTime=1613726122000}'; nested exception is com.rometools.rome.io.ParsingFeedException: Invalid XML: Error on line 1: DOCTYPE is disallowed when the feature "http://apache.org/xml/features/disallow-doctype-decl" set to true.
        at org.springframework.integration.feed.inbound.FeedEntryMessageSource.getFeed(FeedEntryMessageSource.java:239)
        at org.springframework.integration.feed.inbound.FeedEntryMessageSource.populateEntryList(FeedEntryMessageSource.java:202)
        at org.springframework.integration.feed.inbound.FeedEntryMessageSource.doReceive(FeedEntryMessageSource.java:177)
        at org.springframework.integration.feed.inbound.FeedEntryMessageSource.doReceive(FeedEntryMessageSource.java:58)
        at org.springframework.integration.endpoint.AbstractMessageSource.receive(AbstractMessageSource.java:167)
        at org.springframework.integration.endpoint.SourcePollingChannelAdapter.receiveMessage(SourcePollingChannelAdapter.java:250)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:359)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint.pollForMessage(AbstractPollingEndpoint.java:328)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$null$1(AbstractPollingEndpoint.java:275)
        at org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute$0(ErrorHandlingTaskExecutor.java:57)
        at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
        at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:55)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$2(AbstractPollingEndpoint.java:272)
        at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
        at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:93)
        at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
        at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
        at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
        at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: com.rometools.rome.io.ParsingFeedException: Invalid XML: Error on line 1: DOCTYPE is disallowed when the feature "http://apache.org/xml/features/disallow-doctype-decl" set to true.
        at com.rometools.rome.io.WireFeedInput.build(WireFeedInput.java:236)
        at com.rometools.rome.io.SyndFeedInput.build(SyndFeedInput.java:150)
        at org.springframework.integration.feed.inbound.FeedEntryMessageSource.getFeed(FeedEntryMessageSource.java:226)
        ... 20 more
Caused by: org.jdom2.input.JDOMParseException: Error on line 1: DOCTYPE is disallowed when the feature "http://apache.org/xml/features/disallow-doctype-decl" set to true.
        at org.jdom2.input.sax.SAXBuilderEngine.build(SAXBuilderEngine.java:232)
            at org.jdom2.input.sax.SAXBuilderEngine.build(SAXBuilderEngine.java:303)
        at org.jdom2.input.SAXBuilder.build(SAXBuilder.java:1196)
        at com.rometools.rome.io.WireFeedInput.build(WireFeedInput.java:233)
            ... 22 more
Caused by: org.xml.sax.SAXParseException; lineNumber: 1; columnNumber: 10; DOCTYPE is disallowed when the feature "http://apache.org/xml/features/disallow-doctype-decl" set to true.
        at org.apache.xerces.util.ErrorHandlerWrapper.createSAXParseException(Unknown Source)
        at org.apache.xerces.util.ErrorHandlerWrapper.fatalError(Unknown Source)
        at org.apache.xerces.impl.XMLErrorReporter.reportError(Unknown Source)
        at org.apache.xerces.impl.XMLErrorReporter.reportError(Unknown Source)
        at org.apache.xerces.impl.XMLErrorReporter.reportError(Unknown Source)
        at org.apache.xerces.impl.XMLScanner.reportFatalError(Unknown Source)
        at org.apache.xerces.impl.XMLDocumentScannerImpl$PrologDispatcher.dispatch(Unknown Source)
        at org.apache.xerces.impl.XMLDocumentFragmentScannerImpl.scanDocument(Unknown Source)
        at org.apache.xerces.parsers.XML11Configuration.parse(Unknown Source)
        at org.apache.xerces.parsers.XML11Configuration.parse(Unknown Source)
        at org.apache.xerces.parsers.XMLParser.parse(Unknown Source)
        at org.apache.xerces.parsers.AbstractSAXParser.parse(Unknown Source)
        at org.apache.xerces.jaxp.SAXParserImpl$JAXPSAXParser.parse(Unknown Source)
        at org.jdom2.input.sax.SAXBuilderEngine.build(SAXBuilderEngine.java:217)
        ... 25 more

所以错误被抛出，它被记录下来，生活还在继续，除了出于某种原因，poll fetch 似乎被保存以供重试。如果它重试一次，这不会是一个问题，但它只是一遍又一遍地重试，当足够多的错误堆积起来时，它只是不断重试这些错误，并且没有获取任何新内容。
例如，轮询了 500 个提要，其中 10 个引发异常，其中提要已被移动，因此不再存在，它的 xml 错误或可能发生了其他事情。现在轮询器被卡住了一遍又一遍地轮询这 10 个提要，直到它不再失败(这永远不会发生)。我记得在某处读到如果 handleError 中抛出异常函数然后它会继续重试，但如果没有抛出异常，它应该继续。这似乎不是这里的情况。我现在花了 3 天时间考虑这个问题，尝试了不同的解决方案，但它总是在大约 4-10 分钟后阻塞，这取决于获取足够多的错误提要需要多长时间。

Answer 1

我会说您尝试谈论许多不同的异常:不可用、格式错误、处理错误等。我认为最佳实践是区分这些异常并为每个异常做出适当的业务决策:我们绝对不能处理“不可用”作为处理错误，必须重试，直到它恢复可用。
对于处理，我建议查看 ExpressionEvaluatingRequestHandlerAdvice在您的提要处理条目所在的服务上:https://docs.spring.io/spring-integration/docs/current/reference/html/messaging-endpoints.html#message-handler-advice-chain
对于格式错误的源，我建议从上下文中查看流删除。因此，您不会再尝试轮询错误的来源。你绝对可以从那个 feedErrors 的错误流中做到这一点。 channel 。
对于您在这种情况下面临的其他错误，依此类推。
我们可能会修改您的10 are of those throw an exception where the feeds have been removed for whatever reason更详细，因为 FeedEntryMessageSource 中的逻辑基于 updatedDate的条目。所以，如果我们不能正确转换它并产生它，它真的会在下一次再次被拉动，因为 lastTime状态未适当更新。但是，让我们在单独的 SO 线程中执行此操作，并了解更多细节!
更新
关于您的堆栈跟踪的一些想法。
像 org.xml.sax.SAXParseException; lineNumber: 1; columnNumber: 15681; Character refer 这样的异常不好，必须当成致命的。您绝对无法再解析该 RSS 源，或者至少在它恢复正常之前无法解析。这种异常可以像 stop() 一样处理。对于那个Feed.inboundAdapter() .因此，您在 feedErrors 中捕获了一个异常。 channel 流量，调用stop()引起的端点。或者甚至更好地删除此 RSS 源的整个动态流。
异常 org.xml.sax.SAXParseException; lineNumber: 1; columnNumber: 10; DOCTYPE is disallowed when the feature "http://apache.org/xml/features/disallow-doctype-decl" set to true.也不好，可能会以相同的致命方式对待后续的stop()为端点。或者您可以调查 xerces库如何允许该 DTD 声明。可能像 syndFeedInput(SyndFeedInput) 这样的选项还可以为您提供一些用于 XML 解析的钩子(Hook)。例如，我看到它有这个选项:

/**
 * Since ROME 1.5.1 we fixed a security vulnerability by disallowing Doctype declarations by default. 
 * This change breaks the compatibility with at least RSS 0.91N because it requires a Doctype declaration. 
 * You are able to allow Doctype declarations again with this property. You should only activate it 
 * when the feeds that you process are absolutely trustful. 
 *  
 * @param allowDoctypes true when Doctype declarations should be allowed again, false otherwise
 */
public void setAllowDoctypes(boolean allowDoctypes) {