学成在线day06

server/2024/11/30 1:13:39/

上传视屏

断点续传

通常视频文件都比较大，所以对于媒资系统上传文件的需求要满足大文件的上传要求。http协议本身对上传文件大小没有限制，但是客户的网络环境质量、电脑硬件环境等参差不齐，如果一个大文件快上传完了网断了没有上传完成，需要客户重新上传，用户体验非常差，所以对于大文件上传的要求最基本的是断点续传。

什么是断点续传：

引用百度百科：断点续传指的是在下载或上传时，将下载或上传任务（一个文件或一个压缩包）人为的划分为几个部分，每一个部分采用一个线程进行上传或下载，如果碰到网络故障，可以从已经上传或下载的部分开始继续上传下载未完成的部分，而没有必要从头开始上传下载，断点续传可以提高节省操作时间，提高用户体验性。

断点续传流程如下图：

流程如下：

1、前端上传前先把文件分成块

2、一块一块的上传，上传中断后重新上传，已上传的分块则不用再上传

3、各分块上传完成最后在服务端合并文件

文件分块测试

java">package com.xuecheng.media;import org.junit.jupiter.api.Test;
import org.springframework.util.DigestUtils;
import java.io.*;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;public class TestBigFile {//文件分块@Testpublic void testChunkUpload() throws Exception{//源文件File sourceFile = new File("D:\\test\\1.mp4");//目标文件位置String ChunkFiles = "D:\\test\\chunk\\";//设置分块大小int chunkSize = 1024 * 1024 * 10;//获取文件要分块的个数Integer size =(int) Math.ceil(sourceFile.length() * 1.0 / chunkSize);System.out.println(size);//使用RandomAccessFile访问文件RandomAccessFile raf_read = new RandomAccessFile(sourceFile, "r");//创建缓冲区byte[] b = new byte[1024*1024];for (Integer i = 0; i < size; i++) {//创建分块文件File chunkFile = new File(ChunkFiles+i);//判断该文件是否存在if (chunkFile.exists()) {System.out.println("分块文件已存在");//将该文件删除chunkFile.delete();}//创建临时文件boolean newFile = chunkFile.createNewFile();if (newFile){//用于记录本次读取的字节数int len = -1;//使用RandomAccessFile访问文件RandomAccessFile raf_w = new RandomAccessFile(chunkFile, "rw");while ((len = raf_read.read(b))!= -1){//将分块文件写入到输出流raf_w.write(b,0,len);if (chunkFile.length() >= chunkSize){break;}}//关闭流raf_w.close();}}raf_read.close();}//文件合并@Testpublic void testMerge() throws Exception{//源文件File sourceFile = new File("D:\\test\\1.mp4");//分块文件的文件夹File chunkFilesFolder = new File("D:\\test\\chunk");//合并文件位置File mergeFile = new File("D:\\test\\1_test.mp4");//根据文件夹获取该文件夹中的文件列表File[] files = chunkFilesFolder.listFiles();//将数组转化为list集合List<File> fileList = Arrays.asList(files);//集合排序Collections.sort(fileList, new Comparator<File>() {@Overridepublic int compare(File o1, File o2) {return Integer.parseInt(o1.getName()) - Integer.parseInt(o2.getName());}});//合并文件//创建临时文件mergeFile.createNewFile();//创建缓冲区byte[] b = new byte[1024];RandomAccessFile raf_write = new RandomAccessFile(mergeFile, "rw");for (File file : fileList) {//使用RandomAccessFile访问文件RandomAccessFile raf_read = new RandomAccessFile(file, "r");int len = -1;while ((len = raf_read.read(b)) != -1) {//将分块文件写入到合并文件raf_write.write(b, 0, len);}raf_read.close();}raf_write.close();//校验文件String mergeFileMd5 = DigestUtils.md5DigestAsHex(new FileInputStream(mergeFile));String sourceFileMd5 = DigestUtils.md5DigestAsHex(new FileInputStream(sourceFile));if (mergeFileMd5.equals(sourceFileMd5)){System.out.println("合并成功");}}
}

文件分块上传minio，合并文件，使用minio接口来合并文件：

