Processamento Batch em JavaEE 7

As aplicações Batch para a plataforma Java estão especificadas na JSR 352. A especificação completa pode ser baixada no site da Oracle.

Um processamento em Batch é a execução de uma série de jobs ou tarefas sendo adequado para ambientes que não precisem de interatividade, e que possua tarefas de longa execução. Exemplos típicos para esse tipo de processamento são: geração de dados consolidados no final do mês, cálculos de interesse de uma organização ao final do dia, carga de dados e o processo de ETL (extract-transform-load) em um data warehouse. Essas tarefas são computacionalmente intensivas, executam sequencialmente ou em paralelo, e podem ser iniciadas através de vários modelos de invocação como ad hoc, agendada ou sob demanda.

A especificação também define um modelo de programação para aplicações batch e um runtime utilizado para agendamento e execução de jobs.

Segue abaixo alguns dos principais conceitos dessa arquitetura:

• Um job é definido como uma entidade que encapsula um processo batch inteiro. Um job é tipicamente colocado junto com um Job Specification Language e consiste de uma ou mais etapas ou steps. Como veremos no decorrer do artigo um Job Specification Language na plataforma JavaEE 7 é implementado com XML e é referenciado como “Job XML”.
• Um step é um objeto do domínio que encapsula uma fase independente e sequencial de um job. Um step contém tudo que é necessário para definir e controlar o processamento do batch atual.
• O JobOperator fornece uma interface para gerenciar todos os aspectos do processamento de um job, incluindo comandos operacionais como start, restart e stop, além de comandos de repositório para jobs.
• Um JobRepository contém informações sobre jobs sendo executados e jobs que foram executados no passado. O JobOperator fornece acesso a este repositório.
• O padrão "Reader-Processor-Writer" é o padrão primário e usa o estilo de processamento orientado a bloco, na qual ItemReader é uma abstração que representa a recuperação de uma entrada para um step, ItemProcessor é uma abstração que representa o processamento de negócio de um item, e ItemWriter é uma abstração que representa a saída de um step.

Veremos no restante do artigo como se dá o funcionamento dos batches na plataforma JavaEE 7 e outros conceitos relacionados.

Processamento Orientado Por Bloco (Chunk-Oriented)

O processamento Orientado Por Bloco (Chunk-Oriented) é o padrão primário para processamento em Batch na especificação. Este padrão é orientado a um item de processamento onde múltiplos itens são lidos e processados para criar "pedaços" que são então escritos na saída, tudo dentro de uma única transação. Assim, temos que neste tipo de processamento procura-se quebrar um grande processo em pequenos pedaços ou blocos (ou chunks).

A interface ItemReader é usada para ler um fluxo de itens, sendo um item por vez. Um ItemReader fornece um indicador quando não há mais itens a serem fornecidos. A interface ItemProcessor opera em um item de entrada e produz um item de saída através da transformação ou da aplicação de outros processos de negócios. Um ItemProcessor entrega o item processado ao ItemWriter para agregação. A interface ItemWriter é utilizada para escrever um fluxo de "pedaços" de itens agregados.

Geralmente, um item escritor não tem conhecimento das próximas informações que ele receberá, apenas do item atual.

As classes abstratas AbstractItemReader e AbstractItemWriter fornecem implementações para métodos menos comumente implementados.

Um registro de entrada pode ser definido da mesma maneira que a apresentada na Listagem 1.

Listagem 1. Exemplo de definição de um registro de entrada.

  public class MeuRegistroEntrada {
                  private String s;
   
                  public MeuRegistroEntrada() { }
   
                  public MeuRegistroEntrada(String s) {
                                 this.s = s;;
                  }
   
                  //Mais códigos aqui
   
  }

MeuRegistroEntrada definido acima é um item que é lido da fonte de entrada.

Um registro de saída pode ser definido da mesma maneira que a apresentada na Listagem 2.

