各位大佬,帮忙看下Java中fork/join是否正确。多线程查询分页数据

黄贤达 发布于 09/18 10:49
阅读 649
收藏 0

各位大佬,我想查询通过分页的方式查询数据库里前10页的数据,每页100条。所以想通过fork/join去异步查询,再汇总数据。请大佬帮我看看下面的写法是否有问题,我在debug的过程中发现,有些页面重复查询了,这是为什么呢?谢谢

 

public class ExportTask extends RecursiveTask<List<RechargeSummaryDTO>> {

    private static final int THRESHOLD = 1;

    //起始页
    private int startPage;
    //结束页
    private int endPage;
    private RechargeFacade rechargeFacade;
    //每页大小
    private int pageSize;
    查询条件
    private RechargeSummaryQuery query;

    public ExportTask(RechargeFacade rechargeFacade, int pageSize, int startPage, int endPage, RechargeSummaryQuery query) {
        this.rechargeFacade = rechargeFacade;
        this.startPage = startPage;
        this.endPage = endPage;
        this.pageSize = pageSize;
        this.query = query;
    }


    @Override
    protected List<RechargeSummaryDTO> compute() {
        if ((endPage - startPage) <= THRESHOLD) {
            Pageable pageable = new Pageable();
            pageable.setPageNum(endPage);
            pageable.setPageSize(pageSize);
            PageResult<RechargeSummaryDTO> rechargeSummaryDTOPageResult = rechargeFacade.rechargeSummary(pageable, query);
            return rechargeSummaryDTOPageResult.getRecords();
        } else {
            int middle = (startPage + endPage) / 2;
            ExportTask left = new ExportTask(rechargeFacade, pageSize, startPage, middle, query);
            ExportTask right = new ExportTask(rechargeFacade, pageSize, middle, endPage, query);
            left.fork();
            right.fork();
            List<RechargeSummaryDTO> leftResult = left.join();
            List<RechargeSummaryDTO> rightResult = right.join();
            leftResult.addAll(rightResult);
            return leftResult;
        }
    }
加载中
0
jzdayz
jzdayz

其实有个最简单的写法 例如查询50到100页数据

List<Object> r = IntStream.range(50, 100).parallel().mapToObj((i) -> {
    // 查询数据
    log.info("MAP");
    List<Object> rs = new ArrayList<>();
    rs.add(new Object());
    return rs;
}).collect(Collectors.toList()).stream().reduce((a, b) -> {
    log.info("REDUCE");
    a.addAll(b);
    return a;
}).orElse(Collections.emptyList());
log.info("count:{}",r.size());

stream的并行也是用的forkjointask,去做的

中间要转list,是因为上面是并行流,重新弄成了串行流,否则addAll会有问题,或者使用线程安全的list聚合

黄贤达
OK。非常感谢,这方法甚好。:+1:
0
jzdayz
jzdayz

首先有两个问题:

1.查询1到3页的数据,在你的代码中只查询了页码2和页码3

2.forkjoin不要这么写,这么写你原本可以继续执行分发的任务又被你分发出去了,就相当于一个线程只是判断了一次,浪费资源,参考javadoc的写法,只fork一个任务出去,另外一个任务由本线程继续执行

```java

public static class Fibonacci extends RecursiveTask<Integer> {
    final int n;

    public Fibonacci(int n) {
        this.n = n;
    }

    public Integer compute() {
        if (n <= 1)
            return n;
        Fibonacci f1 = new Fibonacci(n - 1);
        f1.fork();
        Fibonacci f2 = new Fibonacci(n - 2);
        return f2.compute() + f1.join();
    }
}

```

黄贤达
谢谢
0
jzdayz
jzdayz

你的代码简化一下运行1到100,这应该是不会重复的,只是少了一个1

public static void main(String[] args) {
    test1();
}

private static void test1() {

    new A(1,100).compute();

    CONTAINER.stream().sorted().forEach(k-> System.out.println(k.toString()));
}

public static List<Integer> CONTAINER = new CopyOnWriteArrayList<>();

@AllArgsConstructor
public static class A extends RecursiveTask<Integer> {
    private int start;
    private int end;
    final static int THRESHOLD = 1;


    public Integer compute() {
        if ((end - start) <= THRESHOLD) {
            System.out.println("查询数据页码:"+end);
            CONTAINER.add(end);
            return null;
        } else {
            int middle = (start + end) / 2;
            A left = new A(start, middle);
            A right = new A(middle, end);
            left.fork();
            right.fork();
            left.join();
            right.join();
            return null;
        }

    }
}

2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100

 

0
innerloop
innerloop


我觉得 你写个用例试一下,应该把这块处理下 就行 else 不用addAll 也不要返回结果,这个else 就是用来多个fork的,具体还是靠<=1的返回

jzdayz
jzdayz
你这个我都差点怀疑了,如果这里不聚合,肯定是有问题 假设我们只查询2页数据 leftResult返回第一页,rightResult返回第二页,不聚合肯定是有问题,上面的if只是返回最小单元的数据,所有页的数据聚合才是最终结果 参考 javadoc上面的示例,很明显
黄贤达
好的 谢谢
返回顶部
顶部