在secuence上有孔的重新排序器 [英] resequencer with holes on secuence

问题描述

我们有一个使用重排序器的 ETL 场景.

we have an ETL scenario where we use the resequencer.

消息到达流时带有一个序列号，重新排序器使用它来按顺序发送消息，但有时消息之前被丢弃(因为数据验证)并且没有到达重新排序器.这会在序列中产生漏洞，并且重新排序器使用默认发布策略停止发送消息.为了避免这种情况，我们开发了一个新的 SequenceTimeoutReleaseStrategy，它混合了默认策略和 SI 的 TimeoutCountSequenceSizeReleaseStrategy.当消息到达时，它会检查超时并在必要时释放它.

Messages arrive to the flow with a sequence number that the resequencer uses it to send messages in order, but sometimes messages are discarded previously (because of data validation) and do not arrive to the resequencer. This produces holes in the sequence and resequencer stops sending messages using the default release strategy. To avoid this, we developed a new SequenceTimeoutReleaseStrategy that is a mix between default strategy and TimeoutCountSequenceSizeReleaseStrategy from SI. When a message arrives, it checks the timeout and release it if necesary.

除非最后一条消息在超时之前到达并且有漏洞，否则所有这些都运行良好.此消息不会由策略发布.我们可以使用收割者，但序列中的序列中可能有多个孔，因此当重新排序器释放它们时，它将在第一个序列中断处停止并删除丢失其余消息的组.所以，问题是:有没有办法在序列中可能存在漏洞的地方使用重测序器?

All this worked well unless for the last messages that arrive before the timeout and have holes. This messages aren't release by the strategy. We could use a reaper but the secuence may have more than one hole in the sequence so when the resequencer release them it will stop in the first sequence break and remove the group losing the rest of the messages. So, the question is: is there a way to use the resequencer where there can be holes in the sequence?

我们拥有并希望避免的一种解决方案是使用计划任务直接从消息存储中删除消息，但这可能是并发性等方面的问题，因此我们更喜欢其他解决方案.

One solution we have and want to avoid is having a scheduled tasks that removes the messages directly from the message store, but this could be a problem with concurrency and so on, so we prefer other solutions.

这里感谢任何帮助

问候古兹曼

推荐答案

涉及两个组件；发布策略说某物"可以发布；MessageGroupProcessor 执行关于释放什么的实际决定.在这种情况下，一个 ResequencingMessageGroupProcessor.

There are two components involved; the release strategy says "something" can be released; the actual decision as to what is released is performed by the MessageGroupProcessor. In this case, a ResequencingMessageGroupProcessor.

您需要自定义该类以跳过"漏洞.

You would need to customize that class to "skip" the hole(s).

您不能使用 <reseequencer/> 命名空间连接自定义 MGP，您必须使用 <bean/> 进行连接- 一个 ResequencingMessageHandler 和一个 ConsumerEndpointFactoryBean.

You can't wire in a customized MGP using the <reseequencer/> namespace, you would have to wire up using <bean/> s - a ResequencingMessageHandler and a ConsumerEndpointFactoryBean.

或者使用 BeanFactoryPostProcessor 将构造函数参数更改为您的自定义类.

Or use a BeanFactoryPostProcessor to change the constructor argument to your custom class.

这篇关于在secuence上有孔的重新排序器的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！