通過fastdfs-java-client的api按塊下載文件,下載成功后寫入到輸出流并將進度按用戶通過websocket推送到客戶端
注:該demo只是單純實現(xiàn)了有進度條的下載,如果下載的接口不做其它處理用戶會卡在下載進程里,后續(xù)再做處理
源碼git地址:https://github.com/xujun738/spring-uploadfile.git這里的代碼已經(jīng)在用戶請求后另起一個線程進行下載處理
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:???????????? #縮略圖生成參數(shù)
??? width: 150
??? height: 150
? pool:
??? jmx-enabled: false
??? max-total: 200
??? max-wait-millis: 30000
? trackerList:??????????? #TrackerList參數(shù),支持多個
??? - 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 {
??? //靜態(tài)變量德澈，用來記錄當前在線連接數(shù)窃爷。應該把它設計成線程安全的屑埋。
??? private static int onlineCount = 0;
??? //concurrent包的線程安全Set爱咬，用來存放每個客戶端對應的MyWebSocket對象筐摘。
??? private static CopyOnWriteArraySet<MyWebSocket> webSocketSet = new CopyOnWriteArraySet<MyWebSocket>();
??? //與某個客戶端的連接會話范抓，需要通過它來給客戶端發(fā)送數(shù)據(jù)
??? private Session session;
??? private String userId = null;
??? /**
???? * 連接建立成功調(diào)用的方法
???? */
??? @OnOpen
??? public void onOpen(Session session, @PathParam("userId") String userId) {
??????? this.session = session;
??????? this.userId = userId;
??????? webSocketSet.add(this);???? //加入set中
??????? addOnlineCount();?????????? //在線數(shù)加1
??????? System.out.println("有新連接加入蘸泻！當前在線人數(shù)為" + getOnlineCount());
??????? try {
??????????? sendMessage("連接成功");
??????? } catch (IOException e) {
??????????? System.out.println("IO異常");
??????? }
??? }
??? /**
???? * 連接關(guān)閉調(diào)用的方法
???? */
??? @OnClose
??? public void onClose() {
??????? webSocketSet.remove(this);? //從set中刪除
??????? subOnlineCount();?????????? //在線數(shù)減1
??????? System.out.println("有一連接關(guān)閉！當前在線人數(shù)為" + getOnlineCount());
??? }
??? /**
???? * 收到客戶端消息后調(diào)用的方法
???? *
???? * @param message 客戶端發(fā)送過來的消息
???? */
??? @OnMessage
??? public void onMessage(String message, Session session) {
??????? System.out.println("來自客戶端的消息:" + message);
??????? //群發(fā)消息
??????? for (MyWebSocket item : webSocketSet) {
??????????? try {
??????????????? item.sendMessage(message);
??????????? } catch (IOException e) {
??????????????? e.printStackTrace();
??????????? }
??????? }
??? }
??? /**
???? * 發(fā)生錯誤時調(diào)用
???? *
???? * @OnError public void onError(Session session, Throwable error) {
???? * System.out.println("發(fā)生錯誤");
???? * error.printStackTrace();
???? * }
???? * <p>
???? * <p>
???? * public void sendMessage(String message) throws IOException {
???? * this.session.getBasicRemote().sendText(message);
???? * //this.session.getAsyncRemote().sendText(message);
???? * }
???? * <p>
???? * <p>
???? * /**
???? * 群發(fā)自定義消息
???? */
??? 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;
??????????? }
??????? }
??? }
??? /**
???? * 實現(xiàn)服務器主動推送
???? */
??? 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')
????}
????//連接發(fā)生錯誤的回調(diào)方法
????websocket.onerror = function(){
????????setMessageInnerHTML("error");
????};
????//連接成功建立的回調(diào)方法
????websocket.onopen = function(event){
????????setMessageInnerHTML("open");
????}
????//接收到消息的回調(diào)方法
????websocket.onmessage = function(event){
????????setMessageInnerHTML(event.data);
????}
????//連接關(guān)閉的回調(diào)方法
????websocket.onclose = function(){
????????setMessageInnerHTML("close");
????}
????//監(jiān)聽窗口關(guān)閉事件莹弊，當窗口關(guān)閉時涤久，主動去關(guān)閉websocket連接，防止連接還沒斷開就關(guān)閉窗口忍弛，server端會拋異常响迂。
????window.onbeforeunload = function(){
????????websocket.close();
????}
????//將消息顯示在網(wǎng)頁上
????function setMessageInnerHTML(innerHTML){
????????document.getElementById('message').innerHTML += innerHTML + '<br/>';
????}
????//關(guān)閉連接
????function closeWebSocket(){
????????websocket.close();
????}
????//發(fā)送消息
????function send(){
????????var message = document.getElementById('text').value;
????????websocket.send(message);
????}
</script>
</html>
