Java 多线程 —— 内存合并

Java 多线程 —— 内存合并

​  最近在 B站 看到有关于利用 多线程 进行 业务任务的内存合并处理 的例子觉得很好，在此做一下记录。
 主要内容是：将多个用户请求合并成一个请求队列，利用 多线程 进行并行处理，达到批量处理请求、提高服务 dps 的目的。
 关键点：线程间通信。

Demo 视频链接：https://www.bilibili.com/video/BV1Hv4y1P7ta/?spm_id_from=333.788&vd_source=8b6dfb9416c7aad22744de7bcdae9a95

知识背景 —— Executor 框架

概述

​  Executor 框架 是 Java 5 引进的用于 启动、调度和管理线程的 API，其内部使用了线程池机制，同时也是一个 基于 生产者-消费者模式 的异步执行框架，其提交任务的线程相当于生产者、执行任务的线程相当于消费者（使用 Runnable 表示任务），处于 java.util.cocurrent 包下。

主要内容

• 线程池

• Executor

  ​  顶层接口。

• Executors

  ​  工厂类，用于创建线程池，其返回的线程池都实现了 ExecutorService 接口。

  • public static ExecutorService newFiexedThreadPool(int Threads)：创建固定数量线程的线程池
  • public static ExecutorService newCachedThreadPool()：创建一个可缓存的线程池
  • public static ExecutorService newSingleThreadExecutor()：创建一个单线程化的 Executor
  • public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)：创建一个支持定时及周期性任务执行的线程池

• ExecutorService

  ​  Executor 的子接口，其提供了生命周期管理的方法，返回 Future 对象 和 可跟踪一个或多个异步任务执行状况的方法。

• CompletionService

• Future

• Callable

C#内存释放-线程控制-线程启动-线程暂停

rar 1星 超过10%的资源 68KB

下载

扩展：

​  CountDownLatch 类：允许一个或者多个线程去等待其他线程完成操作，接收一个 int 类型的参数，表示要等待的工作线程的个数。

​ 应用场景：

• 某个线程需要在其他 n 个线程执行完毕后再向下执行
• 多个线程并行执行同一个任务，提高响应速度

Demo

package com.hong.demo.thread_merger;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

public class KillDemo {
    /**
     * 启动 10 个线程
     * 库存 6 个
     * 生成一个合并队列
     * 每个用户能拿到自己的请求响应
     */
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        KillDemo killDemo = new KillDemo();
        killDemo.mergeJob();
        Thread.sleep(2000);

        List<Future<Result>> futureList = new ArrayList<>();
        CountDownLatch countDownLatch = new CountDownLatch(10);
        for (int i = 0; i < 6; i++) {
            final Long orderId = i + 100L;
            final Long userId = Long.valueOf(i);
            Future<Result> future = executorService.submit(() -> {
                countDownLatch.countDown();
                countDownLatch.await(1000, TimeUnit.MILLISECONDS);
                return killDemo.operate(new UserRequest(orderId, userId, 1));
            });
            futureList.add(future);
        }

        futureList.forEach(future -> {
            try {
                Result result = future.get(300, TimeUnit.MILLISECONDS);
                System.out.println(Thread.currentThread().getName() + ":客户端请求响应:" + result);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

    // 模拟数据库行
    private Integer stack = 10;

    private BlockingQueue<RequestPromise> queue = new LinkedBlockingQueue<>(10);

    /**
     * 用户库存扣减
     *
     * @param userRequest
     */
    public Result operate(UserRequest userRequest) throws InterruptedException {
        // TODO 阈值判断
        // TODO 队列的创建
        RequestPromise requestPromise = new RequestPromise(userRequest);
        synchronized (requestPromise) {
            boolean enqueueSuccess = queue.offer(requestPromise, 100, TimeUnit.MILLISECONDS);
            if (!enqueueSuccess) {
                return new Result(false, "系统繁忙");
            }
            try {
                requestPromise.wait(200);
                if (requestPromise.getResult() == null) {
                    return new Result(false, "等待超时");
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return requestPromise.getResult();
    }

    /**
     * 开启一个异步线程，执行任务
     */
    public void mergeJob() {
        new Thread(() -> {
            List<RequestPromise> list = new ArrayList<>();
            while (true) {
                if (queue.isEmpty()) {
                    try {
                        Thread.sleep(10);
                        continue;
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }

                int batchSize = queue.size();
                for (int i = 0; i < batchSize; i++) {
                    list.add(queue.poll());
                }

                System.out.println(Thread.currentThread().getName() + ":合并扣减库存" + list);

                int sum = list.stream().mapToInt(e -> e.getUserRequest().getCount()).sum();
                // 两种情况
                // 1. 总量未超库存
                if (sum <= stack) {
                    stack -= sum;
                    // notify user
                    list.forEach(requestPromise -> {
                        requestPromise.setResult(new Result(true, "ok"));
                        synchronized (requestPromise) {
                            requestPromise.notify();
                        }
                    });
                    continue;
                }
                // 2. 总量超库存，退化成串行处理
                for (RequestPromise requestPromise : list) {
                    int count = requestPromise.getUserRequest().getCount();
                    if (count <= stack) {
                        stack -= count;
                        requestPromise.setResult(new Result(true, "ok"));
                    } else {
                        requestPromise.setResult(new Result(false, "库存不足"));
                    }
                    synchronized (requestPromise) {
                        requestPromise.notify();
                    }
                }
                list.clear();
            }
        }, "mergeThread").start();
    }
}
// 锁资源
class RequestPromise {
    private UserRequest userRequest;
    private Result result;

    public RequestPromise(UserRequest userRequest) {
        this.userRequest = userRequest;
    }

    public RequestPromise(UserRequest userRequest, Result result) {
        this.userRequest = userRequest;
        this.result = result;
    }

    public UserRequest getUserRequest() {
        return userRequest;
    }

    public void setUserRequest(UserRequest userRequest) {
        this.userRequest = userRequest;
    }

    public Result getResult() {
        return result;
    }

    public void setResult(Result result) {
        this.result = result;
    }

    @Override
    public String toString() {
        return "RequestPromise{" +
                "userRequest=" + userRequest +
                ", result=" + result +
                '}';
    }
}
// 结果通知
class Result {
    private Boolean success;
    private String msg;

    public Result(Boolean success, String msg) {
        this.success = success;
        this.msg = msg;
    }

    public Boolean getSuccess() {
        return success;
    }

    public void setSuccess(Boolean success) {
        this.success = success;
    }

    public String getMsg() {
        return msg;
    }

    public void setMsg(String msg) {
        this.msg = msg;
    }

    @Override
    public String toString() {
        return "Result{" +
                "success=" + success +
                ", msg='" + msg + '\'' +
                '}';
    }
}
// 用户请求
class UserRequest {
    private Long orderId;
    private Long userId;
    private Integer count;

    public UserRequest(Long orderId, Long userId, Integer count) {
        this.orderId = orderId;
        this.userId = userId;
        this.count = count;
    }

    public Long getOrderId() {
        return orderId;
    }

    public void setOrderId(Long orderId) {
        this.orderId = orderId;
    }

    public Long getUserId() {
        return userId;
    }

    public void setUserId(Long userId) {
        this.userId = userId;
    }

    public Integer getCount() {
        return count;
    }

    public void setCount(Integer count) {
        this.count = count;
    }

    @Override
    public String toString() {
        return "UserRequest{" +
                "orderId=" + orderId +
                ", userId=" + userId +
                ", count=" + count +
                '}';
    }
}
1
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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245