Listagem 2. Exemplo de definição de um registro de saída.

  public class MeuRegistroSaida {
                  private String s;
   
                  public MeuRegistroSaida() { }
   
                  public MeuRegistroSaida(String s) {
                                 this.s = s;
                  }
   
                  //Mais códigos aqui
   
  }

MeuRegistroSaida é um item que é gerado após o processamento de um item.

As classes MeuRegistroEntrada e MeuRegistroSaida parecem ser muito semelhantes neste caso, no entanto eles poderiam ser bem diferentes dependendo da aplicação.

Um Job XML é usado para definir um chunk de um step, ou literalmente um pedaço de uma etapa, conforme podemos verificar no exemplo da Listagem 3.

Listagem 3. Exemplo de um Job XML definindo um chunk e um step com seus respectivos atributos.

  <job id="meuJob" xmlns="http://xmlns.jcp.org/xml/ns/javaee" version="1.0">
                  <step id="meuStep">
                                 <chunk item-count="3">
                                                 <reader ref="meuLeitorDeItem"/>
                                                 <processor ref="meuProcessadorDeItem"/>
                                                 <writer ref="meuEscritorDeItem"/>
                                 </chunk>
                  </step>
  </job>

Neste código podemos verificar que o elemento job identifica um job que tem um nome lógico (id) usado para identificação. Um job pode conter qualquer número de etapas identificadas pelo elemento "step". Cada "step" tem um nome lógico (id) usado para identificação. O elemento "chunk" define um pedaço da etapa. Este pedaço é periodicamente verificado de acordo com uma política de verificação configurada. Por padrão, a política de verificação é o "item", o que significa que o chunk é verificado após um determinado número de itens lidos/processados/escrito. Podemos especificar um valor "customizado" usando o atributo "checkpoint-policy". O "item-count" especifica o número de itens a serem processados por "chunk". O valor padrão é 10. Este atributo é ignorado para a política de verificação customizada. Também podemos usar o atributo “item-count” para definir os limites da transação. Além disso, temos neste código o meuLeitorDeItem que é identificado como o leitor, sendo que o seu valor é o nome do bean CDI de uma classe que implementa a interface ItemReader ou estende a classe abstrata AbstractItemReader. Também temos o meuProcessadorDeItem que é identificado como o processador, sendo seu valor o nome do bean CDI de uma classe que implementa a interface ItemProcessor. Este elemento é opcional. Se este item não for especificado, então todos os elementos do item leitor são passados para o item escritor para agregação. Por fim, meuEscritorDeItem é identificado como o escritor, sendo que o seu valor é o nome do bean CDI de uma classe que implementa a interface ItemWriter ou estende a classe AbstractItemWriter.

O leitor de item é uma implementação da interface ItemReader ou estende a classe AbstractItemReader conforme exemplificado na Listagem 4.

Listagem 4. Exemplo de um leitor de item.

  @Named
  public class MeuLeitorDeItem extends AbstractItemReader {
   
                  List<String> list;
                  
                  @Override
                  public void open(Serializable c) throws Exception {
                                 list = ...
                  }
   
                  @Override
                  public MeuRegistroEntrada readItem() {
                                 for (String s : list) {
                                                 return new MeuRegistroEntrada(s);
                                 }
                                 return null;
                  }
   
  }

Neste código tornamos MeuLeitorDeItem um leitor de item apenas estendendo a classe AbstractItemReader. Entre os métodos utilizados e sobrescritos temos o método open que prepara o leitor para ler itens. Neste método List é inicializado e o parâmetro de entrada "c" representa a última verificação para este leitor em uma dada instância do job. Os dados do ponto de verificação são definidos por este leitor e fornecidos através do método checkpointInfo. Estes dados também fornecem ao leitor todas as informações necessárias para retomar a leitura dos itens após o reinício. Um valor nulo do ponto de verificação será transmitido na inicialização. Além disso, temos o método readItem que retorna o próximo item para ser processado. Para todas as strings lidas do List, uma nova instância de MeuregistroEntrada é criada e retornada do método readItem. Retornando um null indica o fim do processamento. A anotação @Named assegura que este bean pode ser referenciado em um Job XML.