java">    //文件上传minio@org.junit.jupiter.api.Testpublic void testUploadminio() throws Exception {
//        D:\test\chunkfor (int i = 0; i < 7; i++) {//准备数据UploadObjectArgs testbucket = UploadObjectArgs.builder().bucket("testbucket").object("chunk/"+i)//文件要存放的路径和文件名.filename("D:\\test\\chunk\\"+i).build();minioClient.uploadObject(testbucket);}}//将minio中的文件合并,minio默认文件分块大小要超过5mb@org.junit.jupiter.api.Testpublic void testMerge() throws Exception{List<ComposeSource> sources = new ArrayList<>();for (int i = 0; i < 7; i++) {ComposeSource source = ComposeSource.builder().bucket("testbucket").object("chunk/"+i).build();sources.add(source);}//        sources = (List<ComposeSource>) Stream.iterate(0, i -> ++i).limit(7).map(i -> ComposeSource.builder().bucket("testbucket").object("chunk/" + i).build());ComposeObjectArgs testbucket = ComposeObjectArgs.builder().bucket("testbucket").object("1.mp4").sources(sources).build();minioClient.composeObject(testbucket);}

文件上传相关接口

文件上传前检查

java">    @ApiOperation(value = "文件上传前检查文件")@PostMapping("/upload/checkfile")public RestResponse<Boolean> checkfile(@RequestParam("fileMd5") String fileMd5) throws Exception {RestResponse<Boolean> booleanRestResponse = mediaFileService.checkFile(fileMd5);return booleanRestResponse;}@ApiOperation(value = "分块文件上传前的检测")@PostMapping("/upload/checkchunk")public RestResponse<Boolean> checkchunk(@RequestParam("fileMd5") String fileMd5,@RequestParam("chunk") int chunk) throws Exception {RestResponse<Boolean> booleanRestResponse = mediaFileService.checkChunk(fileMd5, chunk);return booleanRestResponse;}
service层
java"> /*** @description 检查文件是否存在* @param fileMd5 文件的md5* @return com.xuecheng.base.model.RestResponse<java.lang.Boolean> false不存在，true存在* @author Mr.M* @date 2022/9/13 15:38*/public RestResponse<Boolean> checkFile(String fileMd5);/*** @description 检查分块是否存在* @param fileMd5  文件的md5* @param chunkIndex  分块序号* @return com.xuecheng.base.model.RestResponse<java.lang.Boolean> false不存在，true存在* @author Mr.M* @date 2022/9/13 15:39*/public RestResponse<Boolean> checkChunk(String fileMd5, int chunkIndex);
java">    /*** 检查文件是否存在* @param fileMd5 文件的md5* @return*/@Overridepublic RestResponse<Boolean> checkFile(String fileMd5) {//检查数据库中是否存在该文件MediaFiles mediaFiles = mediaFilesMapper.selectById(fileMd5);if(mediaFiles != null){//数据库存在则检查minio中是否存在//初始化数据//存储目录String filePath = mediaFiles.getFilePath();//文件流InputStream stream = null;try {GetObjectArgs testbucket = GetObjectArgs.builder().bucket(bucket_video).object(filePath).build();//读取数据获取到输入流stream = minioClient.getObject(testbucket);//判断是否获取到流if (stream != null){return RestResponse.success(true);}} catch (Exception e) {throw new RuntimeException(e);}}return RestResponse.success(false);}/*** 检查分块是否存在* @param fileMd5  文件的md5* @param chunkIndex  分块序号* @return*/@Overridepublic RestResponse<Boolean> checkChunk(String fileMd5, int chunkIndex) {//得到分块文件目录String chunkFileFolderPath = getChunkFileFolderPath(fileMd5);//得到分块文件的路径String chunkFilePath = chunkFileFolderPath + chunkIndex;//文件流InputStream fileInputStream = null;try {fileInputStream = minioClient.getObject(GetObjectArgs.builder().bucket(bucket_video).object(chunkFilePath).build());if (fileInputStream != null) {//分块已存在return RestResponse.success(true);}} catch (Exception e) {}//分块未存在return RestResponse.success(false);}//获取分块路径//得到分块文件的目录private String getChunkFileFolderPath(String fileMd5) {return fileMd5.substring(0, 1) + "/" + fileMd5.substring(1, 2) + "/" + fileMd5 + "/" + "chunk" + "/";}

文件上传和合并

