Spring无法将事务传播到ForkJoin的RecursiveAction

Question

我正在尝试实现一个多线程解决方案，以便并行化我的业务逻辑，包括读写数据库。

技术栈：Spring 4.0.2，Hibernate 4.3.8

这里有一些要讨论的代码：

配置

@Configuration
public class PartitionersConfig {

    @Bean
    public ForkJoinPoolFactoryBean forkJoinPoolFactoryBean() {
        final ForkJoinPoolFactoryBean poolFactory = new ForkJoinPoolFactoryBean();
        return poolFactory;
    }
}

服务

@Service
@Transactional
public class MyService {

    @Autowired
    private OtherService otherService;

    @Autowired
    private ForkJoinPool forkJoinPool;

    @Autowired
    private MyDao myDao;

    public void performPartitionedActionOnIds() {
        final ArrayList<UUID> ids = otherService.getIds();

        MyIdPartitioner task = new MyIdsPartitioner(ids, myDao, 0, ids.size() - 1);
        forkJoinPool.invoke(task);
    }
}

仓库 / 数据访问对象

@Repository
@Transactional(propagation = Propagation.MANDATORY)
public class IdsDao {

    public MyData getData(List<UUID> list) {
        // ... 
    }
}

递归操作

public class MyIdsPartitioner extends RecursiveAction {

    private static final long serialVersionUID = 1L;
    private static final int THRESHOLD = 100;

    private ArrayList<UUID> ids;
    private int fromIndex;
    private int toIndex;

    private MyDao myDao;

    public MyIdsPartitioner(ArrayList<UUID> ids, MyDao myDao, int fromIndex, int toIndex) {
        this.ids = ids;
        this.fromIndex = fromIndex;
        this.toIndex = toIndex;
        this.myDao = myDao;
    }

    @Override
    protected void compute() {
        if (computationSetIsSamllEnough()) {
            computeDirectly();
        } else {
            int leftToIndex = fromIndex + (toIndex - fromIndex) / 2;
            MyIdsPartitioner leftPartitioner = new MyIdsPartitioner(ids, myDao, fromIndex, leftToIndex);
            MyIdsPartitioner rightPartitioner = new MyIdsPartitioner(ids, myDao, leftToIndex + 1, toIndex);

            invokeAll(leftPartitioner, rightPartitioner);
        }
    }

    private boolean computationSetIsSamllEnough() {
        return (toIndex - fromIndex) < THRESHOLD;
    }

    private void computeDirectly() {
        final List<UUID> subList = ids.subList(fromIndex, toIndex);
        final MyData myData = myDao.getData(sublist);
        modifyTheData(myData);
    }

    private void modifyTheData(MyData myData) {
        // ...
        // write to DB
    }
}

执行后我得到了以下结果：

没有找到与传播“mandatory”的标记标记的事务对应的现有事务

我理解这是很正常的，因为事务不会在不同的线程中传播。因此，一种解决方案是在每个线程中手动创建事务如另一个类似问题所建议的那样。但是这对我来说还不够令人满意，所以我继续搜索。

在Spring论坛中，我发现了关于该主题的讨论。其中一个段落我认为非常有趣：

"我可以想象人们可以手动将事务上下文传播到另一个线程，但我认为你真的不应该尝试。事务绑定到单个线程是有原因的 - 基本的底层资源 - jdbc连接 - 不是线程安全的。在多个线程中使用一个连接将破坏基本的jdbc请求/响应协议，并且如果它能在更复杂的示例中工作，那将是一个小奇迹。"

所以第一个问题是：并行读写数据库值得吗？这样做会真正损害数据库的一致性吗？
如果上面的引用不是真的，我怀疑，是否有办法实现以下内容：MyIdPartitioner由Spring管理 - 使用@Scope("prototype") - 并将递归调用所需的参数传递给它，这样就可以将事务管理留给Spring？

- nyxz

Answer 1

经过进一步阅读，我成功解决了我的问题。就我现在的看法来说，一开始并没有问题。

由于我从数据库中读取的数据是分块的，并且我确信在此期间结果不会被编辑，所以我可以在事务之外进行操作。

在我的情况下，写入也是安全的，因为我写入的所有值都是唯一的，不会发生任何约束冲突。因此，我也将事务从那里移除了。

我所说的“移除事务”是在我的DAO中覆盖方法的传播模式，如下所示：

@Repository
@Transactional(propagation = Propagation.MANDATORY)
public class IdsDao {

    @Transactional(propagation = Propagation.SUPPORTS)
    public MyData getData(List<UUID> list) {
        // ... 
    }
}

如果您决定出于某些原因需要该事务，则仍然可以通过将传播设置为REQUIRED来将事务管理交给Spring。

因此，解决方案比我想象的要简单得多。

回答我的其他问题：

“并行读/写数据库是否值得，并且这样做真的会影响数据库一致性吗？”

是的，这很值得。只要每个线程有一个事务，就不用担心。

“有没有办法实现以下内容：MyIdPartitioner由Spring管理-使用@Scope（“prototype”）-并将所需的参数传递给它的递归调用，从而将事务管理留给Spring？”

是有方法的using pool(另一个stackoverflow问题)。或者您可以将bean定义为@Scope(value = "prototype", proxyMode = ScopedProxyMode.TARGET_CLASS)，但如果需要对其设置参数，则无法工作，因为每个实例的每次使用都会给您一个新的实例。例如：

@Autowire
MyIdsPartitioner partitioner;

public void someMethod() {
    ...
    partitioner.setIds(someIds);
    partitioner.setFromIndex(fromIndex);
    partitioner.setToIndex(toIndex);
    ...
}

这将创建3个实例，但由于字段没有设置，您将无法受益于该对象。

简而言之，有一种方法，但我起初并不需要使用它。

- nyxz

Answer 2

这应该可以通过使用Atomikos（http://www.atomikos.com）实现，如果需要还可以使用嵌套事务。

如果您这样做，请注意避免死锁，如果多个同一根事务的线程写入数据库中的相同表格。

- Guy Pardon

---

好的，这似乎是值得关注的事情，但我希望保持我的技术栈不变。此外，这并不完全回答了我的两个问题。 - nyxz