O processador de item é uma implementação da interface ItemProcessor conforme mostra o exemplo da Listagem 5.

Listagem 5. Exemplo de um processador de item.

  @Named
  public class MeuProcessadorDeItem implements ItemProcessor {
                  @Override
                  public MeuRegistroSaida processItem(Object t) {
                                 MeuRegistroSaida o = new MeuRegistroSaida();
                                 //Mais código aqui
                                 return o;
                  }
  }

Neste código tornamos MeuProcessadorDeItem um processador de itens apenas implementando a interface ItemProcessor. O método processItem aceita uma entrada do leitor de item e retorna uma saída que é passada ao escritor para agregação. Neste caso, o método recebe um item do tipo MeuRegistroEntrada, aplicado a lógica de negócio, e retorna um item de saída do tipo MeuRegistroSaida. O item de saída é então agregado e escrito. Retornando null indica que o item não deveria continuar sendo processado. Este processador permite que itens de entrada indesejados sejam filtrados por processItem, por isso afirma-se que o processador de itens está entre o step e o chunk. A anotação @Named assegura que este bean pode ser referenciado em um Job XML.

O escritor de itens implementa a interface ItemWriter ou estende a classe AbstractItemWriter conforme mostra o exemplo da Listagem 6.

Listagem 6. Exemplo de um escritor de item.

  @Named
  public class MeuEscritorDeItem extends AbstractItemWriter {
                  @Override
                  public void writeItems(List list) {
                                 //Mais código aqui
                  }
  }

Neste código tornamos MeuEscritorDeItem um escritor de itens apenas estendendo a classe AbstractItemWriter. O método writeItems recebe os itens agregados e implementa a lógica de escrita para o escritor de item. Uma lista de MeuRegistroSaida é recebida. A anotação @Named assegura que este bean pode ser referenciado em um Job XML.

Se o Job XML é definido em um arquivo meuJob.xml e empacotado no diretório META-INF/batch-jobs, então podemos iniciar este job usando JobOperator conforme mostra o exemplo da Listagem 7.

Listagem 7. Exemplo utilizando JobOperator para iniciar o job.

  JobOperator jo = BatchRuntime.getJobOperator();
  long jid = jo.start("meuJob", new Properties());

Neste código JobOperator fornece uma interface para operação de jobs. O método start cria uma nova instância de um job e inicia a primeira execução desta instância. O Job XML deve estar disponível no diretório META-INF/batch-jobs para arquivos ".jar" ou no diretório WEB-INF/classes/META-INF/batch-jobs para arquivos ".war". Arquivos Job XML seguem a convenção de nomeação “.xml”. O método retorna o id de execução para a primeira instância.

Já que este job pode ser executado explicitamente ele pode ser executado através de um Servlet, EJB, entre outros. Segue na Listagem 8 um exemplo de um job sendo explicitamente executado num Servlet:

Listagem 8. Chamando um job através de um Servlet.

  protected void processRequest(HttpServletRequest request, HttpServletResponse response)
                  throws ServletException, IOException {
   
                  response.setContentType("text/html;charset=UTF-8");
                  try (PrintWriter out = response.getWriter()) {
                                 out.println("<html>");
                                 out.println("<head>");
                                 out.println("<title>Exemplo de Invocação de Job - Servlet</title>");
                                 out.println("</head>");
                                 out.println("<body>");
                                 out.println("<h1>Invocando Jobs no Servlet:</h1>");
   
                                 JobOperator jo = BatchRuntime.getJobOperator();
                                 long jid = jo.start("meuJob", new Properties());
   
                                 out.println("Job submetido: " + jid + "<br>");
                                 out.println("</body>");
                                 out.println("</html>");
                  } catch (JobStartException | JobSecurityException ex) {
                                 Logger.getLogger(TestServlet.class.getName()).log(Level.SEVERE, null, ex);
                  }
  }