接口层
java">@ApiOperation(value = "上传分块文件")@PostMapping("/upload/uploadchunk")public RestResponse uploadchunk(@RequestParam("file") MultipartFile file,@RequestParam("fileMd5") String fileMd5,@RequestParam("chunk") int chunk) throws Exception {//创建临时文件,用来存储分块文件File tempFile = File.createTempFile("minio", "temp");//上传的文件拷贝到临时文件file.transferTo(tempFile);//文件路径String absolutePath = tempFile.getAbsolutePath();log.error("上传文件路径：{}",absolutePath);log.error("上传文件md5：{}",fileMd5);return mediaFileService.uploadChunk(fileMd5, chunk,absolutePath);}@ApiOperation(value = "合并文件")@PostMapping("/upload/mergechunks")public RestResponse mergechunks(@RequestParam("fileMd5") String fileMd5,@RequestParam("fileName") String fileName,@RequestParam("chunkTotal") int chunkTotal) throws Exception {Long companyId = 1232141425L;UploadFileParamsDto uploadFileParamsDto = new UploadFileParamsDto();uploadFileParamsDto.setFilename(fileName);uploadFileParamsDto.setFileType("001002");uploadFileParamsDto.setTags("课程视频");uploadFileParamsDto.setRemark("");return mediaFileService.mergechunks(companyId, fileMd5, chunkTotal, uploadFileParamsDto);}
service层
java">/*** 分块上传文件* @param fileMd5 文件md5* @param chunk 分块序号* @param localChunkFilePath 文件的本地路径* @return*/public RestResponse uploadChunk(String fileMd5,int chunk,String localChunkFilePath);/*** @description 合并分块* @param companyId  机构id* @param fileMd5  文件md5* @param chunkTotal 分块总和* @param uploadFileParamsDto 文件信息* @return com.xuecheng.base.model.RestResponse* @author Mr.M* @date 2022/9/13 15:56*/public RestResponse mergechunks(Long companyId,String fileMd5,int chunkTotal,UploadFileParamsDto uploadFileParamsDto);

实现：

