java - Spring批处理聚合值并写入单个值

Question

我正在使用 spring batch，我需要实现以下目标

1. 读取包含日期和金额等详细信息的 csv 文件
2. 汇总同一日期的所有金额的总和
3. 保留一个带有日期和总和的条目

我过去使用过批处理，我想到了以下方法。创建包含 2 个步骤的批处理。

第 1 步:

1. Reader:使用 FlatFileItemReader 遍历整个文件
2. 处理器:使用键作为日期、值作为数量来填充 map 。如果条目存在，则获取该值并将其添加到新值
3. 作家:没有操作作家，因为我不想写

第 2 步:

1. 读者:遍历 map 的值
2. 作者:坚持值(value)观

我能够完成填充 Map 的第 1 步。此 Map 已使用 @JobScope 声明

我对如何为 step2 创建只需要读取值列表的读取器感到困惑。我尝试了 ListItemReader，但无法从 ListItemReader 访问 Map。

请提出解决方案，或者您是否有更好的方法来解决这个问题

谢谢

Answer 1

选项 1:如果你的 cvs 已经按日期排序，你可以实现一个组阅读器，它读取行直到键值改变。之后，整个组可以作为一个项目传递给处理器。

这样的群组阅读器可能如下所示:

  private SingleItemPeekableItemReader<I> reader;
  private ItemReader<I> peekReaderDelegate;

  @Override
  public void afterPropertiesSet() throws Exception {
    Assert.notNull(peekReaderDelegate, "The 'itemReader' may not be null");
    this.reader= new SingleItemPeekableItemReader<I>();
    this.reader.setDelegate(peekReaderDelegate);
  }

  @Override
  // GroupDTO is just a simple container. It is also possible to use
  // List<I> instead of GroupDTO<I>
  public GroupDTO<I> read() throws Exception {
    State state = State.NEW; // a simple enum with the states NEW, READING, and COMPLETE
    GroupDTO<I> group = null;
    I item = null;

    while (state != State.COMPLETE) {
      item = reader.read();

      switch (state) {
        case NEW: {
          if (item == null) {
            // end reached
            state = State.COMPLETE;
            break;
          }

          group = new GroupDTO<I>();
          group.addItem(item);
          state = State.READING;
          I nextItem = reader.peek();
          // isGroupBreak returns true, if 'item' and 'nextItem' do NOT belong to the same group
          if (nextItem == null || getGroupBreakStrategy.isGroupBreak(item, nextItem)) {
            state = State.COMPLETE;
          }
          break;
        }
        case READING: {
          group.addItem(item);

          // peek and check if there the peeked entry has a new date
          I nextItem = peekEntry();
          // isGroupBreak returns true, if 'item' and 'nextItem' do NOT belong to the same group
          if (nextItem == null || getGroupBreakStrategy.isGroupBreak(item, nextItem)) {
            state = State.COMPLETE;
          }
          break;
        }
        default: {
          throw new org.springframework.expression.ParseException(groupCounter, "ParsingError: Reader is in an invalid state");
        }
      }
    }

    return group;
  }

您需要一个 SingleItemPeekableItemReader，以便预读下一个元素。这个包装了您的普通阅读器。

选项 2:第一步如您所建议，但只需为步骤 2 编写一个 tasklet。无需使用读取器-进程-编写器方法，而是可以使用一个简单的 tasklet 将您的 map 内容写入文件。

选项 3:如果您真的想在第 2 步中使用阅读器-处理器-编写器方法，请编写您自己的阅读器来遍历您的 map 。

类似于(我没有测试该代码):

public class MapReader implements ItemReader {

     private MapContainer container;
     private Iterator<Map.Entry<Date, Integer> mapIterator;

     @PostConstruct
     public void afterPropertiesSet() {
        Assert.notNull(container);
        iterator = container.getMap().entry().iterator;
     }

     public void setMapContainer(MapContainer container) {
         this.container = container;
     }

     public Map.Entry<Date, Integer> read() {
        if (iterator.hasNext()) {
           return iterator.next();
        }
        return null;
      }
}

@Component
public class MapContainer {
    private Map<Date, Integer> data = new Hashmap<>();

    public Map<Date, Integer> getMap() {
        return data;
    }

    // add modifier method as needed for step 1

}

因此，您为容器创建了一个单独的 spring-bean 实例，将其注入(inject)到步骤 2 的处理器中，在那里填充它，同时将其注入(inject)到上面的读取器中。