Também podemos reiniciar o job usando o método JobOperator.restart conforme mostra o exemplo da Listagem 9.

Listagem 9. Reiniciando um job.

  jo.restart(jid, props);

Neste código, reiniciamos uma instância particular de um job. Um novo conjunto de propriedades pode ser especificado quando o job é reiniciado.

Podemos cancelar o job utilizando o método JobOperator.abandon conforme mostra o exemplo da Listagem 10.

Listagem 10. Cancelando um job.

  jo.abandon(jid);

Neste código, a execução do job é utilizada para cancelar uma instância particular de um job.

Podemos obter informações de um job que está executando como mostra o exemplo da Listagem 11.

Listagem 11. Obtendo informações de um job em execução.

  JobExecution jexec = jo.getJobExecution(jid);
  Date createTime = jexec.getCreateTime();
  Date startTime = jexec.getStartTime();

Outra possibilidade é podermos especificar um conjunto diferente de propriedades durante múltiplas execuções do mesmo job.

O número de instâncias de um job com um nome particular pode ser encontrado conforme demonstra o exemplo da Listagem 12.

Listagem 12. Obtendo o número de instância de um job com um nome especificado.

  int conta = jo.getJobInstanceCount("meuJob");

Neste código, teremos como retorno o número de instâncias de meuJob submetidos por esta aplicação, executando ou não.

Todos os jobs com nomes conhecidos pelo batch em tempo de execução podem ser obtidos conforme mostra o exemplo da Listagem 13.

Listagem 13. Obtendo todos os jobs.

  Set<String> jobs = jo.getJobNames();

Este código retorna o nome único do conjunto de jobs desta aplicação.

Pontos de Verificação Customizáveis

Pontos de verificação permitem marcarmos periodicamente o progresso atual de um step para que possamos habilitar o reinicio da execução a partir do último ponto de consistência, seguindo uma interrupção planejada ou não. Por padrão, o final do processamento de cada "chunk" é um ponto natural para um ponto de verificação.

Podemos especificar uma política para um ponto de verificação customizável usando o atributo checkpoint-policy em Job XML, conforme mostra a Listagem 14.

Listagem 14. Especificando uma política para um ponto de verificação customizado.

  <chunk item-count="3" checkpoint-policy="custom">
                  <reader ref="meuLeitorDeItem"/>
                  <processor ref="meuProcessadorDeItem"/>
                  <writer ref="meuEscritorDeItem"/>
                  <checkpoint-algorithm ref="meuAlgoritmoCheckpoint"/>
  </chunk>

Neste fragmento do Job XML o valor de checkpoint-policy é especificado como "custom", indicando que um algoritmo de ponto de verificação customizável é utilizado. checkpoint-algorithm é um subelemento dentro do "chunk" em "step" cujo valor é um bean CDI que implementa a interface CheckpointAlgorithm ou estende a classe AbstractCheckpointAlgorithm. Segue na Listagem 15 um exemplo estendo a classe AbstractCheckpointAlgorithm.

Listagem 15. Implementando meuAlgoritmoCheckpoint especificado no Job XML.

  public class MeuAlgoritmoCheckpoint extends AbstractCheckpointAlgorithm {
                  @Override
                  public boolean isReadyToCheckpoint() throws Exception {
                                 if (MeuLeitorDeItem.COUNT % 5 == 0)
                                                 return true;
                                 else
                                                 return false;
                  }
  }

Neste código, método isReadyToCheckpoint é invocado em tempo de execução assim que cada item é lido. No corpo do método isReadyToCheckpoint será determinado se é hora de verificar o chunk atual. O método retorna true se o chunk precisa ser verificado ou false caso contrário.

Gerenciando Exceções

