Spring Batch 3.0 – Часть 2: Кастомный ItemReader, ItemPocessor и ItemWriter

В этой статье мы напишем свою реализацию интерфейсов ItemReader, ItemPocessor и ItemWriter, а так же сервис для управления всеми задачами в приложении. Написание собственных реализаций рассматриваемых интерфейсов может быть полезно для оптимизации и ускорения конкретной пакетной обработки.

Скачайте репозиторий и переключитесь на ветку step1:

git clone https://github.com/JavaGrinko/batch-example.git
git checkout -f step1

1 2	git clone https://github.com/JavaGrinko/batch-example.git git checkout -f step1

Интерфейс ItemReader работает как итератор и используется как источник данных, которые необходимо обработать:

public interface ItemReader<t> {
T read() throws Exception, ...;
}

public interface ItemReader<t> {

T read() throws Exception, ...;

}

Источник данных ItemReader за одну итерацию должен возвращать один объект T, в нашем примере это один товар Product. Процесс итерирования продолжается до тех пор, пока метод read() не вернет null, тогда выполняется последняя обработка, последнее сохранение, и шаг заканчивается.

В последней версии Spring Batch 3.0.7 поставляются стандартные реализации ItemReader:

AggregateItemReader, AmqpItemReader, FlatFileItemReader, HibernateCursorItemReader, HibernatePagingItemReader, IbatisPagingItemReader, ItemReaderAdapter, JdbcCursorItemReader, JdbcPagingItemReader, JmsItemReader, JpaPagingItemReader, ListItemReader, MongoItemReader, Neo4jItemReader, RepositoryItemReader, StoredProcedureItemReader, StaxEventItemReader.

Для нашей задачи больше всего подходит FlatFileItemReader, но нам не нужна вся универсальность этой реализации и для увеличения производительности мы напишем свой узкопрофильный CsvItemReader:

package javagrinko.batch.example.batch;

import lombok.Data;
import lombok.extern.log4j.Log4j;
import org.springframework.batch.core.JobParameter;
import org.springframework.batch.core.annotation.AfterStep;
import org.springframework.batch.core.annotation.BeforeStep;
import org.springframework.batch.item.ItemReader;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.convert.ConversionService;
import org.springframework.core.io.ClassPathResource;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Scanner;

@Data
public class CsvItemReader<t> implements ItemReader {</t>

@Autowired
private ConversionService conversionService;
private File source;
private Scanner scanner;
private Class<t> typeParameterClass;</t>

public CsvItemReader(Class<t> typeParameterClass, String fileName) {
try {
source = new ClassPathResource(fileName).getFile();
} catch (IOException e) {
e.printStackTrace();
}
this.typeParameterClass = typeParameterClass;
}</t>

@BeforeStep
public void open() throws FileNotFoundException {
scanner = new Scanner(source);
}

@Override
public Object read() throws Exception {
if (scanner.hasNextLine()) {
String s = scanner.nextLine();
return conversionService.convert(s, typeParameterClass);
} else {
return null;
}
}

@AfterStep
public void destroy() {
scanner.close();
}
}

package javagrinko.batch.example.batch;

import lombok.Data;

import lombok.extern.log4j.Log4j;

import org.springframework.batch.core.JobParameter;

import org.springframework.batch.core.annotation.AfterStep;

import org.springframework.batch.core.annotation.BeforeStep;

import org.springframework.batch.item.ItemReader;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.core.convert.ConversionService;

import org.springframework.core.io.ClassPathResource;

import javax.annotation.PostConstruct;

import javax.annotation.PreDestroy;

import java.io.File;

import java.io.FileNotFoundException;

import java.io.IOException;

import java.util.Scanner;

@Data