java">  /*** 上传分块文件到minio* @param fileMd5 文件md5* @param chunk 分块序号* @param localChunkFilePath 文件的本地路径* @return*/@Overridepublic RestResponse uploadChunk(String fileMd5, int chunk, String localChunkFilePath) {//获取文件的保存父路径String chunkFileFolderPath = getChunkFileFolderPath(fileMd5);//获取文件保存路径String chunkFilePath = chunkFileFolderPath + chunk;//获取文件的mimeTypeString mimeType = getMimeType(null);try {//上传分块文件到minioboolean b = updataFile(mimeType, bucket_video, chunkFilePath, localChunkFilePath);if (!b){log.debug("上传分块文件失败:{}",chunkFilePath);return RestResponse.validfail(false,"上传分块文件失败");}}catch (Exception e){e.printStackTrace();}log.debug("上传分块文件成功:{}",chunkFilePath);return RestResponse.success(true);}/*** @description 合并分块* @param companyId  机构id* @param fileMd5  文件md5* @param chunkTotal 分块总和* @param uploadFileParamsDto 文件信息* @return com.xuecheng.base.model.RestResponse* @author Mr.M* @date 2022/9/13 15:56*/@Overridepublic RestResponse mergechunks(Long companyId, String fileMd5, int chunkTotal, UploadFileParamsDto uploadFileParamsDto) {//合并分块//获取分块文件路径
//        fileMd5.substring(0, 1) + "/" + fileMd5.substring(1, 2) + "/" + fileMd5 + "/" + "chunk" + "/";String chunkFileFolderPath = getChunkFileFolderPath(fileMd5);List<ComposeSource> sources = new ArrayList<>();for (int i = 0; i < chunkTotal; i++) {ComposeSource source = ComposeSource.builder().bucket(bucket_video).object(chunkFileFolderPath+i).build();sources.add(source);}//准备合并后文件保存位置//文件名称String fileName = uploadFileParamsDto.getFilename();//文件扩展名String extName = fileName.substring(fileName.lastIndexOf("."));//合并文件路径String mergeFilePath = getFilePathByMd5(fileMd5, extName);ComposeObjectArgs testbucket = ComposeObjectArgs.builder().bucket(bucket_video).object(mergeFilePath).sources(sources).build();try {minioClient.composeObject(testbucket);} catch (Exception e) {log.error("合并文件失败",e);return RestResponse.validfail(false,"合并文件失败");}//校验文件是否完整//下载已经合并完成的文件File file = downloadFileFromMinIO(bucket_video, mergeFilePath);if(file == null){log.debug("下载合并后文件失败,mergeFilePath:{}",mergeFilePath);return RestResponse.validfail(false, "下载合并后文件失败。");}
//        try {
//            //获取合并后文件的md5
//            String mergeFile_md5 = getFileMd5(file.getAbsolutePath());
//            if (!fileMd5.equals(mergeFile_md5)){
//                //合并后文件的md5与源文件的md5不一致
//                return RestResponse.validfail(false,"文件合并校验失败");
//            }
//        } catch (Exception e) {
//            e.printStackTrace();
//        }try (InputStream newFileInputStream = new FileInputStream(file)) {//minio上文件的md5值String md5Hex = DigestUtils.md5DigestAsHex(newFileInputStream);//比较md5值，不一致则说明文件不完整if(!fileMd5.equals(md5Hex)){return RestResponse.validfail(false, "文件合并校验失败，最终上传失败。");}//文件大小uploadFileParamsDto.setFileSize(file.length());}catch (Exception e){log.debug("校验文件失败,fileMd5:{},异常:{}",fileMd5,e.getMessage(),e);return RestResponse.validfail(false, "文件合并校验失败，最终上传失败。");}finally {if(file!=null){file.delete();}}//将文件信息保存到数据库currentProxy.addMediaFilesToDb(companyId,fileMd5,uploadFileParamsDto,bucket_video,mergeFilePath);//删除分块文件
//        removeChunkFiles(chunkFileFolderPath,chunkTotal);clearChunkFiles(chunkFileFolderPath,chunkTotal);return RestResponse.success(true);}/*** 删除分块文件* @param chunkFileFolderPath 分块文件的位置* @param chunkTotal 分块文件的数量*/private void clearChunkFiles(String chunkFileFolderPath, int chunkTotal) {List<DeleteObject> sources = new ArrayList<>();for (int i = 0; i < chunkTotal; i++) {DeleteObject source = new DeleteObject(chunkFileFolderPath.concat(Integer.toString(i)));sources.add(source);}//这个方法要真正删除要遍历一下Iterable<Result<DeleteError>> results = minioClient.removeObjects(RemoveObjectsArgs.builder().bucket(bucket_video).objects(sources).build());results.forEach(r->{DeleteError deleteError = null;try {deleteError = r.get();} catch (Exception e) {e.printStackTrace();log.error("清楚分块文件失败,objectname:{}",deleteError.objectName(),e);}});}/*** 从minio中下载文件* @param bucket* @param objectName* @return*/public File downloadFileFromMinIO(String bucket,String objectName){//创建临时文件File minioFile = null;FileOutputStream  outputStream = null;try {GetObjectArgs testbucket = GetObjectArgs.builder().bucket(bucket).object(objectName).build();//读取数据获取到输入流InputStream  stream = minioClient.getObject(testbucket);//创建临时文件minioFile=File.createTempFile("minio", ".merge");outputStream = new FileOutputStream(minioFile);IOUtils.copy(stream,outputStream);return minioFile;}catch (Exception e){e.printStackTrace();}finally {//关闭资源if (outputStream != null){try {outputStream.close();} catch (IOException e) {throw new RuntimeException(e);}}}return null;}//获取分块路径//得到分块文件的目录private String getChunkFileFolderPath(String fileMd5) {return fileMd5.substring(0, 1) + "/" + fileMd5.substring(1, 2) + "/" + fileMd5 + "/" + "chunk" + "/";}/*** 得到合并后的文件的地址* @param fileMd5 文件id即md5值* @param fileExt 文件扩展名* @return*/private String getFilePathByMd5(String fileMd5,String fileExt){return   fileMd5.substring(0,1) + "/" + fileMd5.substring(1,2) + "/" + fileMd5 + "/" +fileMd5 +fileExt;}/*** @description 将文件信息添加到文件表* @param companyId  机构id* @param fileMd5  文件md5值* @param uploadFileParamsDto  上传文件的信息* @param bucket  桶* @param objectName 对象名称* @return com.xuecheng.media.model.po.MediaFiles* @author Mr.M* @date 2022/10/12 21:22*/@Transactionalpublic MediaFiles addMediaFilesToDb(Long companyId,String fileMd5,UploadFileParamsDto uploadFileParamsDto,String bucket,String objectName){//从数据库查询文件MediaFiles mediaFiles = mediaFilesMapper.selectById(fileMd5);if (mediaFiles == null) {mediaFiles = new MediaFiles();//拷贝基本信息BeanUtils.copyProperties(uploadFileParamsDto, mediaFiles);mediaFiles.setId(fileMd5);mediaFiles.setFileId(fileMd5);mediaFiles.setCompanyId(companyId);mediaFiles.setUrl("/" + bucket + "/" + objectName);mediaFiles.setBucket(bucket);mediaFiles.setFilePath(objectName);mediaFiles.setCreateDate(LocalDateTime.now());mediaFiles.setAuditStatus("002003");mediaFiles.setStatus("1");//保存文件信息到文件表int insert = mediaFilesMapper.insert(mediaFiles);if (insert < 0) {log.error("保存文件信息到数据库失败,{}",mediaFiles.toString());XueChengPlusException.cast("保存文件信息失败");}log.debug("保存文件信息到数据库成功,{}",mediaFiles.toString());}return mediaFiles;}