Por padrão quando qualquer batch lança uma exceção, a execução do job finaliza com um status de "FAILED". Podemos sobrescrever o comportamento padrão para o leitor, processador e escritor configurando exceções para ignorar ou tentar a operação novamente. Segue na Listagem 16 um exemplo.

Listagem 16. Definindo exceções para um job.

  <chunk item-count="3" skip-limit="3">
                  <reader .../>
                  <processor .../>
                  <writer .../>
   
                  <skippable-exception-classes>
                                 <include class="java.lang.Exception"/>
                                 <exclude class="java.io.IOException"/>
                  </skippable-exception-classes>
   
                  <retryable-exception-classes>
                                 <include class="java.lang.Exception"/>
                  </retryable-exception-classes>
  </chunk>

Neste fragmento de código skip-limit especifica o número de exceções que este step irá ignorar. skippable-exception-class especifica um conjunto de exceções que o processamento de um chunk irá ignorar. retryable-exception-class especifica um conjunto de exceções que o processamento do chunk tentará novamente realizar a operação. include especifica a classe de uma exceção ou uma superclasse de exceção para ignorar ou tentar novamente a operação. Múltiplos elementos include podem ser especificados. exclude especifica a classe de uma exceção ou uma superclasse de exceção para não ignorar ou não tentar novamente a operação. Múltiplos elementos exclude também podem ser especificados. Essas classes especificadas reduzem o número de exceções elegíveis para serem ignoradas ou tentadas novamente.

Neste fragmento de código acima temos que todas as exceções serão ignoradas, exceto java.io.IOException.

As interfaces SkipReadListener, SkipProcessListener, e SkipWriteListener podem ser implementadas para receber o controle quando uma exceção que foi marcada para ser ignorada é lançada.

As interfaces RetryReadListener, RetryProcessListener, e RetryWriteListener podem ser implementadas para receber o controle quando uma exceção passível de novas tentativas for lançada.

Processamento Batchlet

O estilo batchlet implementa um padrão batch chamado roll-your-own. Este é um estilo de processamento orientado a tarefa onde uma tarefa é invocada uma vez, executa até completar, e retorna um status de saída.

A interface Batchlet é usada para implementar um step. A classe abstrata AbstractBatchlet já fornece implementações dos métodos menos comumente implementados.

O Job XML é usado para definir um step com um batchlet conforme podemos verificar na Listagem 17.

Listagem 17. Definindo um step usando batchlet.

  <job id="meuJob" xmlns="http://xmlns.jcp.org/xml/ns/javaee" version="1.0">
                  <step id="meuStep" >
                                 <batchlet ref="meuBatchlet"/>
                  </step>
  </job>

Neste código o elemento job identifica um job. Este elemento tem um nome lógico "id" que é usado para identificação. Um job pode conter qualquer número de etapas identificado por um elemento "step". Este também tem um "id" usado para identificação. O elemento batchlet define um step do tipo batchlet. Este é mutualmente exclusivo com o elemento chunk. O atributo "ref" é identificado como o nome de um bean CDI de uma classe que implementa a interface Batchlet ou estende AbstractBatchlet. Segue na Listagem 18 um exemplo.

Listagem 18. Definindo a classe especificada no atributo ref do batchlet.

  @Named
  public class MeuBatchlet extends AbstractBatchlet {
                  @Override
                  public String process() {
                                 //. . .
                                 return "COMPLETED";
                  }
  }

Neste código MeuBatchlet é a implementação do batchlet definido anteriormente no Job XML. O método process é chamado para executar o trabalho do batchlet. Podemos notar neste caso um status explicito de COMPLETE que é retornado como um status do job. Se este método lançar uma exceção, o batchlet termina com um status de FAILED. A anotação @Named assegura que este bean pode ser referenciado em um Job XML.

Listeners

Os Listeners podem ser utilizados para interceptar uma execução de um batch. Listeners podem ser especificados em um job, step, chunk, leitor/escritor/processador para um chunk, skipping, e nas novas tentativas após exceções.

