Java: распространение транзакций XA во многих потоках - PullRequest
Новая
       62

Java: распространение транзакций XA во многих потоках

4 голосов
/ 12 марта 2011

Как использовать диспетчер транзакций (например, Bitronix , JBoss TS или Atomikos ) в Java SE (не Java EE или Spring) дляподдерживаем следующий вариант использования:

Предположим, у нас есть следующий класс: 

public class Dao {

    public void updateDatabase(DB db) {
        connet to db
        run a sql
    }

}

, и мы создаем из него Java Runnable, как показано ниже: 

public class MyRunnable extends Runnable {

    Dao dao;
    DB db;

    public MyRunnable(Dao dao, DB db) {
        this.dao=dao;
        this.db = db;
    }           

    public run() throws Exception {
        return dao.updateDatabase(db);
    }
}

Теперь в нашем слое Service у нас есть другой класс: 

public class Service {

    public void updateDatabases() {

        BEGIN TRANSACTION;

        ExecutorService es = Executors.newFixedThreadPool(10);

        ExecutorCompletionService ecs = new ExecutorCompletionService(es);

        List<Future<T>> futures = new ArrayList<Future<T>>(n);

        Dao dao = new Dao();

        futures.add(ecs.submit(new MyRunnable(dao, new DB("db1")));
        futures.add(ecs.submit(new MyRunnable(dao, new DB("db2")));
        futures.add(ecs.submit(new MyRunnable(dao, new DB("db3")));

        for (int i = 0; i < n; ++i) {
            completionService.take().get();
        }

       END TRANSACTION;
    }

}

И клиент может быть сервлетом или любой другой многопоточной средой: 

public MyServlet extend HttpServlet {

    protected void service(final HttpServletRequest request, final HttpServletResponse response) throws IOException {

        Service service = new Service();

        service.updateDatabases();

    }

}

Что будет правильнымкод для частей НАЧАЛО СДЕЛКИ и КОНЕЦ СДЕЛКИ?Это вообще возможно?Если нет, что нужно изменить?Требуется, чтобы метод updateDatabases () был параллельным (поскольку он будет одновременно обращаться к нескольким базам данных) и транзакционным.

Ответы [ 3 ]

3 голосов
/ 15 марта 2011

Похоже, что это можно сделать, используя Atomikos , используя SubTxThread 

//first start a tx
TransactionManager tm = ...
tm.begin();

Waiter waiter = new Waiter();

//the code that calls the first EIS; defined by you
SubTxCode code1 = ...

//the associated thread
SubTxThread thread1 = new SubTxThread ( waiter , code1 );

//the code that calls the second EIS; defined by you
SubTxCode code2 = ...

//the associated thread
SubTxThread thread2 = new SubTxThread ( waiter , code2 );

//start each thread
thread1.start();

thread2.start();

//wait for completion of all calls
waiter.waitForAll();

//check result
if ( waiter.getAbortCount() == 0 ) {
    //no failures -> commit tx
    tm.commit();
} else {
    tm.rollback();
}
1 голос
/ 13 марта 2011

Спецификация XA требует, чтобы все вызовы XA выполнялись в одном и том же контексте потока.Чтобы объяснить причину этого, это потому, что коммит может быть вызван до того, как любая из ветвей транзакций будет даже создана в ваших потоках.

, если вы просто заинтересованы в том, как выполнить эти три вызова в транзакции XA вJBoss TS

Сначала убедитесь, что ваш -ds.xml указывает ваш источник данных как <xa-datasource>

InitialContext ctx = new InitialContext(parms);
UserTransaction ut = (UserTransaction) ctx.lookup("java:comp/UserTransaction");

ut.begin();

//Some Transactional Code

ut.commit();

. Имейте в виду, что с кодом выше вы не сможете использовать ExecutorService для распараллеливаниязвонки.

Примечание: я не знаю много об этом, но JTS / OTS утверждает, что позволяет нескольким потокам участвовать в транзакции.Я думаю, что это происходит путем распространения транзакционного контекста, аналогичного ws-координированию / ws-транзакции, и поддерживается JBossTS.Это может быть красная сельдь, но если у вас нет времени, возможно, стоит исследовать.

0 голосов
/ 30 января 2014

Как насчет вас

  1. BEGIN_TRANSATION: Подключитесь ко всем 3 базам данных в вашей службе,
  2. передайте объекты Connection (вместо объекта db) в MyRunnable
  3. END_TRANSACTION: вызвать фиксацию и закрыть все 3 соединения в вашей службе
...