面试

5.6面试
1、什么情况Spring事务会失效
1)在方法中捕获异常没有抛出去
2)非事务方法调用事务方法
3)事务方法部调用事务方法
4)@Transactional标记的方法不是public
5)抛出的异常与rollbackFor指定的异常不匹配，默认rollbackFor指定的异常为RuntimeException
6)数据库表不支持事务，比如MySQL的MyISAM
7)Spring的传播行为导致事务失效，比如：PROPAGATION_NEVER、PROPAGATION_NOT_SUPPORTED
PROPAGATION REQUIRED-支持当前事务，如果当前没有事务，就新建一个事务。这是最常见的选择。
PROPAGATION SUPPORTS-支持当前事务，如果当前没有事务，就以非事务方式执行。
PROPAGATION MANDATORY-支持当前事务，如果当前没有事务，就抛出异常。
PROPAGATION REQUIRES NEW-新建事务，如果当前存在事务，把当前事务挂起。
PROPAGATION NOT_SUPPORTED-以非事务方式执行操作，如果当前存在事务，就把当前事务挂起。
PROPAGATION_NEVER-以非事务方式执行，如果当前存在事务，则抛出异常。
PROPAGATION NESTED-如果当前存在事务，则在嵌套事务内执行。如果当前没有事务，则与
PROPAGATION REQUIRED类似的操作。

4、断点续传是怎么做的？
我们是基于分块上传的模式实现断点续传的需求，当文件上传一部分断网后前边已经上传过的不再上传。
1)前端对文件分块。
2)前端使用多线程一块一块上传，上传前给服务端发一个消息校验该分块是否上传，如果已上传则不再上传。
3)等所有分块上传完毕，服务端合并所有分块，校验文件的完整性。
因为分块全部上传到了服务器，服务器将所有分块按顺序进行合并，就是写每个分块文件内容按顺序依次写入一个
文件中。使用字节流去读写文件。
4)前端给服务传了一个md5值，服务端合并文件后计算合并后文件的md5是否和前端传的一样，如果一样则说文
件完整，如果不一样说明可能由于网络丢包导致文件不完整，这时上传失败需要重新上传。
5、分块文件清理问题？
上传一个文件进行分块上传，上传一半不传了，之前上传到miio的分块文件要清理吗？怎么做的？
1、在数据库中有一张文件表记录minio中存储的文件信息。
2、文件开始上传时会写入文件表，状态为上传中，上传完成会更新状态为上传完成。
3、当一个文件传了一半不再上传了说明该文件没有上传完成，会有定时任务去查询文件表中的记录，如果文件未
上传完成则删除minio中没有上传成功的文件目录。

视频处理

视频编码

视频上传成功后需要对视频进行转码处理。

什么是视频编码？查阅百度百科如下：

详情参考 ：视频编码_百度百科

首先我们要分清文件格式和编码格式：

文件格式：是指.mp4、.avi、.rmvb等 这些不同扩展名的视频文件的文件格式 ，视频文件的内容主要包括视频和音频，其文件格式是按照一 定的编码格式去编码，并且按照该文件所规定的封装格式将视频、音频、字幕等信息封装在一起，播放器会根据它们的封装格式去提取出编码，然后由播放器解码，最终播放音视频。

音视频编码格式：通过音视频的压缩技术，将视频格式转换成另一种视频格式，通过视频编码实现流媒体的传输。比如：一个.avi的视频文件原来的编码是a，通过编码后编码格式变为b，音频原来为c，通过编码后变为d。

音视频编码格式各类繁多，主要有几下几类：

MPEG系列