Segue na Tabela 1 uma lista de interfaces e classes abstratas que podem ser implementadas para interceptar a execução de um batch.

Tabela 1. Interfaces e classes abstratas para interceptar a execução de um batch.

Os Listeners podem ser especificados no Job XML conforme a Listagem 19.

Listagem 19. Definindo Listeners no Job XML.

  <job id="meuJob" xmlns="http://xmlns.jcp.org/xml/ns/javaee" version="1.0">
                  <listeners>
                                 <listener ref="meuJobListener"/>
                  </listeners>
   
                  <step id="meuStep" >
                                 <listeners>
                                                 <listener ref="meuStepListener"/>
                                                 <listener ref="meuChunkListener"/>
                                                 <listener ref="meuItemReadListener"/>
                                                 <listener ref="myeutemProcessorListener"/>
                                                 <listener ref="meuItemWriteListener"/>
                                 </listeners>
   
                                 <chunk>
                                                 . . .
                                 </chunk>
                  </step>
  </job>

Os listeners são especificados como um filho de . Todos os outros listeners são especificados como um filho de . O valor do atributo ref é o nome do bean CDI de uma classe que implementa o listener correspondente.

Sequenciamento no Job

Um step é um elemento básico de execução que encapsula uma fase independente e sequencial de um job.

Um job pode conter qualquer número de steps. Cada um desses steps pode ser um step do tipo chunk ou um step do tipo batchlet.

O próximo step na sequência de execução do job precisa ser explicitamente especificado através do atributo "next" conforme podemos verificar na Listagem 20.

Listagem 20. Definindo a sequência de steps no Job XML.

  <job id="meuJob" xmlns="http://xmlns.jcp.org/xml/ns/javaee" version="1.0">
                  <step id="step1" next="step2">
                                 <chunk item-count="3">
                                                 <reader ref="meuLeitorDeItem"></reader>
                                                 <processor ref="meuProcessadorDeItem"></processor>
                                                 <writer ref="meuEscritorDeItem"></writer>
                                 </chunk>
                  </step>
   
                  <step id="step2" >
                                 <batchlet ref="meuBatchlet"/>
                  </step>
  </job>

Neste Job XML definimos um job usando dois steps com os nomes lógicos "step1" e "step2". step1 é definido como um step do tipo chunk e step2 é definido como um step do tipo batchlet. step1 é executado primeiro e então seguido por step2. A ordem dos steps é identificada pelo atributo next em step1. Dessa forma, step2 é o último passo no job. Além do step, a especificação descreve outros elementos de execução que definem a sequência de um job:

• Flow: Define uma sequência de elementos que executam juntos como uma unidade;
• Split: Define um conjunto de fluxos que executam concorrentemente;
• Decision: Fornece uma maneira personalizada de determinar o sequenciamento entre steps (passos), flows (fluxos) e splits (divisões).

O primeiro step, flow, ou split define o primeiro step, flow ou split a ser executado para um dado Job XML.

Nas próximas seções detalharemos mais cada um desses elementos.

Flow

Um elemento de execução flow define uma sequência de elementos que executam juntos como uma unidade. Quando o flow é finalizado, o fluxo inteiro transaciona para o próximo elemento de execução. Segue na Listagem 21 um exemplo de como podemos especificar um flow.

Listagem 21. Definindo um flow e os próximos elementos que serão executados no Job XML.

  <job id="meuJob" xmlns="http://xmlns.jcp.org/xml/ns/javaee" version="1.0">
                  <flow id="flow1" next="step3">
                                 <step id="step1" next="step2">
                                                 . . .
                                 </step>
   
                                 <step id="step2" >
                                                 . . .
                                 </step>
                  </flow>
                  
                  <step id="step3" >
                                 . . .
                  </step>
  </job>