public class CsvItemReader<t> implements ItemReader {</t>

@Autowired

private ConversionService conversionService;

private File source;

private Scanner scanner;

private Class<t> typeParameterClass;</t>

public CsvItemReader(Class<t> typeParameterClass, String fileName) {

try {

source = new ClassPathResource(fileName).getFile();

} catch (IOException e) {

e.printStackTrace();

}

this.typeParameterClass = typeParameterClass;

}</t>

@BeforeStep

public void open() throws FileNotFoundException {

scanner = new Scanner(source);

}

@Override

public Object read() throws Exception {

if (scanner.hasNextLine()) {

String s = scanner.nextLine();

return conversionService.convert(s, typeParameterClass);

} else {

return null;

}

@AfterStep

public void destroy() {

scanner.close();

}

Методы, аннотированные @BeforeStep, будут выполнены 1 раз перед запуском всего шага. Аналогично, метод destroy с аннотацией @AfterStep будет вызван после завершения шага вне зависимости от результата.

Для преобразования строки в объект мы используем не рефлексивный маппинг, а conversionService, который работает быстрее рефлексии. Для работы conversionService нужно заранее зарегистрировать в него конвертер строки в продукт StringProductConverter:

@Service
public class StringProductConverter implements Converter<string, product=""> {
@Override
public Product convert(String source) {
String[] split = source.split(",");
Product product = new Product();
product.setId(Long.parseLong(split[0]));
product.setName(split[1]);
product.setDescription(split[2]);
product.setPrice(Double.parseDouble(split[3]));
return product;
}
}

@Service

public class StringProductConverter implements Converter<string, product=""> {

@Override

public Product convert(String source) {

String[] split = source.split(",");

Product product = new Product();

product.setId(Long.parseLong(split[0]));

product.setName(split[1]);

product.setDescription(split[2]);

product.setPrice(Double.parseDouble(split[3]));

return product;

}

Теперь напишем реализацию интерфейса ItemProcessor:

public interface ItemProcessor<i, o=""> {
O process(I item) throws Exception;
}

public interface ItemProcessor<i, o=""> {

O process(I item) throws Exception;

}

В нашем примере название товара мы переводим в ВЕРХНИЙ РЕГИСТР:

package javagrinko.batch.example.batch;

import javagrinko.batch.example.model.Product;
import org.springframework.batch.item.ItemProcessor;

public class SimpleProcessor implements ItemProcessor<product, product=""> {
@Override
public Product process(Product item) throws Exception {
item.setName(item.getName().toUpperCase());
return item;
}
}

package javagrinko.batch.example.batch;

import javagrinko.batch.example.model.Product;

import org.springframework.batch.item.ItemProcessor;

public class SimpleProcessor implements ItemProcessor<product, product=""> {

@Override

public Product process(Product item) throws Exception {

item.setName(item.getName().toUpperCase());

return item;

}

И последнее, что нужно сделать – это реализовать интерфейс ItemWriter:

public interface ItemWriter<t> {
void write(List<!--? extends T--> items) throws Exception;
}

public interface ItemWriter<t> {

void write(List items) throws Exception;

}

В последней версии Spring Batch 3.0.7 поставляются стандартные реализации ItemWriter:

AmqpItemWriter, CompositeItemWriter, FlatFileItemWriter, GemfireItemWriter, HibernateItemWriter, IbatisBatchItemWriter, ItemWriterAdapter, JdbcBatchItemWriter, JmsItemWriter, JpaItemWriter, MimeMessageItemWriter, MongoItemWriter, Neo4jItemWriter, PropertyExtractingDelegatingItemWriter, RepositoryItemWriter, StaxEventItemWriter.

Для записи в базу данных мы будем использовать JdbcTemplate:

package javagrinko.batch.example.batch;

import javagrinko.batch.example.model.Product;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;

import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

public class JdbcItemWriter implements ItemWriter<product> {</product>

@Autowired
public JdbcTemplate jdbcTemplate;

@Override
public void write(List<!--? extends Product--> items) throws Exception {
StringBuilder values = new StringBuilder(" VALUES ");
for (int i = 0; i &lt; items.size(); i++) {
            values.append(" (?, ?, ?, ?)");
            if (i &lt; items.size() - 1) values.append(", ");
        }
        List<object> arguments = items.stream().map(p -&gt; new Object[]{p.getId(), p.getName(), p.getDescription(), p.getPrice()}).flatMap(objects -&gt; Arrays.asList(objects).stream()).collect(Collectors.toList());String sql = "INSERT INTO PRODUCTS (PRODUCT_ID, NAME, DESCRIPTION, PRICE)" + values.toString();jdbcTemplate.update(sql, arguments.toArray());}}

package javagrinko.batch.example.batch;

import javagrinko.batch.example.model.Product;

import org.springframework.batch.item.ItemWriter;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.jdbc.core.JdbcTemplate;

import java.util.Arrays;

import java.util.List;

import java.util.stream.Collectors;

public class JdbcItemWriter implements ItemWriter<product> {</product>

@Autowired

public JdbcTemplate jdbcTemplate;

@Override

public void write(List items) throws Exception {

StringBuilder values = new StringBuilder(" VALUES ");

for (int i = 0; i < items.size(); i++) {

values.append(" (?, ?, ?, ?)");

if (i < items.size() - 1) values.append(", ");

}

List<object> arguments = items.stream().map(p -> new Object[]{p.getId(), p.getName(), p.getDescription(), p.getPrice()}).flatMap(objects -> Arrays.asList(objects).stream()).collect(Collectors.toList());String sql = "INSERT INTO PRODUCTS (PRODUCT_ID, NAME, DESCRIPTION, PRICE)" + values.toString();jdbcTemplate.update(sql, arguments.toArray());}}

Здесь формируется один запрос к базе данных на вставку всех элементов списка items. Items – это пачка (chunk) записей, полученная в результате многократного вызова ItemReader.read(). Размер items задается параметром chunk в StepBuilder (см. предыдущую часть). Особенностью данной реализации является отсутствие транзакции, что экономит немного ресурсов.

Теперь, модифицируем код из первой части руководства и заменим стандартные реализации ItemReader и ItemWriter на наши собственные, а так же заменим лямбда-выражение на SimpleProcessor. В результате получим код, который можно посмотреть в репозитории на шаге step1:

git clone https://github.com/JavaGrinko/batch-example.gitgit checkout -f step1

1	git clone https://github.com/JavaGrinko/batch-example.gitgit checkout -f step1

Теперь, когда всё готово, напишем небольшой сервис, который будет искать нашу задачу в реестре задач приложения и запускать ее на выполнение. Для начала определим интерфейс:

public interface JobService {void start(String jobName);}

1	public interface JobService {void start(String jobName);}

Реализация выглядит следующим образом:

package javagrinko.batch.example.service;import lombok.extern.log4j.Log4j;import org.springframework.batch.core.Job;import org.springframework.batch.core.JobParameters;import org.springframework.batch.core.configuration.JobRegistry;import org.springframework.batch.core.launch.JobLauncher;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Service;@Service@Log4jpublic class JobServiceImpl implements JobService {@Autowiredprivate JobLauncher jobLauncher;@Autowiredprivate JobRegistry jobRegistry;@Overridepublic void start(String jobName) {try {Job job = jobRegistry.getJob(jobName);jobLauncher.run(job, new JobParameters());} catch (Exception e) {e.printStackTrace();}}}

package javagrinko.batch.example.service;import lombok.extern.log4j.Log4j;import org.springframework.batch.core.Job;import org.springframework.batch.core.JobParameters;import org.springframework.batch.core.configuration.JobRegistry;import org.springframework.batch.core.launch.JobLauncher;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Service;@Service@Log4jpublic class JobServiceImpl implements JobService {@Autowiredprivate JobLauncher jobLauncher;@Autowiredprivate JobRegistry jobRegistry;@Overridepublic void start(String jobName) {try {Job job = jobRegistry.getJob(jobName);jobLauncher.run(job, new JobParameters());} catch (Exception e) {e.printStackTrace();}}}

Здесь используется стандартный бин JobRegistry, который знает обо всех задачах, определенных в проекте, и умеет возвращать экземпляр Job по названию задачи. Для использования бина JobRegistry необходимо в конфигурационный файл добавить следующие bean definitions:

@Beanpublic JobRegistry jobRegistry() {return new MapJobRegistry();}@Beanpublic JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor() {JobRegistryBeanPostProcessor bpp = new JobRegistryBeanPostProcessor();bpp.setJobRegistry(jobRegistry());return bpp;}

1	@Beanpublic JobRegistry jobRegistry() {return new MapJobRegistry();}@Beanpublic JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor() {JobRegistryBeanPostProcessor bpp = new JobRegistryBeanPostProcessor();bpp.setJobRegistry(jobRegistry());return bpp;}

В следующей части мы поговорим о передаче данных между шагами и о перезапуске пакетной задачи.

Spring Batch 3.0 – Часть 2: Кастомный ItemReader, ItemPocessor и ItemWriter