问题标签 [spring-batch-stream]

我有一个文件要阅读，如下所示。一条记录被分成多行。每条记录可以有任意数量的行；识别新记录的唯一方法是当一行以“ABC”开头并且另一行具有标识符 ABC_SUB。此记录的每一行都需要有一个单独的映射器，该映射器由以行开头的模式标识（例如，ABC、line2、line3、line、ABC_SUB、line3、line4） 可以存在相同的模式 line3 和 line4，但它们需要基于前一个线型标识符的不同映射器。在这个例子中，

1. line3 pattern(with mapper1) 存在于以 ABC 和开头的行之后
2. line3 模式（使用 mapper2）在以 ABC_SUB 开头的行之后退出。

如何识别 ABC 下的 line3 模式与 ABC_SUB 下的 line3 模式？

我尝试了 PatternMatchingCompositeTokenizer，但这给出了第一个匹配映射器。

有没有办法在识别子类型（如 ABC 或 ABC_SUB）并给出相应的映射器之前检查几行？

下面是 XML 配置

在 Spring 批处理应用程序中，我们如何在处理每个文件时将文件从 BatchFileDir 文件夹移动到目标文件夹。BatchFileDir 文件夹中会有一堆文件，所以我使用 MultiResourceItemReader。我也在关注基于块的处理。

我有所有的批处理文件：

项目阅读器是：

并且项目处理器是：

ModifiedItemProcessor 是：

步骤是：

我有一个场景，我需要大约 50-60 个不同的进程同时运行并执行一个任务。

每个进程都必须使用 sql 查询从数据库中获取数据，方法是传递一个值并获取要在后续任务中运行的数据。从 table_1 中选择 col_1、col_2、col_3 其中 col_1 = :Process_1;

由于我们在 Apache Camel 中有 Aggregator 和 parallelProcessing，Spring Batch 是否有任何类似的功能可以完成相同的工作？

我是 Spring Batch 的新手，目前正在探索它是否可以处理该卷。因为这将是一个 24*7 运行的重负载应用程序，并且每个进程都需要同时运行，其中每个线程应该能够支持进程内的多个线程。

有没有办法监视这些进程，以便它无论如何都会被终止，我应该能够重新启动该特定进程？请帮助解决这个问题。

我的批处理作业读取、处理和写入 DB 。ItemProcessor返回至少一项时，作业运行良好。如果它返回 null ，则ItemWriter @AfterStep抛出无法调用方法。

这是代码

下面的工作项处理器

以下 ItemProcessor 代码在 ItemWritter @Afterstep 中引发异常

为什么我用 null 进行测试？处理器有过滤条件。它是 - 在 ItemProcessor 案例测试中通过过滤跳过所有项目。实际场景中，100个item中，即使1个item没有被过滤掉，也不会抛出异常。只有在处理器中过滤了所有 100 个项目时，才会发生异常。

ItemWriter如果ItemProcessor过滤掉所有块的所有记录，是否有一个选项可以跳过。

类似于：在运行时更改 spring-cloud-stream 实例索引/计数

我在微服务器架构中启动批处理时做了一个 poc，我正在使用 Spring 批处理和 Spring Cloud Stream Kafka。我正在寻找一种方法来动态创建消费者（处理器）应用程序的多个实例。我看到可以定义多个实例

但是我还没有找到动态更改instanceIndex 值的方法。是否可以使用 Spring Cloud Stream kafka 动态修改此值。

谢谢你的帮助。

我正在尝试进行下一个实施。

由于大小原因，我必须将输出文件拆分为例如 10k 行块。

所以，我需要在文件“out1.csv”中转储 10k，在文件“out2.csv”中转储下一个 10k，依此类推。

使用一个输出文件，模式批处理：带有读取器-处理器-写入器的块是简单而直接的。

输出流是在块内的 batch:streams XML 部分中打开的，所以我避免了“Writer must be open before it can be write to”异常。

我想做一个避免这种严格和预设的解决方案的实现：

假设20个作家就足够了。我正在寻找一种解决方案来动态（可能以编程方式）创建输出编写器，打开它们并避免上述异常。

我spring-batch在spring-boot应用程序中使用。Spring Boot 版本是2.3.3.RELEASE.

我打算达到的目标

我必须read a xml file包含数千个Transactions（header tagfileInformation）。对事务执行一些业务逻辑，然后使用事务中的更新值将文件写回。我StaxEventItemReader用于读取文件和StaxEventItemWriter写入文件。然后我有几个ItemProcessors用于处理业务逻辑。XML 文件看起来像：

我面临的问题是标题标签的值。

我已经配置了OmegaXmlHeaderCallBack生成所需的标头标签，但这些标签中的值应该从输入文件中复制。据我所知，它StaxWriterCallback是在阅读器、处理器和编写器之前初始化的。所以我无法使用late binding. 这看起来像是一个基本要求，但在stackoverflow.

这是配置spring批处理作业的代码片段。

下面是 OmegaXmlHeaderCallBack课堂。由于没有后期绑定，我总是在标题标签中得到空值。

代码ItemProcessor如下。我正在设置ExecutionContextheaderCallback 打算读取的标题数据（遗憾的是不会发生）。

我引用的链接：

• 访问-the-executioncontext-values-in-headercallback
• 如何通过在 ItemReader 中修改 setFragmentRootElementNames 和 marshaler 来读取 header 标签

任何人都可以建议我需要遵循的方法来实现以下要求吗？

背景：

1. 有一个远程位置，例如"//severname/somefolder/somefile"。某些文件将在远程位置每隔一秒或一分钟连续生成。

2. 应该要求 Spring Boot 应用程序持续检查在远程位置（通过某些调度程序）生成的新文件。

3. 如果文件可用，我需要一个一个读取最旧到最新的文件并处理它们可能存储到数据库中。

4. 处理后，需要从现有远程位置删除特定文件并移动到另一个远程位置文件夹。

我心目中的几点：

1. 通过使用弹簧批处理，我们可以一次读取一个文件。但是我们如何动态读取最旧的文件呢？

2. 如何处理这样的场景：如果我的批处理正在处理其中一个文件仍在进行中。如果计划再次运行作业，则有可能选择相同的文件进行处理。

感谢您的解决方案和更好的建议:)

我对 Spring Batch 很陌生，并尝试使用单线程运行 Spring Batch。现在我需要逐步添加多线程并具有以下配置，但是并行处理在一段时间后挂起，并且在处理一些记录后控制台上没有任何跟踪。之前对于单线程，我使用 JdbcCursorItemReader，然后切换到 JdbcPagingItemReader 用于线程安全读取器。Reader 正在从 postgres DB 中读取条目，然后处理器（调用其他 rest web 服务并将响应返回给 writer）和 writer（在 DB 中创建新文件并更新状态数据）可以并行执行。

这里的多线程配置有什么问题？当使用 JdbcCursorItemReader 并且没有 TaskExecutor bean 时，批处理一次可以很好地处理单个记录：

我是新手，我需要你的帮助！

我想将异常捕获到流（写入器）或流（erreurWriter）中以保持信息项失败。
实际上，我正在使用监听器“skipListener”来执行此操作，它适用于阅读器、处理器和分类器编写器。
我该怎么做这个流？

谢谢！