Neste Job XML definimos um job usando um "flow" com o nome "flow1" e um step com o nome "step3". Também definimos "flow1" usando dois steps: "step1" e "step2". Dentro do "flow", "step1" é seguido por "step2". Um "flow" pode conter qualquer elemento de execução. O elemento de execução dentro de um flow pode transacionar apenas entre eles, eles não podem transacionar para elementos fora do fluxo. Por padrão, "flow" é o último elemento de execução no job. Podemos especificar o próximo elemento de execução usando o atributo "next". step3 é executado após todos os passos em "flow1" serem finalizados. O valor do atributo "next" pode ser um nome lógico (id) de um step, flow, split ou decision.

Split

Um elemento split define um conjunto de flows que executam concorrentemente. Segue na Listagem 22 um exemplo.

Listagem 22. Exemplificando o uso de split no Job XML.

  <job id="meuJob" xmlns="http://xmlns.jcp.org/xml/ns/javaee" version="1.0">
                  <split id="split1" next="step3">
                                 <flow id="flow1">
                                                 <step id="step1">
                                                                 . . .
                                                 </step>
                                 </flow>
   
                                 <flow id="flow2">
                                                 <step id="step2">
                                                                 . . .
                                                 </step>
                                 </flow>
   
                  </split>
                                                 
                  <step id="step3">
                                 . . .
                  </step>
  </job>

Neste Job XML definimos um job usando um "split" com o nome lógico "split1" e um "step" com o nome lógico "step3". Um "split" pode apenas conter elementos "flow". O "split" acima contém dois elementos "flow" com os nomes lógicos "flow1" e "flow2". "flow1" tem um step chamado "step1" e "flow2" tem um "step" chamado "step2". Cada "flow" executa em uma thread separada. Por padrão, "split" é o último elemento de execução no job. Podemos especificar o próximo elemento de execução utilizando o atributo "next". O "split" é finalizado após todos os "flows" completarem. Quando o "split" inteiro é finalizado, executa-se o próximo elemento de execução. "step3" é executado após todos "steps" em "split" serem finalizados. O valor do próximo atributo pode ser um nome lógico de um step, flow, split ou decision.

Decision

Um elemento decision fornece uma forma customizada de determinarmos o sequenciamento entre steps, flows e splits.

Quatro elementos de transição são definidos para sequenciar a execução de um job ou terminar a execução de um job. São eles:

• next: Direciona a execução do flow para o próximo elemento de execução.
• fail: Causa a finalização de um job com um status FAILED.
• end: Causa o fim de um job com um status COMPLETED.
• stop: Causa o fim do job com um status STOPPED.

O decision usa qualquer elemento de transição para selecionar a próxima transição. Segue na Listagem 23 um exemplo.

Listagem 23. Exemplificando um decision no Job XML.

  <job id="meuJob" xmlns="http://xmlns.jcp.org/xml/ns/javaee" version="1.0">
                  <step id="step1" next="decider1">
                                 . . .
                  </step>
   
                  <decision id="decider1" ref="meuDecider">
                                 <next on="DATA_LOADED" to="step2"/>
                                 <end on="NOT_LOADED"/>
                  </decision>
   
                  <step id="step2">
                                 . . .
                  </step>
  </job>

Neste Job XML definimos um job usando um step com o nome lógico step1, um elemento decision com o nome lógico decider1, e outro step com o nome lógico step2. Um elemento decision é o destino do próximo elemento de um step, flow, split, ou outro decision. Neste caso, decider1 é especificado como o valor do próximo atributo de step1. O elemento decision segue para um step, flow, ou split.

O elemento decision tem uma referência para um batch Decider. Um Decider recebe o controle como parte de um elemento decision em um job e decide a próxima transição. O método decide recebe um array de objetos StepExecution como entrada. Esses objetos representam o elemento de execução que fazem a transição para a execução deste decisor. O método decide retorna um status de saída que atualiza a execução do job atual. Segue na Listagem 24 um exemplo de implementação de um Decider conforme referenciamos no Job XML anterior.

