1

2

3

4

5

<dependency>

    <groupId>org.springframework.batch</groupId>

    <artifactId>spring-batch-core</artifactId>

    <version>3.0.8.RELEASE</version>

</dependency>

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

<beans xmlns="http://www.springframework.org/schema/beans"

       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

       xsi:schemaLocation="

http://www.springframework.org/schema/beans

http://www.springframework.org/schema/beans/spring-beans.xsd">

    <!-- 事务管理器 -->

    <bean id="transactionManager" class="org.springframework.batch.support.transaction.ResourcelessTransactionManager"/>

    <!-- 任务仓库 -->

    <bean id="jobRepository" class="org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean">

        <property name="transactionManager" ref="transactionManager"/>

    </bean>

    <!-- 任务加载器 -->

    <bean id="jobLauncher" class="org.springframework.batch.core.launch.support.SimpleJobLauncher">

        <property name="jobRepository" ref="jobRepository"/>

    </bean>

</beans>

1

2

3

4

5

6

7

8

9

10

1,PENDING

2,PENDING

3,PENDING

4,PENDING

5,PENDING

6,PENDING

7,PENDING

8,PENDING

9,PENDING

10,PENDING

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

public class  DeviceCommand {

    private String id;

    private String status;

    public String getId() {

        return id;

    }

    public void  setId(String id) {

        this.id = id;

    }

    public String getStatus() {

        return status;

    }

    public void  setStatus(String status) {

        this.status = status;

    }

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

public class  HelloLineMapper implements LineMapper<DeviceCommand> {

    @Override

    public DeviceCommand mapLine(String line, int lineNumber) throws Exception {

        // 逗号分割每一行数据

        String[] args = line.split(",");

        // 创建DeviceCommand对象

        DeviceCommand deviceCommand = new DeviceCommand();

        // 设置id值到对象中

        deviceCommand.setId(args[0]);

        // 设置status值到对象中

        deviceCommand.setStatus(args[1]);

        // 返回对象

        return deviceCommand;

    }

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

public class  HelloItemProcessor implements ItemProcessor<DeviceCommand, DeviceCommand> {

    @Override

    public DeviceCommand process(DeviceCommand deviceCommand) throws Exception {

        // 模拟下发命令给设备

        System.out.println("send command to device, id="  + deviceCommand.getId());

        // 更新命令状态

        deviceCommand.setStatus("SENT");

        // 返回命令对象

        return deviceCommand;

    }

}

1

2

3

4

5

6

7

8

9

10

11

12

13

public class  HelloLineAggregator implements LineAggregator<DeviceCommand> {

    @Override

    public String aggregate(DeviceCommand deviceCommand) {

        StringBuffer sb = new StringBuffer();

        sb.append(deviceCommand.getId());

        sb.append(",");

        sb.append(deviceCommand.getStatus());

        return sb.toString();

    }

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

public class  Main {

    public static  void  main(String[] args) throws Exception {

        // 加载上下文

        String[] configLocations = {"applicationContext.xml"};

        ApplicationContext applicationContext = new ClassPathXmlApplicationContext(configLocations);

        // 获取任务启动器

        JobLauncher jobLauncher = applicationContext.getBean(JobLauncher.class);

        JobRepository jobRepository = applicationContext.getBean(JobRepository.class);

        PlatformTransactionManager transactionManager = applicationContext.getBean(PlatformTransactionManager.class);

        // 创建reader

        FlatFileItemReader<DeviceCommand> flatFileItemReader = new FlatFileItemReader<>();

        flatFileItemReader.setResource(new FileSystemResource("src/main/resources/batch-data.csv"));

        flatFileItemReader.setLineMapper(new HelloLineMapper());

        // 创建processor

        HelloItemProcessor helloItemProcessor = new HelloItemProcessor();

        // 创建writer

        FlatFileItemWriter<DeviceCommand> flatFileItemWriter = new FlatFileItemWriter<>();

        flatFileItemWriter.setResource(new FileSystemResource("src/main/resources/batch-data.csv"));

        flatFileItemWriter.setLineAggregator(new HelloLineAggregator());

        // 创建Step

        StepBuilderFactory stepBuilderFactory = new StepBuilderFactory(jobRepository, transactionManager);

        Step step = stepBuilderFactory.get("step")

                                      .<DeviceCommand, DeviceCommand>chunk(1)

                                      .reader(flatFileItemReader)       // 读操作

                                      .processor(helloItemProcessor)    // 处理操作

                                      .writer(flatFileItemWriter)       // 写操作

                                      .build();

        // 创建Job

        JobBuilderFactory jobBuilderFactory = new JobBuilderFactory(jobRepository);

        Job job = jobBuilderFactory.get("job")

                                   .start(step)

                                   .build();

        // 启动任务

        jobLauncher.run(job, new JobParameters());

    }

}

1

2

3

4

5

6

7

8

9

10

send command to device, id=1

send command to device, id=2

send command to device, id=3

send command to device, id=4

send command to device, id=5

send command to device, id=6

send command to device, id=7

send command to device, id=8

send command to device, id=9

send command to device, id=10

1

2

3

4

5

6

7

8

9

10

1,SENT

2,SENT

3,SENT

4,SENT

5,SENT

6,SENT

7,SENT

8,SENT

9,SENT

10,SENT

1

2

3

4

5

public interface  JobLauncher {

    public JobExecution run(Job job, JobParameters jobParameters) throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException, JobParametersInvalidException;

}

1

2

3

4

5

public interface  ItemReader<T> {

    T read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException;

}

1

2

3

4

5

public interface  ItemProcessor<I, O> {

    O process(I item) throws Exception;

}

1

2

3

4

5

public interface  ItemWriter<T> {

    void write(List<? extends T> items) throws Exception;

}