How to pass transactions in multithreading (如何在多线程中进行事务传递)

[Sparkel-wsl]

I encountered such a situation, I need to insert a batch of data into multiple tables in the database, this process can be a very time-consuming operation, so I think to use multithreading to improve efficiency, but also to ensure transactional, if a thread insertion error, other threads need to roll back, only all threads do not make mistakes before committing.
But I ran into transaction issues while coding, and I didn't know how transactions were shared between multiple threads in quarkus and hiberna-orm frameworks. I looked at the two guidelines for https://quarkus.io/version/3.2/guides/context-propagation and https://quarkus.io/version/3.2/guides/transaction
I found that transaction passing can be done in the Context Propagation extension, but I am not familiar with Mutiny and have tried to do it the Mutiny way, but not the way I expected (each thread is responsible for a batch of data). Is not in the https://quarkus.io/version/3.2/guides/transaction guidelines
Mentioned how the transactions, but also try QuarkusTransaction.com MIT ()/the begin ()/rollback (); TransactionManager tm; Are independent affairs.
So I used CountDownLatch to control the commit or rollback of these individual transactions.
I provide a simple piece of code to illustrate my situation.

    public void concurrency(){
        //start async update
        CompletableFuture.runAsync(this::asyncUpdate).handle((ret,throwable) -> {
            log.error(throwable.getMessage());
            return null;
        });
        System.out.println("response finished");
    }

    public void asyncUpdate(){
        AtomicBoolean hasException = new AtomicBoolean(false);
        //Control transaction commit or rollback
        CountDownLatch countDownLatch = new CountDownLatch(2);
        //task one
        CompletableFuture<Void> asyncUser = managedExecutor.runAsync(
                        () -> QuarkusTransaction.requiringNew()
                                .exceptionHandler(throwable -> {
                                    log.error("transaction A rollback:{}", throwable.getMessage());
                                    return TransactionExceptionResult.ROLLBACK;
                                })
                                .run(() -> {
                                    try {
                                        try {
                                            // Simulate a very time-consuming insert operation
                                            userDao.update("set name = 'test' where id = 1");
                                        } catch (Exception e) {
                                            countDownLatch.countDown();
                                            throw new RuntimeException(e);
                                        }
                                        countDownLatch.countDown();
                                        countDownLatch.await();
                                        if (hasException.get()) {
                                            log.error("task is have a exception");
                                            throw new RuntimeException("rollback");
                                        }
                                    } catch (Exception e) {
                                        hasException.set(true);
                                        throw new RuntimeException(e.getMessage());
                                    }
                                }))
                .exceptionally(throwable -> {
                    log.error("asyncUser exception:{}", throwable.getMessage());
                    return null;
                });
        //task two
        CompletableFuture<Void> asyncOrder = managedExecutor.runAsync(
                        () -> QuarkusTransaction.requiringNew()
                                .exceptionHandler(throwable -> {
                                    log.error("transaction B rollback:{}", throwable.getMessage());
                                    return TransactionExceptionResult.ROLLBACK;
                                })
                                .run(() -> {
                                    try {
                                        try {
                                            //Simulate a very time-consuming insert operation
                                            orderDao.update("set userName = 'test' where id1 = 11");
                                        } catch (Exception e) {
                                            countDownLatch.countDown();
                                            throw new RuntimeException(e);
                                        }
                                        countDownLatch.countDown();
                                        countDownLatch.await();
                                        if (hasException.get()) {
                                            log.error("task is have a exception");
                                            throw new RuntimeException("rollback");
                                        }
                                    } catch (Exception e) {
                                        hasException.set(true);
                                        throw new RuntimeException(e.getMessage());
                                    }
                                }))
                .exceptionally(throwable -> {
                    log.error("asyncOrder exception:{}", throwable.getMessage());
                    return null;
                });
        //await task one finished and task two finished
        CompletableFuture.allOf(asyncOrder, asyncUser).join();
        log.info("--------------------------");

    }

Is there a better way to implement multiple threads sharing a transaction?

我碰到这样一个情况，我需要将一批数据插入到数据库的多个表中，这个过程可能是非常耗时的操作，于是我觉得采用多线程来提升效率，但同时也要保证事务性，如果某个线程插入时出错了，其它线程需要回滚，只有全部线程都不出错的情况下才提交。
但是在编码的时候碰到了事务上的问题，我不知道在quarkus与hiberna-orm框架中事务如何在多线程之间传递共享，我查阅了这两个指南https://quarkus.io/version/3.2/guides/context-propagation 和https://quarkus.io/version/3.2/guides/transaction ，
发现Context Propagation extension 中可以进行事务传递，但是我对Mutiny使用不熟练，并且也尝试过使用Mutiny方式来进行事务传递，但并不是我期望的那样(每个线程负责一批数据)。在https://quarkus.io/version/3.2/guides/transaction指南中也并未
提到事务如何传递，但也尝试了QuarkusTransaction.commit()/begin()/rollback();TransactionManager tm;均是独立的事务。
于是我采用CountDownLatch来控制这些独立事务的提交或回滚。
我提供一段简易的代码来说明我的情况。

请问有没有更好的解决办法来实现多个线程共享一个事务。

[Sparkel-wsl]

I've found a solution by adding the @transactional annotation to the asyncUpdate () method. The rest of the threads QuarkusTransaction. JoiningExisting () transfer transaction.
Modified code
public void concurrency(){
//start async update
completableFuture = CompletableFuture.runAsync(this::asyncUpdate).handle((ret,throwable) -> {
log.error(throwable.getMessage());
return null;
});
System.out.println("response finished");
}

@Transactional(rollbackOn = Exception.class)
public void asyncUpdate()   {

    //task one
    CompletableFuture<Void> asyncUser = managedExecutor.runAsync(
                    () -> QuarkusTransaction.joiningExisting()
                           
                            .run(() -> {
                                // Simulate a very time-consuming insert operation
                                userDao.update("set name = 'test' where id = 1");
                              
                            }));
    //task two
    CompletableFuture<Void> asyncOrder = managedExecutor.runAsync(
                    () -> QuarkusTransaction.joiningExisting()
                            .run(() -> {
                                //Simulate a very time-consuming insert operation
                                orderDao.update("set userName = 'test' where id1 = 11");
                            }))
            ;
    //await task one finished and task two finished
    CompletableFuture.allOf(asyncOrder, asyncUser).join();


    log.info("--------------------------");

}

[maxandersen]

if i understand right you are trying to run with multiple concurrent threads in one transaction ?

That is not going to be of much help for you as everything would need to be serialized on the jdbc connection anyways.

If your operations really are that slow to do in the database then no amount of multithreading is going to improve the situation.

If the operations is more a in memory calculation taking time then look at do that work concurrently and then perform the database updates at the end - possibly in a way that gives you bulk/batch improvements.

See https://docs.jboss.org/hibernate/orm/6.4/userguide/html_single/Hibernate_User_Guide.html#batch and especially the new refreshed introductury guide covers best practices on using database efficiently with Hibernate in https://docs.jboss.org/hibernate/orm/6.4/introduction/html_single/Hibernate_Introduction.html