8

Uma solução Java SE, utilizando o framework Spring, está sendo desenvolvida com o intuito de paralelizar a execução de uma Stored Procedure no Oracle 9i, Procedure esta que recebe como parâmetro uma ou mais linhas de um arquivo gigante a ser processado.

Isso será feito por intermédio de um pool de Threads, onde cada Thread invocará essa mesma Stored Procedure passando como parâmetro linhas diferentes deste arquivo, a fim de otimizar o tempo de processamento em comparação à somente uma execução desta Procedure. Entretanto, é necessário garantir atomicidade transacional sobre todas as execuções/transações desta mesma Procedure, como é garantido hoje em uma única execução desta Procedure — ou seja, commitar as transações no final e somente caso nenhuma outra acuse erro: e caso haja erro em alguma execução de uma Thread, será necessário efetuar rollback em todas as transações das demais.

Foi logicamente cogitado utilizar uma implementação JTA (XA) no Spring, como aqui, onde cada Thread/transação seria considerada um recurso participante do two-phase commit, porém acredito que o design desta nossa solução fere o princípio do mecanismo JTA, pois a princípio ele somente garante atomicidade sobre os recursos utilizados em um mesmo método transacional, ou seja, em somente uma mesma Thread.

Como garantir isso de uma maneira não programática?

1 Resposta 1

5

Utilizando apenas o Atomikos, uma implementação do JTA e do XA, fiz um exemplo simples que permite executar processamento em várias threads dentro de uma transação.

O projeto completo está disponível no meu GitHub.

Implementação

Antes de mais nada, temos a inicialização do DataSource e do TransactionManager:

// Atomikos implementations
private static UserTransactionManager utm;
private static AtomikosDataSourceBean adsb;

// initialize resources
public static void init() {
    utm = new UserTransactionManager();
    try {
        utm.init();
        adsb = new AtomikosDataSourceBean();
        adsb.setMaxPoolSize(20);
        adsb.setUniqueResourceName("postgres");
        adsb.setXaDataSourceClassName("org.postgresql.xa.PGXADataSource");
        Properties p = new Properties();
        p.setProperty("user", "postgres");
        p.setProperty("password", "0");
        p.setProperty("serverName", "localhost");
        p.setProperty("portNumber", "5432");
        p.setProperty("databaseName", "postgres");
        adsb.setXaProperties(p);
    } catch (SystemException e) {
        e.printStackTrace();
        throw new RuntimeException(e);
    }
}

Depois, uma thread que recebe a instância da transação (Transaction) principal:

private static class Processamento implements Callable<Integer> {

    private int id;
    private boolean falhar;
    private Transaction transaction;

    public Processamento(int id, boolean falhar, Transaction transaction) {
        this.falhar = falhar;
        this.transaction = transaction;
        this.id = id;
    }

    public Integer call() throws Exception {
        if (falhar) {
            throw new RuntimeException("Falhou inesperadamente!");
        }

        //enlist xa connection
        XAConnection xac = AtomikosDataSource.getDS().getXaDataSource().getXAConnection();
        synchronized (transaction) {
            transaction.enlistResource(xac.getXAResource());
        }

        //normal execution, update row with OK
        Connection c = xac.getConnection();
        Statement s = c.createStatement();
        s.executeUpdate("update teste set processado = 'ok' where id = " + id);
        s.close();
        c.close();

        //delist xa connection
        synchronized (transaction) {
            transaction.delistResource(xac.getXAResource(), XAResource.TMSUCCESS);
        }
        return id;
    }

}

Note que, ao invés de usar o JTA, estou usando diretamente a API do XA implementada pelo Atomikos.

A chamada AtomikosDataSource.getDS().getXaDataSource().getXAConnection() recupera uma conexão do XA, a qual é adicionada à transação principal com o comando transaction.enlistResource(xac.getXAResource()).

Fiz a sincronização em alguns trechos, pois obtive aleatoriamente alguns NullPointerException nos testes, porém isso não deve ser um problema se você usar as conexões com prudência, isto é, sem ficar abrindo-as e fechando-as o tempo todo.

Finalmente, um método que inicia cinco instâncias da thread acima:

public static int processar(boolean falhar) {
    int ok = 0;
    Transaction transaction = null;
    try {

        //start transaction
        AtomikosDataSource.getTM().begin();
        transaction = AtomikosDataSource.getTM().getTransaction();

        //create thread pool
        ExecutorService executor = Executors.newFixedThreadPool(5);
        List<Callable<Integer>> processos = new ArrayList<Callable<Integer>>();

        //create 5 threads, passing the main transaction as argument
        for (int i = 0; i < 5; i++) {
            processos.add(new Processamento(i + 1, i == 4 && falhar, transaction));
        }

        //execute threads and wait
        List<Future<Integer>> futures = executor.invokeAll(processos);

        //count the result; get() will fail if thread threw an exception
        Throwable ex = null;
        for (Future<Integer> future : futures) {
            try {
                int threadId = future.get();
                System.out.println("Thread " + threadId + " sucesso!");
                ok++; 
            } catch (Throwable e) {
                ex = e;
            }
        }

        if (ex != null) {
            throw ex;
        }

        //finish transaction normally
        transaction.commit();

    } catch (Throwable e) {

        e.printStackTrace();
        try {
            //try to rollback
            if (transaction != null) {
                AtomikosDataSource.getTM().rollback();
            }
        } catch (IllegalStateException e1) {
            e1.printStackTrace();
        } catch (SecurityException e1) {
            e1.printStackTrace();
        } catch (SystemException e1) {
            e1.printStackTrace();
        }

    }
    return ok;
}

Fiz alguns testes tanto num cenário de sucesso quanto de falha para validar o resultado.

No cenário de sucesso, cada uma das cinco threads atualiza uma linha da tabela TESTE com o valor ok e faz o commit da transação.

No cenário de falha, a última thread sempre lança uma exceção, forçando o rollback das outras quatro threads.

Veja o código no GitHub para maiores detalhes.

Notas sobre a configuração

Usei o PostgreSQL no exemplo. Foi necessário habilitar a configuração max_prepared_transactions com um valor maior que zero no arquivo de configuração postgresql.conf.

É importante verificar se o seu driver de banco de dados suporta transações distribuídas. Li em algum lugar que o MySQL pode ter alguns problemas quanto a isso.

Observações

Para fazer o exemplo funcionar com Spring, basta configurar as classes criadas manualmente em beans no XML ou através de anotações. Fica a seu critério.

Cuidado se resolver implementar algo assim dentro de um Servidor de Aplicação, de modo a não interferir nas transações normais do sistema.

Pessoalmente, não vejo uma real necessidade em processamento paralelo dentro de uma mesma transação. É muito mais eficiente dividir o processamento em blocos transacionais. Há várias técnicas para se fazer isso sem tornar o estado do sistema inconsistente, por exemplo, usando colunas adicionais na tabela ou mesmo uma tabela extra.

8
  • Luiz, vc pode não enxergar razão alguma para isso, porém a necessidade, no meu caso e tipicamente em muitos outros, existe e é uma premissa para garantir a consistência dos dados, sem delegar essa responsabilidade a outros mecanismos programáticos de reprocessamento de blocos, muitas vezes definidos como compensating transactions. O protocolo XA, ao contrário do que vc afirmou, garante que uma transação global orquestre outras transações e garanta o estado consistente de uma base de dados. 17/04/2014 às 20:56
  • 1
    @AlbertoFragaláNeto Não afirmei que o XA não faz isso, apenas que não vejo muito sentido nesse contexto. Não sei até hoje de um BD com tamanho volume de dados onde foi necessário manter uma transação e ao mesmo tempo processar em paralelo. Sobre a questão de consistência, não vejo nenhum impedimento em garanti-la por bloco. Pelas experiências que tenho em trabalhar com volume, mantenho minha posição: em geral esse requisito de consistência "tudo ou nada" é desnecessário se o processamento for bem planjado. Algo impede o seu sistema de ter uma flag "processado" para cada bloco?
    – utluiz
    18/04/2014 às 0:50
  • @AlbertoFragaláNeto De qualquer forma, vou atualizar a resposta com um projeto de exemplo que fiz.
    – utluiz
    18/04/2014 às 0:50
  • Luiz, após vários testes, foi notado que não está ocorrendo o "enlistment" das conexões de cada Thread. As conexões estão sendo commitadas independentemente, no momento em que é invocado o método "executeUpdate". Isso pode ser observado se vc invocar apenas o primeiro teste "sucesso()", e apenas este, e realizar um rollback no TransactionManager ao invés de commit no final. Vc observará todas as linhas atualizadas para "ok". Eu garanti o parâmetro que vc citou, observando no log (INFO: USING com.atomikos.icatch.threaded_2pc = true). O que deve estar ocorrendo? 22/04/2014 às 22:07
  • Luiz, estive lendo a documentação desse parâmetro que vc me passou, e notei que na verdade ele não realiza isso que vc propôs. A documentação cita que na verdade o processo multi-thread é em relação ao processo de commit do two-phase-commit, para efeitos de performance quando existem muitos recursos envolvidos na transação: (atomikos.com/Documentation/JtaProperties) 23/04/2014 às 21:29

Você deve fazer log-in para responder a esta pergunta.

Esta não é a resposta que você está procurando? Pesquise outras perguntas com a tag .