通过fastdfs-java-client的api按块下载文件,下载成功后写入到输出流并将进度按用户通过websocket推送到客户端
注:该demo只是单纯实现了有进度条的下载,如果下载的接口不做其它处理用户会卡在下载进程里,后续再做处理
源码git地址:https://github.com/xujun738/spring-uploadfile.git这里的代码已经在用户请求后另起一个线程进行下载处理
1.pom.xml
<!—fastdfs-client—>
<dependency>
<groupId>com.github.tobato</groupId>
<artifactId>fastdfs-client</artifactId>
<version>1.26.5</version>
</dependency>
<!—springboot websocket—>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
springboot集成fastdfs
配置application.yml
fdfs:
so-timeout: 600000
connectTimeout: 600
thumbImage: #缩略图生成参数
width: 150
height: 150
pool:
jmx-enabled: false
max-total: 200
max-wait-millis: 30000
trackerList: #TrackerList参数,支持多个
- ip:22122 # 107.182.180.143:22122
visit:
url:http://ip #47.100.116.208
port: 8888
Websocket配置类
@Configuration
public class WebSocketConfig {
@Bean
public ServerEndpointExporter serverEndpointExporter() {
return new ServerEndpointExporter();
}
}
websocket服务类
/**
* <p>Description : </p>
* <p>Copyright : Copyright (c) 2018</p>
* <p>Company : tgram </p>
*
* @author eric
* @version 1.0
* @Date 2019/3/8 上午9:57
*/
@ServerEndpoint("/websocket/{userId}")
@Component
public class MyWebSocket {
//静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
private static int onlineCount = 0;
//concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
private static CopyOnWriteArraySet<MyWebSocket> webSocketSet = new CopyOnWriteArraySet<MyWebSocket>();
//与某个客户端的连接会话，需要通过它来给客户端发送数据
private Session session;
private String userId = null;
/**
* 连接建立成功调用的方法
*/
@OnOpen
public void onOpen(Session session, @PathParam("userId") String userId) {
this.session = session;
this.userId = userId;
webSocketSet.add(this); //加入set中
addOnlineCount(); //在线数加1
System.out.println("有新连接加入！当前在线人数为" + getOnlineCount());
try {
sendMessage("连接成功");
} catch (IOException e) {
System.out.println("IO异常");
}
}
/**
* 连接关闭调用的方法
*/
@OnClose
public void onClose() {
webSocketSet.remove(this); //从set中删除
subOnlineCount(); //在线数减1
System.out.println("有一连接关闭！当前在线人数为" + getOnlineCount());
}
/**
* 收到客户端消息后调用的方法
*
* @param message 客户端发送过来的消息
*/
@OnMessage
public void onMessage(String message, Session session) {
System.out.println("来自客户端的消息:" + message);
//群发消息
for (MyWebSocket item : webSocketSet) {
try {
item.sendMessage(message);
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* 发生错误时调用
*
* @OnError public void onError(Session session, Throwable error) {
* System.out.println("发生错误");
* error.printStackTrace();
* }
* <p>
* <p>
* public void sendMessage(String message) throws IOException {
* this.session.getBasicRemote().sendText(message);
* //this.session.getAsyncRemote().sendText(message);
* }
* <p>
* <p>
* /**
* 群发自定义消息
*/
public static void sendInfo(String message, @PathParam("userId") String userId) throws IOException {
for (MyWebSocket item : webSocketSet) {
try {
//这里可以设定只推送给这个userId的，为null则全部推送
if (userId == null) {
item.sendMessage(message);
} else if (item.userId.equals(userId)) {
item.sendMessage(message);
}
} catch (IOException e) {
continue;
}
}
}
/**
* 实现服务器主动推送
*/
public void sendMessage(String message) throws IOException {
this.session.getBasicRemote().sendText(message);
}
public static synchronized int getOnlineCount() {
return onlineCount;
}
public static synchronized void addOnlineCount() {
MyWebSocket.onlineCount++;
}
public static synchronized void subOnlineCount() {
MyWebSocket.onlineCount--;
}
}
5.控制层类
@Controller
@RequestMapping("/upload")
public class UploadCtrl {
@Autowired
private FastDFSClientWrapper fastDFSClientWrapper;
@Autowired
private FastFileStorageClient storageClient;
@RequestMapping(value = "", method = RequestMethod.POST)
@ResponseBody
public InfoMsg fileUpload(@RequestParam("uploadFile") MultipartFile file) {
InfoMsg infoMsg = new InfoMsg();
if (file.isEmpty()) {
infoMsg.setCode("error");
infoMsg.setMsg("Please select a file to upload");
return infoMsg;
}
try {
String url = fastDFSClientWrapper.uploadFile(file);
System.out.println("上传的文件URL: " + url);
JSONObject jsonObject = new JSONObject();
jsonObject.put("url", url);
jsonObject.put("filesize", file.getSize());
// File tmp = new File(TMP_PATH, file.getOriginalFilename());
// if(!tmp.getParentFile().exists()){
// tmp.getParentFile().mkdirs();
// }
// file.transferTo(tmp);
infoMsg.setCode("success");
infoMsg.setMsg("You successfully uploaded '" + file.getOriginalFilename() + "'");
} catch (IOException e) {
infoMsg.setCode("error");
infoMsg.setMsg("Uploaded file failed");
}
return infoMsg;
}
@RequestMapping(value = "/delete", method = RequestMethod.POST, produces = "application/json;charset=UTF-8")
@ResponseBody
public String deleteFile(String fileName) {
try {
fastDFSClientWrapper.deleteFile(fileName);
return "删除成功";
} catch (Exception e) {
e.printStackTrace();
return "删除失败";
}
}
@RequestMapping(value = "/download")
@ResponseBody
public void download(String fileName, HttpServletResponse response, String userId) throws IOException {
StorePath storePath = StorePath.parseFromUrl(fileName);
// 配置文件下载
response.setHeader("content-type", "application/octet-stream");
response.setContentType("application/octet-stream");
// 下载文件能正常显示中文
response.setHeader("Content-Disposition", "attachment;filename=" + URLEncoder.encode(fileName, "UTF-8"));
// byte[] r = storageClient.downloadFile(storePath.getGroup(), storePath.getPath(), new DownloadCallback<byte[]>() {
// @Override
// public byte[] recv(InputStream ins) throws IOException {
// byte[] reulst = IOUtils.toByteArray(ins);
// System.out.println(reulst.length);
// return reulst;
// }
// });
// response.getOutputStream().write(r);
FileInfo fileInfo = storageClient.queryFileInfo(storePath.getGroup(), storePath.getPath());
long fileSize = fileInfo.getFileSize();
System.out.println("文件总大小:" + fileSize);
long slice = Math.floorDiv(fileSize, 100);
long left = fileSize - slice * 99;
byte[] sliceBytes = null;
int downloadBytes = 0;
ByteBuffer bb = new ByteBuffer();
for (int i = 0; i < 100; i++) {
if (i != 99) {
sliceBytes = storageClient.downloadFile(storePath.getGroup(), storePath.getPath(), i * slice, slice, ins -> {
byte[] result = IOUtils.toByteArray(ins);
response.getOutputStream().write(result);
return result;
});
} else {
sliceBytes = storageClient.downloadFile(storePath.getGroup(), storePath.getPath(), 99 * slice, left, ins -> {
byte[] result = IOUtils.toByteArray(ins);
response.getOutputStream().write(result);
return result;
});
}
downloadBytes = downloadBytes + sliceBytes.length;
MyWebSocket.sendInfo((i + 1) + "", userId);
}
response.getOutputStream().flush();
// 新起一个线程,然后按段下载文件,每段下载成功后将进度值推送到对应的用户
System.out.println("共下载:" + downloadBytes);
}
}
服务启动类
@SpringBootApplication
@Import(FdfsClientConfig.class)
@EnableAutoConfiguration
public class StudyApplication {
public static void main(String[] args) {
new SpringApplication(StudyApplication.class).run(args);
}
}
7.html页面
<!DOCTYPE HTML>
<html>
<head>
<title>My WebSocket</title>
</head>
<body>
Welcome<br/>
<input id="text" type="text" /><button onclick="send()">Send</button> <button onclick="closeWebSocket()">Close</button>
<div id="message">
</div>
</body>
<script type="text/javascript">
var websocket = null;
//判断当前浏览器是否支持WebSocket
if('WebSocket' in window){
websocket = new WebSocket("ws://localhost:8080/websocket/admin1");
}
else{
alert('Not support websocket')
}
//连接发生错误的回调方法
websocket.onerror = function(){
setMessageInnerHTML("error");
};
//连接成功建立的回调方法
websocket.onopen = function(event){
setMessageInnerHTML("open");
}
//接收到消息的回调方法
websocket.onmessage = function(event){
setMessageInnerHTML(event.data);
}
//连接关闭的回调方法
websocket.onclose = function(){
setMessageInnerHTML("close");
}
//监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。
window.onbeforeunload = function(){
websocket.close();
}
//将消息显示在网页上
function setMessageInnerHTML(innerHTML){
document.getElementById('message').innerHTML += innerHTML + '<br/>';
}
//关闭连接
function closeWebSocket(){
websocket.close();
}
//发送消息
function send(){
var message = document.getElementById('text').value;
websocket.send(message);
}
</script>
</html>