（由ISO[国际标准组织机构]下属的MPEG[运动图象专家组]开发 ）视频编码方面主要是Mpeg1（vcd用的就是它）、Mpeg2（DVD使用）、Mpeg4（的DVDRIP使用的都是它的变种，如：divx，xvid等）、Mpeg4 AVC（正热门）；音频编码方面主要是MPEG Audio Layer 1/2、MPEG Audio Layer 3（大名鼎鼎的mp3）、MPEG-2 AAC 、MPEG-4 AAC等等。注意：DVD音频没有采用Mpeg的。

H.26X系列

（由ITU[国际电传视讯联盟]主导，侧重网络传输，注意：只是视频编码）

包括H.261、H.262、H.263、H.263+、H.263++、H.264（就是MPEG4 AVC-合作的结晶）

目前最常用的编码标准是视频H.264，音频AAC。

提问：

H.264是编码格式还是文件格式？

mp4是编码格式还是文件格式？

FFmpeg 的基本使用

我们将视频录制完成后，使用视频编码软件对视频进行编码，本项目 使用FFmpeg对视频进行编码 。

FFmpeg被许多开源项目采用，QQ影音、暴风影音、VLC等。

下载：FFmpeg Download FFmpeg

请从常用工具软件目录找到ffmpeg.exe，并将ffmpeg.exe加入环境变量path中。

测试是否正常：cmd运行 ffmpeg -version

安装成功，作下简单测试

将一个.avi文件转成mp4、mp3、gif等。

比如我们将nacos.avi文件转成mp4，运行如下命令：

D:\soft\ffmpeg\ffmpeg.exe -i 1.avi 1.mp4

可以将ffmpeg.exe配置到环境变量path中，进入视频目录直接运行：ffmpeg.exe -i 1.avi 1.mp4

转成mp3：ffmpeg -i nacos.avi nacos.mp3

转成gif：ffmpeg -i nacos.avi nacos.gif

官方文档（英文）：ffmpeg Documentation

视频处理工具类

将课程资料的工具类中的util拷贝至base工程。

其中Mp4VideoUtil类是用于将视频转为mp4格式，是我们项目要使用的工具类。

对Mp4VideoUtil类需要学习使用方法，下边代码将一个avi视频转为mp4视频，如下：

java">    public static void main(String[] args) throws IOException {//ffmpeg的路径String ffmpeg_path = "D:\\yingyong\\tool\\ffmpeg\\ffmpeg.exe";//ffmpeg的安装位置//源avi视频的路径String video_path = "D:\\test\\1.avi";//转换后mp4文件的名称String mp4_name = "test1.mp4";//转换后mp4文件的路径String mp4_path = "D:\\test\\1a.mp4";//创建工具类对象Mp4VideoUtil videoUtil = new Mp4VideoUtil(ffmpeg_path,video_path,mp4_name,mp4_path);//开始视频转换，成功将返回successString s = videoUtil.generateMp4();System.out.println(s);}

执行main方法，最终在控制台输出 success 表示执行成功。

分布式任务处理

什么是分布式任务调度

对一个视频的转码可以理解为一个任务的执行，如果视频的数量比较多，如何去高效处理一批任务呢？

1、多线程

多线程是充分利用单机的资源。

2、分布式加多线程

充分利用多台计算机，每台计算机使用多线程处理。

方案2可扩展性更强。

方案2是一种分布式任务调度的处理方案。

什么是分布式任务调度？

我们可以先思考一下下面业务场景的解决方案：

每隔24小时执行数据备份任务。

12306网站会根据车次不同，设置几个时间点分批次放票。

某财务系统需要在每天上午10点前结算前一天的账单数据，统计汇总。

商品成功发货后，需要向客户发送短信提醒。

类似的场景还有很多，我们该如何实现？

多线程方式实现：

学过多线程的同学，可能会想到，我们可以开启一个线程，每sleep一段时间，就去检查是否已到预期执行时间。

以下代码简单实现了任务调度的功能：

java">public static void main(String[] args) {    //任务执行间隔时间final long timeInterval = 1000;Runnable runnable = new Runnable() {public void run() {while (true) {//TODO：somethingtry {Thread.sleep(timeInterval);} catch (InterruptedException e) {e.printStackTrace();}}}};Thread thread = new Thread(runnable);thread.start();
}

上面的代码实现了按一定的间隔时间执行任务调度的功能。

