feat: 集合队列结束，添加下载视频内容
This commit is contained in:
parent
71e05346af
commit
44cb49b87a
|
|
@ -1,38 +0,0 @@
|
|||
package cn.bunny;
|
||||
|
||||
import junit.framework.Test;
|
||||
import junit.framework.TestCase;
|
||||
import junit.framework.TestSuite;
|
||||
|
||||
/**
|
||||
* Unit test for simple App.
|
||||
*/
|
||||
public class AppTest
|
||||
extends TestCase
|
||||
{
|
||||
/**
|
||||
* Create the test case
|
||||
*
|
||||
* @param testName name of the test case
|
||||
*/
|
||||
public AppTest( String testName )
|
||||
{
|
||||
super( testName );
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the suite of tests being tested
|
||||
*/
|
||||
public static Test suite()
|
||||
{
|
||||
return new TestSuite( AppTest.class );
|
||||
}
|
||||
|
||||
/**
|
||||
* Rigourous Test :-)
|
||||
*/
|
||||
public void testApp()
|
||||
{
|
||||
assertTrue( true );
|
||||
}
|
||||
}
|
||||
|
|
@ -18,6 +18,6 @@
|
|||
</properties>
|
||||
|
||||
<dependencies>
|
||||
|
||||
|
||||
</dependencies>
|
||||
</project>
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
package subtitle;
|
||||
package thead.subtitle;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
|
|
@ -0,0 +1,100 @@
|
|||
package thead.subtitle.video;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.util.StringUtils;
|
||||
import thead.subtitle.video.dao.Response;
|
||||
import thead.subtitle.video.dao.VideoEntity;
|
||||
import thead.subtitle.video.utils.HttpRequestUtils;
|
||||
import thead.subtitle.video.utils.SystemControlUtils;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
@Slf4j
|
||||
public class VideoDownloadVersion1 {
|
||||
// 当前下载页
|
||||
private final static Integer currentPage = 7;
|
||||
|
||||
// 线程池个数
|
||||
private final static Integer threadPoolSize = 10;
|
||||
|
||||
// 开始线程池
|
||||
private final static ExecutorService executorService = Executors.newFixedThreadPool(threadPoolSize);
|
||||
|
||||
public static void main(String[] args) {
|
||||
try {
|
||||
// 记录程序开始时间
|
||||
Instant start = Instant.now();
|
||||
|
||||
String url = "https://mjfh136.cyou/view/videoList/1637462276570050562/" + currentPage + "/80";
|
||||
HttpRequestUtils<Response<VideoEntity>> requestUtils = new HttpRequestUtils<>();
|
||||
Response<VideoEntity> responseData = requestUtils.requestGET(url);
|
||||
|
||||
// 接收到返回信息
|
||||
downloadVideoList(responseData);
|
||||
|
||||
// 执行完成后播放音乐
|
||||
executorService.shutdown();
|
||||
boolean awaitTermination = executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
|
||||
if (awaitTermination) {
|
||||
SystemControlUtils.playMusic();
|
||||
|
||||
// 计算程序运行时间
|
||||
Instant end = Instant.now();
|
||||
Duration duration = Duration.between(start, end);
|
||||
long minutes = duration.toMinutes();
|
||||
long seconds = duration.minusMinutes(minutes).getSeconds();
|
||||
|
||||
log.info("程序运行时间： {} 分钟 {} 秒", minutes, seconds);
|
||||
}
|
||||
} catch (Exception exception) {
|
||||
throw new RuntimeException(exception);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 下载视频列表内容
|
||||
*
|
||||
* @param responseData 返回一页的数据响应
|
||||
*/
|
||||
private static void downloadVideoList(Response<VideoEntity> responseData) {
|
||||
// 使用原子化操作，记录每个下载当前索引
|
||||
AtomicInteger index = new AtomicInteger(0);
|
||||
|
||||
// 拿到每个数据
|
||||
List<VideoEntity> videoEntities = responseData.getData().getList();
|
||||
LinkedBlockingQueue<VideoEntity> videoQueue = new LinkedBlockingQueue<>(videoEntities);
|
||||
videoQueue.stream()
|
||||
.filter(videoEntity -> videoEntity.getPlayUrl().contains("http"))
|
||||
.peek(videoEntity -> {
|
||||
String videoTag = StringUtils.hasText(videoEntity.getVideoTag()) ? videoEntity.getVideoTag() : videoEntity.getVideoTypeTitle();
|
||||
videoEntity.setVideoTag(videoTag);
|
||||
})
|
||||
.forEach((videoEntity) -> executorService.submit(() -> {
|
||||
int currentIndex = index.incrementAndGet();
|
||||
log.info("开始下载第：{}个，线程Id：{}", currentIndex, Thread.currentThread().getName());
|
||||
|
||||
// 整理命令行参数
|
||||
List<String> command = new ArrayList<>();
|
||||
command.add("N_m3u8DL-CLI");
|
||||
command.add("\"" + videoEntity.getPlayUrl() + "\"");
|
||||
command.add("--workDir");
|
||||
command.add("\"G:\\video\\" + videoEntity.getVideoTag() + "\"");
|
||||
command.add("--saveName");
|
||||
command.add("\"" + videoEntity.getTitle().trim() + "\"");
|
||||
command.add("--enableDelAfterDone");
|
||||
command.add("--enableBinaryMerge");
|
||||
command.add("--enableMuxFastStart");
|
||||
|
||||
// 开始下载
|
||||
SystemControlUtils.startProcess(command, currentIndex);
|
||||
}));
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,134 @@
|
|||
package thead.subtitle.video;
|
||||
|
||||
import io.micrometer.common.util.StringUtils;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import thead.subtitle.video.dao.Response;
|
||||
import thead.subtitle.video.dao.VideoEntity;
|
||||
import thead.subtitle.video.utils.HttpRequestUtils;
|
||||
import thead.subtitle.video.utils.SystemControlUtils;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
public class VideoDownloadVersion2 {
|
||||
private static final Logger log = LoggerFactory.getLogger(VideoDownloadVersion2.class);
|
||||
|
||||
// 当前下载页
|
||||
private static final Integer currentPage = 7;
|
||||
|
||||
// 线程池个数
|
||||
private static final Integer threadPoolSize = 10;
|
||||
|
||||
// 下载目录
|
||||
private static final String DOWNLOAD_DIR = "G:\\video\\";
|
||||
|
||||
// 视频下载URL基础部分
|
||||
private static final String BASE_URL = "https://mjfh136.cyou/view/videoList/1637462276570050562/";
|
||||
|
||||
// 开始线程池
|
||||
private static final ExecutorService executorService = Executors.newFixedThreadPool(threadPoolSize);
|
||||
|
||||
public static void main(String[] args) {
|
||||
Instant start = Instant.now();
|
||||
try {
|
||||
String url = BASE_URL + currentPage + "/80";
|
||||
HttpRequestUtils<Response<VideoEntity>> requestUtils = new HttpRequestUtils<>();
|
||||
Response<VideoEntity> responseData = requestUtils.requestGET(url);
|
||||
|
||||
// 接收到返回信息
|
||||
downloadVideoList(responseData);
|
||||
|
||||
// 执行完成后播放音乐
|
||||
shutdownExecutorService();
|
||||
|
||||
// 计算程序运行时间
|
||||
Instant end = Instant.now();
|
||||
Duration duration = Duration.between(start, end);
|
||||
long minutes = duration.toMinutes();
|
||||
long seconds = duration.minusMinutes(minutes).getSeconds();
|
||||
log.info("程序运行时间：{} 分钟 {} 秒", minutes, seconds);
|
||||
|
||||
} catch (Exception exception) {
|
||||
log.error("下载过程中发生异常", exception);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 下载视频列表内容
|
||||
*
|
||||
* @param responseData 返回一页的数据响应
|
||||
*/
|
||||
private static void downloadVideoList(Response<VideoEntity> responseData) {
|
||||
List<VideoEntity> videoEntities = responseData.getData().getList();
|
||||
|
||||
// 使用并行流代替传统的流操作，并且进行命令的预构建
|
||||
videoEntities.parallelStream()
|
||||
.filter(videoEntity -> videoEntity.getPlayUrl().contains("http"))
|
||||
.peek(videoEntity -> {
|
||||
if (StringUtils.isBlank(videoEntity.getVideoTag())) {
|
||||
videoEntity.setVideoTag(videoEntity.getVideoTypeTitle());
|
||||
}
|
||||
})
|
||||
.forEach(videoEntity -> {
|
||||
executorService.submit(() -> {
|
||||
try {
|
||||
String videoTitle = videoEntity.getTitle().trim();
|
||||
String videoUrl = videoEntity.getPlayUrl();
|
||||
|
||||
log.info("开始下载视频：{}，URL：{}", videoTitle, videoUrl);
|
||||
|
||||
List<String> command = buildDownloadCommand(videoEntity);
|
||||
|
||||
// 开始下载
|
||||
SystemControlUtils.startProcess(command);
|
||||
} catch (Exception e) {
|
||||
log.error("下载视频失败，视频标题: {}", videoEntity.getTitle(), e);
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 构建下载命令
|
||||
*
|
||||
* @param videoEntity 视频实体
|
||||
* @return 命令列表
|
||||
*/
|
||||
private static List<String> buildDownloadCommand(VideoEntity videoEntity) {
|
||||
List<String> command = new ArrayList<>();
|
||||
command.add("N_m3u8DL-CLI");
|
||||
command.add("\"" + videoEntity.getPlayUrl() + "\"");
|
||||
command.add("--workDir");
|
||||
command.add("\"" + DOWNLOAD_DIR + videoEntity.getVideoTag() + "\"");
|
||||
command.add("--saveName");
|
||||
command.add("\"" + videoEntity.getTitle().trim() + "\"");
|
||||
command.add("--enableDelAfterDone");
|
||||
command.add("--enableBinaryMerge");
|
||||
command.add("--enableMuxFastStart");
|
||||
|
||||
return command;
|
||||
}
|
||||
|
||||
/**
|
||||
* 优雅地关闭线程池
|
||||
*/
|
||||
private static void shutdownExecutorService() {
|
||||
try {
|
||||
executorService.shutdown();
|
||||
if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
|
||||
executorService.shutdownNow();
|
||||
}
|
||||
SystemControlUtils.playMusic();
|
||||
} catch (InterruptedException e) {
|
||||
executorService.shutdownNow();
|
||||
Thread.currentThread().interrupt();
|
||||
log.error("线程池等待超时，强制关闭线程池", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,10 @@
|
|||
package thead.subtitle.video.dao;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
@Data
|
||||
public class Response<T> {
|
||||
private Integer code;
|
||||
private String message;
|
||||
private ResponseData<T> data;
|
||||
}
|
||||
|
|
@ -0,0 +1,14 @@
|
|||
package thead.subtitle.video.dao;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
@Data
|
||||
public class ResponseData<T> {
|
||||
private Integer pageNumber;
|
||||
private Integer pageSize;
|
||||
private Integer totalPage;
|
||||
private Integer total;
|
||||
private List<T> list;
|
||||
}
|
||||
|
|
@ -0,0 +1,46 @@
|
|||
package thead.subtitle.video.dao;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
@Data
|
||||
public class VideoEntity {
|
||||
|
||||
private String id;
|
||||
|
||||
private String title;
|
||||
|
||||
private String videoTag;
|
||||
|
||||
private String coverUrl;
|
||||
|
||||
private String playUrl;
|
||||
|
||||
private String videoTypeId;
|
||||
|
||||
private String videoTypeTitle;
|
||||
|
||||
private Integer dayPlayCount;
|
||||
|
||||
private Boolean top;
|
||||
|
||||
private String addTime;
|
||||
|
||||
private Integer playCount;
|
||||
|
||||
private Integer collectCount;
|
||||
|
||||
private Integer duration;
|
||||
|
||||
private String videoTags;
|
||||
|
||||
private String userVideoCollectionTime;
|
||||
|
||||
private String userPlayListTime;
|
||||
|
||||
private Boolean isVip;
|
||||
|
||||
private Boolean free;
|
||||
|
||||
private String monthDay;
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,68 @@
|
|||
package thead.subtitle.video.utils;
|
||||
|
||||
import com.alibaba.fastjson2.JSON;
|
||||
import com.alibaba.fastjson2.TypeReference;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import java.net.URI;
|
||||
import java.net.http.HttpClient;
|
||||
import java.net.http.HttpRequest;
|
||||
import java.net.http.HttpResponse;
|
||||
import java.time.Duration;
|
||||
import java.time.temporal.ChronoUnit;
|
||||
|
||||
@Slf4j
|
||||
public class HttpRequestUtils<T> {
|
||||
|
||||
// 创建 HttpClient 实例并复用
|
||||
private static final HttpClient httpClient = HttpClient.newHttpClient();
|
||||
|
||||
/**
|
||||
* 使用GET请求
|
||||
*
|
||||
* @param url 请求地址
|
||||
* @return 返回泛型 T
|
||||
*/
|
||||
public T requestGET(String url) {
|
||||
// 开始时间，用于计算响应时间
|
||||
long startTime = System.currentTimeMillis();
|
||||
|
||||
try {
|
||||
HttpRequest request = HttpRequest.newBuilder()
|
||||
.uri(URI.create(url))
|
||||
.header("Content-Type", "application/json")
|
||||
.header("user-agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36")
|
||||
.GET()
|
||||
.timeout(Duration.of(20, ChronoUnit.SECONDS)) // 请求超时时间
|
||||
.build();
|
||||
|
||||
// 记录请求日志
|
||||
log.info("开始请求 URL：{}", url);
|
||||
|
||||
// 发送请求并获取响应
|
||||
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
|
||||
|
||||
// 记录响应时间
|
||||
long responseTime = System.currentTimeMillis() - startTime;
|
||||
log.info("请求完成，响应时间：{} ms，HTTP 状态码：{}", responseTime, response.statusCode());
|
||||
|
||||
// 判断响应状态码
|
||||
if (response.statusCode() != 200) {
|
||||
log.error("请求失败，HTTP 状态码：{}", response.statusCode());
|
||||
throw new RuntimeException("请求失败，HTTP 状态码：" + response.statusCode());
|
||||
}
|
||||
|
||||
String body = response.body();
|
||||
|
||||
// 使用 TypeReference 来解析泛型类型
|
||||
TypeReference<T> typeReference = new TypeReference<>() {
|
||||
};
|
||||
return JSON.parseObject(body, typeReference.getType());
|
||||
|
||||
} catch (Exception e) {
|
||||
// 详细的异常日志记录
|
||||
log.error("请求异常，URL：{}", url, e);
|
||||
throw new RuntimeException("HTTP 请求失败", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,102 @@
|
|||
package thead.subtitle.video.utils;
|
||||
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import javax.sound.sampled.AudioSystem;
|
||||
import javax.sound.sampled.Clip;
|
||||
import java.io.*;
|
||||
import java.util.List;
|
||||
|
||||
@Slf4j
|
||||
public class SystemControlUtils {
|
||||
|
||||
/**
|
||||
* 播放音乐
|
||||
*/
|
||||
public static void playMusic() {
|
||||
ClassLoader classLoader = SystemControlUtils.class.getClassLoader();
|
||||
try (InputStream inputStream = classLoader.getResourceAsStream("static/download-complete/download-complete.wav")) {
|
||||
if (inputStream == null) {
|
||||
log.warn("无法找到音乐文件！");
|
||||
return;
|
||||
}
|
||||
|
||||
try (Clip clip = AudioSystem.getClip()) {
|
||||
clip.open(AudioSystem.getAudioInputStream(inputStream));
|
||||
clip.start();
|
||||
|
||||
// 等待音乐播放完成
|
||||
Thread.sleep(clip.getMicrosecondLength() / 1000);
|
||||
} catch (Exception e) {
|
||||
log.error("播放音乐时发生错误", e);
|
||||
throw new RuntimeException("播放音乐时发生错误", e);
|
||||
}
|
||||
|
||||
} catch (IOException e) {
|
||||
log.error("加载音乐文件失败", e);
|
||||
throw new RuntimeException("加载音乐文件失败", e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 执行命令行并输出（带有当前下载索引）
|
||||
*
|
||||
* @param command 命令内容
|
||||
* @param currentIndex 当前下载的第几个
|
||||
*/
|
||||
public static void startProcess(List<String> command, int currentIndex) {
|
||||
startProcess(command, currentIndex, "D:\\software\\Plugins\\Mu3U8下载\\");
|
||||
}
|
||||
|
||||
/**
|
||||
* 执行命令行并输出
|
||||
*
|
||||
* @param command 命令内容
|
||||
*/
|
||||
public static void startProcess(List<String> command) {
|
||||
startProcess(command, -1, "D:\\software\\Plugins\\Mu3U8下载\\");
|
||||
}
|
||||
|
||||
/**
|
||||
* 执行命令行并输出（抽象出的公共逻辑）
|
||||
*
|
||||
* @param command 命令内容
|
||||
* @param currentIndex 当前下载的第几个（可选）
|
||||
* @param workDir 工作目录
|
||||
*/
|
||||
private static void startProcess(List<String> command, int currentIndex, String workDir) {
|
||||
try {
|
||||
// 创建进程构建器
|
||||
ProcessBuilder processBuilder = new ProcessBuilder(command);
|
||||
processBuilder.directory(new File(workDir));
|
||||
|
||||
// 启动进程
|
||||
Process process = processBuilder.start();
|
||||
|
||||
// 读取命令输出，指定使用 GBK 编码
|
||||
try (BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream(), "GBK"))) {
|
||||
String line;
|
||||
while ((line = reader.readLine()) != null) {
|
||||
if (currentIndex != -1) {
|
||||
log.info("第 {} 个下载任务输出: {}", currentIndex, line);
|
||||
} else {
|
||||
log.info(line);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 等待命令执行完成
|
||||
int exitCode = process.waitFor();
|
||||
if (currentIndex != -1) {
|
||||
log.info("第 {} 个任务执行完毕，退出码: {}", currentIndex, exitCode);
|
||||
} else {
|
||||
log.info("命令执行完毕，退出码: {}", exitCode);
|
||||
}
|
||||
|
||||
} catch (IOException | InterruptedException e) {
|
||||
log.error("执行命令时发生错误", e);
|
||||
throw new RuntimeException("执行命令时发生错误", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,4 +1,4 @@
|
|||
package init;
|
||||
package thead.thread_0;
|
||||
|
||||
public class DeadlockExample {
|
||||
private static final Object lock1 = new Object();
|
||||
|
|
@ -34,7 +34,6 @@ public class DeadlockExample {
|
|||
}
|
||||
}
|
||||
});
|
||||
|
||||
thread1.start();
|
||||
thread2.start();
|
||||
}
|
||||
|
|
@ -0,0 +1,11 @@
|
|||
package thead.thread_8;
|
||||
|
||||
public class MyThread extends Thread {
|
||||
@Override
|
||||
public void run() {
|
||||
super.run();
|
||||
for (int i = 0; i < 500000; i++) {
|
||||
System.out.println("i=" + (i + 1));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,19 @@
|
|||
package thead.thread_8;
|
||||
|
||||
public class Run {
|
||||
public static void main(String[] args) {
|
||||
try {
|
||||
MyThread thread = new MyThread();
|
||||
thread.start();
|
||||
Thread.sleep(1000);
|
||||
thread.interrupt();
|
||||
|
||||
// 输出false表示当前main从未中断过 interrupted 表示当前线程是否已经是中断状态，执行后清除状态标志为false的功能
|
||||
System.out.println("是否要停止1？=" + Thread.interrupted());
|
||||
System.out.println("是否要停止2？=" + Thread.interrupted());
|
||||
} catch (Exception exception) {
|
||||
System.out.println("main catch");
|
||||
exception.printStackTrace();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,19 @@
|
|||
package thead.thread_8;
|
||||
|
||||
public class Run3 {
|
||||
public static void main(String[] args) {
|
||||
try {
|
||||
MyThread thread = new MyThread();
|
||||
thread.start();
|
||||
Thread.sleep(1000);
|
||||
thread.interrupt();
|
||||
|
||||
// 输出为true，但是会有小概率会出现false，isInterrupted 所在的线程是否是中断状态，不清除状态标志
|
||||
System.out.println("是否要停止1？=" + thread.isInterrupted());
|
||||
System.out.println("是否要停止2？=" + thread.isInterrupted());
|
||||
} catch (Exception exception) {
|
||||
System.out.println("main catch");
|
||||
exception.printStackTrace();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,52 @@
|
|||
package thead.thread_9;
|
||||
|
||||
import tools.Box;
|
||||
|
||||
public class MyThread extends Thread {
|
||||
/**
|
||||
* If this thread was constructed using a separate
|
||||
* {@code Runnable} run object, then that
|
||||
* {@code Runnable} object's {@code run} method is called;
|
||||
* otherwise, this method does nothing and returns.
|
||||
* <p>
|
||||
* Subclasses of {@code Thread} should override this method.
|
||||
*
|
||||
* @see #start()
|
||||
*/
|
||||
@Override
|
||||
public void run() {
|
||||
super.run();
|
||||
try {
|
||||
for (; ; ) {
|
||||
if (this.isInterrupted()) {
|
||||
throw new InterruptedException("线程中断！");
|
||||
}
|
||||
|
||||
for (int i = 0; i < 10000; i++) {
|
||||
String a = "" + Math.random();
|
||||
}
|
||||
Box.list1.add("生产数据A");
|
||||
System.out.println("list1 size=" + Box.list1.size());
|
||||
}
|
||||
} catch (Exception exception) {
|
||||
exception.printStackTrace();
|
||||
}
|
||||
|
||||
try {
|
||||
while (true) {
|
||||
if (this.isInterrupted()) {
|
||||
throw new InterruptedException("线程中断！");
|
||||
}
|
||||
|
||||
for (int i = 0; i < 10000; i++) {
|
||||
String a = "" + Math.random();
|
||||
}
|
||||
|
||||
Box.list2.add("生产数据B");
|
||||
System.out.println("list2 size=" + Box.list2.size());
|
||||
}
|
||||
} catch (Exception exception) {
|
||||
exception.printStackTrace();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,31 @@
|
|||
package thead.thread_9;
|
||||
|
||||
import tools.Box;
|
||||
|
||||
public class Test1 {
|
||||
public static void main(String[] args) {
|
||||
try {
|
||||
MyThread thread = new MyThread();
|
||||
thread.start();
|
||||
|
||||
boolean list1IsInterrupted = false;
|
||||
boolean list2IsInterrupted = false;
|
||||
|
||||
while (thread.isAlive()) {
|
||||
if (Box.list1.size() > 500 && !list1IsInterrupted) {
|
||||
thread.interrupt();
|
||||
list1IsInterrupted = true;
|
||||
}
|
||||
|
||||
if (Box.list2.size() > 500 && !list2IsInterrupted) {
|
||||
thread.interrupt();
|
||||
list2IsInterrupted = true;
|
||||
}
|
||||
|
||||
Thread.sleep(50);
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,31 @@
|
|||
package thread_1.safe0;
|
||||
|
||||
import lombok.Getter;
|
||||
|
||||
@Getter
|
||||
public class SafeCounter {
|
||||
private static Integer count = 0;
|
||||
|
||||
// 使用 synchronized 关键字
|
||||
public static void main(String[] args) throws InterruptedException {
|
||||
Thread[] threads = new Thread[1000];
|
||||
for (int i = 0; i < 1000; i++) {
|
||||
threads[i] = new Thread(() -> {
|
||||
for (int j = 0; j < 1000; j++) {
|
||||
synchronized ("") {
|
||||
count++;
|
||||
}
|
||||
}
|
||||
});
|
||||
threads[i].start();
|
||||
}
|
||||
|
||||
// 等待所有线程完成
|
||||
for (int i = 0; i < 1000; i++) {
|
||||
threads[i].join();
|
||||
}
|
||||
|
||||
// 输出最终计数值
|
||||
System.out.println("Count: " + count); // 结果应该等于 1000000
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,29 @@
|
|||
package thread_1.safe0;
|
||||
|
||||
import lombok.Getter;
|
||||
|
||||
@Getter
|
||||
public class UnsafeCounter {
|
||||
private static Integer count = 0;
|
||||
|
||||
public static void main(String[] args) throws InterruptedException {
|
||||
// 创建 1000 个线程并进行计数
|
||||
Thread[] threads = new Thread[1000];
|
||||
for (int i = 0; i < 1000; i++) {
|
||||
threads[i] = new Thread(() -> {
|
||||
for (int j = 0; j < 1000; j++) {
|
||||
count++;
|
||||
}
|
||||
});
|
||||
threads[i].start();
|
||||
}
|
||||
|
||||
// 等待所有线程完成
|
||||
for (int i = 0; i < 1000; i++) {
|
||||
threads[i].join();
|
||||
}
|
||||
|
||||
// 输出最终计数值
|
||||
System.out.println("Count: " + count); // 结果应该等于 1000000
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,37 @@
|
|||
package thread_1.safe1;
|
||||
|
||||
import lombok.Getter;
|
||||
|
||||
import java.util.concurrent.locks.Lock;
|
||||
import java.util.concurrent.locks.ReentrantLock;
|
||||
|
||||
@Getter
|
||||
public class SafeCounter {
|
||||
|
||||
private static final Lock lock = new ReentrantLock();
|
||||
|
||||
private static Integer count = 0;
|
||||
|
||||
public static void main(String[] args) throws InterruptedException {
|
||||
// 创建 1000 个线程并进行计数
|
||||
Thread[] threads = new Thread[1000];
|
||||
for (int i = 0; i < 1000; i++) {
|
||||
threads[i] = new Thread(() -> {
|
||||
for (int j = 0; j < 1000; j++) {
|
||||
lock.lock();
|
||||
count++;
|
||||
lock.unlock();
|
||||
}
|
||||
});
|
||||
threads[i].start();
|
||||
}
|
||||
|
||||
// 等待所有线程完成
|
||||
for (int i = 0; i < 1000; i++) {
|
||||
threads[i].join();
|
||||
}
|
||||
|
||||
// 输出最终计数值
|
||||
System.out.println("Count: " + count); // 结果应该等于 1000000
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,27 @@
|
|||
package thread_1.safe2;
|
||||
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
public class AtomicSafe {
|
||||
private static final AtomicInteger count = new AtomicInteger(0);
|
||||
|
||||
public static void main(String[] args) throws InterruptedException {
|
||||
// 创建 1000 个线程并进行计数
|
||||
Thread[] threads = new Thread[1000];
|
||||
for (int i = 0; i < 1000; i++) {
|
||||
threads[i] = new Thread(() -> {
|
||||
for (int j = 0; j < 1000; j++) {
|
||||
count.incrementAndGet();
|
||||
}
|
||||
});
|
||||
threads[i].start();
|
||||
}
|
||||
|
||||
// 等待所有线程完成
|
||||
for (Thread t : threads) {
|
||||
t.join();
|
||||
}
|
||||
|
||||
System.out.println(count.get());
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,60 @@
|
|||
package thread_1.safe3;
|
||||
|
||||
public class Run {
|
||||
public static void main(String[] args) {
|
||||
PrivateNum privateNum1 = new PrivateNum();
|
||||
PrivateNum privateNum2 = new PrivateNum();
|
||||
|
||||
new ThreadA(privateNum1).start();
|
||||
new ThreadB(privateNum2).start();
|
||||
}
|
||||
|
||||
/**
|
||||
* 在这个方法中前面是否加 synchronized 对结果都没有影响，因为资源不是共享的
|
||||
*/
|
||||
public static class PrivateNum {
|
||||
synchronized public void testMethod() {
|
||||
try {
|
||||
System.out.println(Thread.currentThread().getName() + " 开始 " + System.currentTimeMillis());
|
||||
Thread.sleep(3000);
|
||||
System.out.println(Thread.currentThread().getName() + " 结束 " + System.currentTimeMillis());
|
||||
} catch (Exception exception) {
|
||||
exception.printStackTrace();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 线程A
|
||||
public static class ThreadA extends Thread {
|
||||
|
||||
private final PrivateNum privateNum;
|
||||
|
||||
public ThreadA(PrivateNum privateNum) {
|
||||
super();
|
||||
this.privateNum = privateNum;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
super.run();
|
||||
privateNum.testMethod();
|
||||
}
|
||||
}
|
||||
|
||||
// 线程B
|
||||
public static class ThreadB extends Thread {
|
||||
|
||||
private final PrivateNum privateNum;
|
||||
|
||||
public ThreadB(PrivateNum privateNum) {
|
||||
super();
|
||||
this.privateNum = privateNum;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
super.run();
|
||||
privateNum.testMethod();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,63 @@
|
|||
package thread_1.safe4;
|
||||
|
||||
public class SynchronizedMethodLocakObject {
|
||||
|
||||
public static void main(String[] args) {
|
||||
MyObject myObject = new MyObject();
|
||||
|
||||
ThreadA threadA = new ThreadA(myObject);
|
||||
threadA.setName("ThreadA");
|
||||
threadA.start();
|
||||
|
||||
ThreadB threadB = new ThreadB(myObject);
|
||||
threadB.setName("ThreadB");
|
||||
threadB.start();
|
||||
}
|
||||
|
||||
// 对象
|
||||
public static class MyObject {
|
||||
synchronized public void methodA() {
|
||||
try {
|
||||
System.out.println(Thread.currentThread().getName() + " 开始 " + System.currentTimeMillis());
|
||||
Thread.sleep(1000);
|
||||
System.out.println(Thread.currentThread().getName() + " 结束 " + System.currentTimeMillis());
|
||||
} catch (Exception exception) {
|
||||
exception.printStackTrace();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 线程A
|
||||
public static class ThreadA extends Thread {
|
||||
|
||||
private final MyObject myObject;
|
||||
|
||||
public ThreadA(MyObject myObject) {
|
||||
super();
|
||||
this.myObject = myObject;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
super.run();
|
||||
myObject.methodA();
|
||||
}
|
||||
}
|
||||
|
||||
// 线程B
|
||||
public static class ThreadB extends Thread {
|
||||
|
||||
private final MyObject myObject;
|
||||
|
||||
public ThreadB(MyObject myObject) {
|
||||
super();
|
||||
this.myObject = myObject;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
super.run();
|
||||
myObject.methodA();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,83 @@
|
|||
package thread_1.safe4;
|
||||
|
||||
public class SynchronizedMethodLocakObject2 {
|
||||
|
||||
public static void main(String[] args) {
|
||||
MyObject myObject = new MyObject();
|
||||
|
||||
ThreadA threadA = new ThreadA(myObject);
|
||||
threadA.setName("ThreadA");
|
||||
|
||||
ThreadB threadB = new ThreadB(myObject);
|
||||
threadB.setName("ThreadB");
|
||||
|
||||
threadA.start();
|
||||
threadB.start();
|
||||
}
|
||||
|
||||
// 测试对象
|
||||
public static class MyObject {
|
||||
|
||||
/**
|
||||
* 测试方法A
|
||||
*/
|
||||
synchronized public void methodA() {
|
||||
try {
|
||||
System.out.println(Thread.currentThread().getName() + " 开始 " + System.currentTimeMillis());
|
||||
Thread.sleep(1000);
|
||||
System.out.println(Thread.currentThread().getName() + " 结束 " + System.currentTimeMillis());
|
||||
} catch (Exception exception) {
|
||||
exception.printStackTrace();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 测试方法B
|
||||
* 如果不加synchronized，整个程序是异步执行的
|
||||
* 如果加synchronized，整个程序是同步执行的
|
||||
*/
|
||||
synchronized public void methodB() {
|
||||
try {
|
||||
System.out.println(Thread.currentThread().getName() + " 开始 " + System.currentTimeMillis());
|
||||
Thread.sleep(1000);
|
||||
System.out.println(Thread.currentThread().getName() + " 结束 " + System.currentTimeMillis());
|
||||
} catch (Exception exception) {
|
||||
exception.printStackTrace();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 线程A
|
||||
public static class ThreadA extends Thread {
|
||||
|
||||
private final MyObject myObject;
|
||||
|
||||
public ThreadA(MyObject myObject) {
|
||||
super();
|
||||
this.myObject = myObject;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
super.run();
|
||||
myObject.methodA();
|
||||
}
|
||||
}
|
||||
|
||||
// 线程B
|
||||
public static class ThreadB extends Thread {
|
||||
|
||||
private final MyObject myObject;
|
||||
|
||||
public ThreadB(MyObject myObject) {
|
||||
super();
|
||||
this.myObject = myObject;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
super.run();
|
||||
myObject.methodB();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,42 @@
|
|||
package thread_1.safe5;
|
||||
|
||||
// 在多线程编程中，synchronized 关键字可以确保一个线程在执行某个方法时，其他线程无法同时访问被同步的方法，从而避免多个线程同时修改共享数据的问题。
|
||||
// 脏读（Dirty Read）指的是一个线程读取到另一个线程还未提交的修改数据，这种数据在另一个线程最终提交或者回滚后可能会发生变化，因此是不可靠的。
|
||||
public class DirtyReadExample {
|
||||
|
||||
private Integer sharedValue = 0; // 共享变量
|
||||
|
||||
// 假设我们有一个多线程程序，其中多个线程同时对共享变量进行操作，未对共享变量进行适当的同步，就可能会发生脏读。
|
||||
// 例如，线程 A 读取了线程 B 未提交的修改，或者线程 B 在 A 修改时正在读取数据，导致读取到的数据不一致。
|
||||
public static void main(String[] args) {
|
||||
DirtyReadExample example = new DirtyReadExample();
|
||||
Thread threadA = new Thread(example::threadA);
|
||||
Thread threadB = new Thread(example::threadB);
|
||||
|
||||
threadA.start();
|
||||
threadB.start();
|
||||
}
|
||||
|
||||
// 线程A方法：修改共享变量
|
||||
synchronized public void threadA() {
|
||||
sharedValue = 100;
|
||||
System.out.println("ThreadA 修改 sharedValue 为: " + sharedValue);
|
||||
try {
|
||||
Thread.sleep(2000); // 模拟长时间操作
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
sharedValue = 200;
|
||||
System.out.println("ThreadA 修改 sharedValue 为: " + sharedValue);
|
||||
}
|
||||
|
||||
// 线程B方法：读取共享变量
|
||||
synchronized public void threadB() {
|
||||
try {
|
||||
Thread.sleep(1000); // 确保线程A先运行
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
System.out.println("ThreadB 读取 sharedValue: " + sharedValue); // 读取共享变量
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,40 @@
|
|||
package thread_1.safe6;
|
||||
|
||||
/**
|
||||
* sychronized 锁重入指的是一个线程在获得一个锁之后，可以再次进入该锁（即再次获取锁）而不会发生死锁。
|
||||
* 这是因为 sychronized 锁是可重入的（reentrant）。
|
||||
* 当一个线程获取锁并进入一个同步方法时，它的锁计数会增加。
|
||||
* 该线程在执行另一个同步方法时，锁计数继续增加，表示它可以再次进入同步方法。
|
||||
* 只有当线程释放锁时，锁计数才会递减，最终锁计数为 0 时，锁才会被释放。
|
||||
*/
|
||||
public class SynchronizedReentrantExample {
|
||||
|
||||
public static void main(String[] args) {
|
||||
SynchronizedReentrantExample example = new SynchronizedReentrantExample();
|
||||
|
||||
// 创建一个线程来调用 method1
|
||||
Thread thread = new Thread(example::method1);
|
||||
thread.start();
|
||||
}
|
||||
|
||||
// 这个方法是同步的，任何线程都只能一个接着一个访问它
|
||||
public synchronized void method1() {
|
||||
System.out.println("In method1.");
|
||||
|
||||
// 在 method1 中调用 method2，这是对同一个锁的重入
|
||||
method2();
|
||||
}
|
||||
|
||||
// 这个方法也是同步的，线程会在这里等待直到获得锁
|
||||
public synchronized void method2() {
|
||||
System.out.println("In method2.");
|
||||
|
||||
// 在 method2 中调用 method3，这是对同一个锁的重入
|
||||
method3();
|
||||
}
|
||||
|
||||
// 这个方法也是同步的，继续重入锁
|
||||
public synchronized void method3() {
|
||||
System.out.println("In method3.");
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,24 @@
|
|||
package thread_1.safe7;
|
||||
|
||||
import lombok.Getter;
|
||||
|
||||
@Getter
|
||||
public class Counter {
|
||||
private final Object lock1 = new Object();
|
||||
private final Object lock2 = new Object();
|
||||
private int count1 = 0;
|
||||
private int count2 = 0;
|
||||
|
||||
public void incrementCount1() {
|
||||
synchronized (lock1) {
|
||||
count1++;
|
||||
}
|
||||
}
|
||||
|
||||
public void incrementCount2() {
|
||||
synchronized (lock2) {
|
||||
count2++;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
package thread_1.safe8;
|
||||
|
||||
import lombok.Getter;
|
||||
|
||||
public class Counter {
|
||||
@Getter
|
||||
private static int count = 0;
|
||||
|
||||
// 静态同步方法
|
||||
public synchronized static void increment() {
|
||||
count++;
|
||||
}
|
||||
|
||||
public static void main(String[] args) throws InterruptedException {
|
||||
// 创建多个线程调用静态同步方法
|
||||
Thread t1 = new Thread(() -> {
|
||||
for (int i = 0; i < 1000; i++) {
|
||||
Counter.increment(); // 调用静态同步方法
|
||||
}
|
||||
});
|
||||
|
||||
Thread t2 = new Thread(() -> {
|
||||
for (int i = 0; i < 1000; i++) {
|
||||
Counter.increment(); // 调用静态同步方法
|
||||
}
|
||||
});
|
||||
|
||||
t1.start();
|
||||
t2.start();
|
||||
|
||||
t1.join();
|
||||
t2.join();
|
||||
|
||||
System.out.println("Final count: " + Counter.getCount());
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,56 @@
|
|||
package thread_1.safe9;
|
||||
|
||||
import java.util.concurrent.ConcurrentSkipListMap;
|
||||
import java.util.concurrent.ConcurrentSkipListSet;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
import java.util.concurrent.CopyOnWriteArraySet;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
public class AtomicIntegerTest {
|
||||
public static void main(String[] args) {
|
||||
MyThread myThread = new MyThread();
|
||||
|
||||
Thread thread1 = new Thread(myThread);
|
||||
Thread thread2 = new Thread(myThread);
|
||||
Thread thread3 = new Thread(myThread);
|
||||
Thread thread4 = new Thread(myThread);
|
||||
Thread thread5 = new Thread(myThread);
|
||||
Thread thread6 = new Thread(myThread);
|
||||
Thread thread7 = new Thread(myThread);
|
||||
Thread thread8 = new Thread(myThread);
|
||||
Thread thread9 = new Thread(myThread);
|
||||
Thread thread10 = new Thread(myThread);
|
||||
|
||||
thread1.start();
|
||||
thread2.start();
|
||||
thread3.start();
|
||||
thread4.start();
|
||||
thread5.start();
|
||||
thread6.start();
|
||||
thread7.start();
|
||||
thread8.start();
|
||||
thread9.start();
|
||||
thread10.start();
|
||||
|
||||
ConcurrentSkipListMap<Integer, String> map = new ConcurrentSkipListMap<>();
|
||||
ConcurrentSkipListSet<String> stringConcurrentSkipListSet = new ConcurrentSkipListSet<>();
|
||||
|
||||
CopyOnWriteArrayList<Integer> integers = new CopyOnWriteArrayList<>();
|
||||
CopyOnWriteArraySet<Integer> integerCopyOnWriteArraySet = new CopyOnWriteArraySet<>();
|
||||
|
||||
new ConcurrentSkipListSet<Object>();
|
||||
new ConcurrentSkipListMap<>();
|
||||
new ConcurrentSkipListSet<>();
|
||||
}
|
||||
|
||||
static class MyThread extends Thread {
|
||||
private final AtomicInteger integer = new AtomicInteger(0);
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
for (int i = 0; i < 10; i++) {
|
||||
System.out.println(integer.incrementAndGet());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,8 @@
|
|||
package tools;
|
||||
|
||||
import java.util.ArrayList;
|
||||
|
||||
public class Box {
|
||||
public static ArrayList<Object> list1 = new ArrayList<>();
|
||||
public static ArrayList<Object> list2 = new ArrayList<>();
|
||||
}
|
||||
Binary file not shown.
Binary file not shown.
17
pom.xml
17
pom.xml
|
|
@ -3,7 +3,12 @@
|
|||
xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-parent</artifactId>
|
||||
<version>3.4.1</version>
|
||||
<relativePath/> <!-- lookup parent from repository -->
|
||||
</parent>
|
||||
<groupId>cn.bunny</groupId>
|
||||
<artifactId>MultiThread</artifactId>
|
||||
<version>1.0-SNAPSHOT</version>
|
||||
|
|
@ -32,11 +37,21 @@
|
|||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-web</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.projectlombok</groupId>
|
||||
<artifactId>lombok</artifactId>
|
||||
<version>1.18.36</version>
|
||||
</dependency>
|
||||
<!-- fastjson2 -->
|
||||
<dependency>
|
||||
<groupId>com.alibaba.fastjson2</groupId>
|
||||
<artifactId>fastjson2</artifactId>
|
||||
<version>2.0.47</version>
|
||||
</dependency>
|
||||
<!-- slf4j -->
|
||||
<dependency>
|
||||
<groupId>org.slf4j</groupId>
|
||||
|
|
|
|||
Loading…
Reference in New Issue