Listagem 24. Implementando a um Decider referenciado no Job XML.

  public class MeuDecider implements Decider {
                  @Override
                  public String decide(StepExecution[] ses) throws Exception {
                                 //. . .
   
                                 if (...)
                                                 return "NOT_LOADED";
   
                                 if (...)
                                                 return "DATA_LOADED";
                  }
  }

Este método retorna NOT_LOADED ou DATA_LOADED no status de saída. O elemento decision usa o próximo elemento de transição para transferir o controle para o step2 se o status de saída é DATA_LOADED. O job é finalizado através do elemento de transição “end” se o status de saída é NOT_LOADED.

Vale ressaltar que Fail, end, e stop são elementos de terminação, pois eles causam a finalização de um job.

Particionando o Job

Um step pode ser executado como um step particionado. Um step particionado executa como múltiplas instâncias do mesmo step através de múltiplas threads, sendo uma partição por thread. Cada partição pode ter parâmetros únicos que especificam em quais dados ele deve operar. Isso permite que um step seja particionado e executado em múltiplas threads, sem qualquer alteração no código Java existente.

O número de partições e o número de threads são controlados através de uma especificação estática no Job XML. Segue um exemplo na Listagem 25 de como podemos especificá-los.

Listagem 25. Especificando o número de partições e threads no Job XML.

  <step id="meuStep" >
                  <chunk item-count="3">
                                 <reader ref="meuLeitorDeItem">
                                                 <properties>
                                                                 <property name="start" value="#{partitionPlan['start']}" />
                                                                 <property name="end" value="#{partitionPlan['end']}" />
                                                 </properties>
                                 </reader>
                                 <processor ref="meuProcessadorDeItem"></processor>
                                 <writer ref="meuEscritorDeItem"></writer>
                  </chunk>
   
                  <partition>
                                 <plan partitions="4" threads="2">
                                                 <properties partition="0">
                                                                 <property name="start" value="1"/>
                                                                 <property name="end" value="10"/>
                                                 </properties>
                                 
                                                 <properties partition="1">
                                                                 <property name="start" value="11"/>
                                                                 <property name="end" value="20"/>
                                                 </properties>
                                 </plan>
                  </partition>
  </step>

Neste código, é um elemento opcional que é usado para especificar que um é um step particionado. O "partition plan" é especificado para um step do tipo chunk, mas pode ser especificado para um step do tipo batchlet também.

Cada tem um plan que especifica o número de partições através do atributo partitions, o número de partições para executarem concorrentemente através do atributo threads, e as propriedades para cada partição através do elemento . O atributo partition de cada uma das propriedades especifica a partição que a propriedade define.

Valores de propriedades são passadas para cada partição através do elemento property. Se essas propriedades precisam ser acessadas no leitor de item, então eles são especificados com #{partitionPlan[' onde PROPERTY-NAME é o nome da propriedade.

  @Inject
  @BatchProperty(name = "start")
  private String startProp;
   
  @Inject
  @BatchProperty(name = "end")
  private String endProp;

  <partition>
                  <mapper ref="meuMapper"/>
  </partition>

  public class MeuMapper implements PartitionMapper {
                  @Override
                  public PartitionPlan mapPartitions() throws Exception {
                                 return new PartitionPlanImpl() {
   
                                                 @Override
                                                 public int getPartitions() {
                                                                 return 2;
                                                 }
   
                                                 @Override
                                                 public int getThreads() {
                                                                 return 2;
                                                 }
   
                                                 @Override
                                                 public Properties[] getPartitionProperties() {
                                                                 Properties[] props = new Properties[getPartitions()];
   
                                                                 for (int i=0; i<getPartitions(); i++) {
                                                                                 props[i] = new Properties();
                                                                                props[i].setProperty("start", String.valueOf(i*10+1));
                                                                                props[i].setProperty("end", String.valueOf((i+1)*10));
                                                                 }
   
                                                                 return props;
                                                 }
                                 };
                  }
  }