Jdk也为我们提供了相关支持，如Timer、ScheduledExecutor，下边我们了解下。

Timer方式实现：

java">public static void main(String[] args){  Timer timer = new Timer();  timer.schedule(new TimerTask(){@Override  public void run() {  //TODO：something}  }, 1000, 2000);  //1秒后开始调度，每2秒执行一次
}

Timer 的优点在于简单易用，每个Timer对应一个线程，因此可以同时启动多个Timer并行执行多个任务，同一个Timer中的任务是串行执行。

ScheduledExecutor方式实现：

java">public static void main(String [] agrs){ScheduledExecutorService service = Executors.newScheduledThreadPool(10);service.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {//TODO：somethingSystem.out.println("todo something");}}, 1,2, TimeUnit.SECONDS);
}

Java 5 推出了基于线程池设计的 ScheduledExecutor，其设计思想是，每一个被调度的任务都会由线程池中一个线程去执行，因此任务是并发执行的，相互之间不会受到干扰。

Timer 和 ScheduledExecutor 都仅能提供基于开始时间与重复间隔的任务调度，不能胜任更加复杂的调度需求。比如，设置每月第一天凌晨1点执行任务、复杂调度任务的管理、任务间传递数据等等。

第三方Quartz方式实现，项目地址：https://github.com/quartz-scheduler/quartz

Quartz 是一个功能强大的任务调度框架，它可以满足更多更复杂的调度需求，Quartz 设计的核心类包括 Scheduler, Job 以及 Trigger。其中，Job 负责定义需要执行的任务，Trigger 负责设置调度策略，Scheduler 将二者组装在一起，并触发任务开始执行。Quartz支持简单的按时间间隔调度、还支持按日历调度方式，通过设置CronTrigger表达式（包括：秒、分、时、日、月、周、年）进行任务调度。

下边是一个例子代码

java">public static void main(String [] agrs) throws SchedulerException {//创建一个SchedulerSchedulerFactory schedulerFactory = new StdSchedulerFactory();Scheduler scheduler = schedulerFactory.getScheduler();//创建JobDetailJobBuilder jobDetailBuilder = JobBuilder.newJob(MyJob.class);jobDetailBuilder.withIdentity("jobName","jobGroupName");JobDetail jobDetail = jobDetailBuilder.build();//创建触发的CronTrigger 支持按日历调度CronTrigger trigger = TriggerBuilder.newTrigger().withIdentity("triggerName", "triggerGroupName").startNow().withSchedule(CronScheduleBuilder.cronSchedule("0/2 * * * * ?")).build();scheduler.scheduleJob(jobDetail,trigger);scheduler.start();
}public class MyJob implements Job {@Overridepublic void execute(JobExecutionContext jobExecutionContext){System.out.println("todo something");}
}

通过以上内容我们学习了什么是任务调度，任务调度所解决的问题，以及任务调度的多种实现方式。

任务调度顾名思义，就是对任务的调度，它是指系统为了完成特定业务，基于给定时间点，给定时间间隔或者给定执行次数自动执行任务。

什么是分布式任务调度？

通常任务调度的程序是集成在应用中的，比如：优惠卷服务中包括了定时发放优惠卷的的调度程序，结算服务中包括了定期生成报表的任务调度程序，由于采用分布式架构，一个服务往往会部署多个冗余实例来运行我们的业务，在这种分布式系统环境下运行任务调度，我们称之为分布式任务调度，如下图：

分布式调度要实现的目标：

不管是任务调度程序集成在应用程序中，还是单独构建的任务调度系统，如果采用分布式调度任务的方式就相当于将任务调度程序分布式构建，这样就可以具有分布式系统的特点，并且提高任务的调度处理能力：

1、并行任务调度

并行任务调度实现靠多线程，如果有大量任务需要调度，此时光靠多线程就会有瓶颈了，因为一台计算机CPU的处理能力是有限的。

如果将任务调度程序分布式部署，每个结点还可以部署为集群，这样就可以让多台计算机共同去完成任务调度，我们可以将任务分割为若干个分片，由不同的实例并行执行，来提高任务调度的处理效率。

2、高可用

若某一个实例宕机，不影响其他实例来执行任务。

3、弹性扩容

当集群中增加实例就可以提高并执行任务的处理效率。

4、任务管理与监测

对系统中存在的所有定时任务进行统一的管理及监测。让开发人员及运维人员能够时刻了解任务执行情况，从而做出快速的应急处理响应。

5、避免任务重复执行

当任务调度以集群方式部署，同一个任务调度可能会执行多次，比如在上面提到的电商系统中到点发优惠券的例子，就会发放多次优惠券，对公司造成很多损失，所以我们需要控制相同的任务在多个运行实例上只执行一次。

XXL-JOB介绍

XXL-JOB配置详情：https://mx67xggunk5.feishu.cn/wiki/SiQAwJ99MiP7w9kZWaicTacKn0g


http://www.ppmy.cn/server/146038.html

相关文章

SpringBoot架构下宠物领养解决方案

SpringBoot架构下宠物领养解决方案

第1章 绪论 1.1 课题背景 二十一世纪互联网的出现&#xff0c;改变了几千年以来人们的生活&#xff0c;不仅仅是生活物资的丰富&#xff0c;还有精神层次的丰富。时代进步的标志&#xff0c;就是让人们过上更好的生活。在互联网诞生之前&#xff0c;地域位置往往是人们思想上不…
阅读更多...
MySQL：常用数据类型

MySQL：常用数据类型

专栏说明&#xff1a;本专栏用于MySQL学习&#xff0c;如对你学习有所帮助&#xff0c;可以点赞收藏关注&#xff0c;感谢三连。 博客主页&#xff1a;Duck Bro 博客主页系列专栏&#xff1a;MySQL数据库关注博主&#xff0c;后期持续更新系列文章如果有错误感谢请大家批评指出…
阅读更多...
ABAP 快速入门之 Hello World 和 ALV 列表

ABAP 快速入门之 Hello World 和 ALV 列表

ABAP 是 SAP 开发的一种编程语言&#xff0c;用于在 SAP 环境中对业务应用程序进行编程。得益于 ABAP 对象增强功能&#xff0c;ABAP 以前只是过程性语言&#xff0c;现在也是一种面向对象语言。 ABAP 750 文档&#xff1a;ABAP Keyword Documentation (sap.com) Hello World…
阅读更多...
4-SpringCloud整合服务间的调用即负载均衡

4-SpringCloud整合服务间的调用即负载均衡

springcloud目录&#xff1a; 1.Spring Cloud简介 2.SpringCloud整合eureka注册中心 3.SpringCloud整合服务注册 4.SpringCloud整合服务间的调用即负载均衡 5.SpringCloud整合Feign调用 6.SpringCloud整合config配置中心 7.SpringCloud整合zuul路由网关 我们复制一个yqx-user服…
阅读更多...
如何有效管理 GitHub API 的请求速率限制并通过认证令牌提高请求额度：完整指南与代码示例

如何有效管理 GitHub API 的请求速率限制并通过认证令牌提高请求额度：完整指南与代码示例

如何有效管理 GitHub API 的请求速率限制并通过认证令牌提高请求额度&#xff1a;完整指南与代码示例 在使用 GitHub API 进行开发时&#xff0c;了解和管理请求速率限制&#xff08;Rate Limiting&#xff09;是至关重要的。GitHub 为了保护其服务器资源&#xff0c;并确保服…
阅读更多...
自动控制原理——BliBli站_DR_CAN

自动控制原理——BliBli站_DR_CAN

自动控制 2 稳定性分析 极点在左半平面 输入为单位冲击&#xff0c;而拉普拉斯变换为1&#xff1b;因此&#xff0c;开环和闭环系统&#xff0c;研究其传递函数的稳定性就可以了 2.5_非零初始条件下的传递函数_含有初始条件的传递函数 如果一个系统的初始条件不为0&#xff0…
阅读更多...
Java知识及热点面试题总结（三）

Java知识及热点面试题总结（三）

1.使用线程池创建线程的优点 降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。提高响应速度。当任务到达时&#xff0c;任务可以不需要的等到线程创建就能立即执行。提高线程的可管理性。线程是稀缺资源&#xff0c;如果无限制的创建&#xff0c;不仅会消…
阅读更多...
scala的守卫语句格式

scala的守卫语句格式

import scala.io.StdIn object test49{//从控制台读入一个数字a,使用&#xff08;StdIn.readInt&#xff09;//如果a>0并且a<3,打印[0-3]//如果a>4并且a<8,打印[4-8]//否则:打印未匹配 // def main(args: Array[String]): Unit { // val aStdIn.readInt()//等…
阅读更多...
最新文章

Copyright @ 2022~2023