springboot集成websocket实现大文件分块上传

遇到一个上传文件的问题，老大说使用http太慢了，因为http包含大量的请求头，刚好项目本身又集成了websocket，想着就用websocket来做文件上传。

创建springboot项目并集成websocket

先是创建一个spring boot项目

我们勾选了三个依赖
分别是 Lombok，web，websocket

这是当前完整的依赖

<?xml version="1.0" encoding="UTF-8"?>

    4.0.0
    
        org.springframework.boot
        spring-boot-starter-parent
        2.1.2.RELEASE
         
    
    com.ccsert
    websocketupload
    0.0.1-SNAPSHOT
    websocketupload
    websocket Upload project

    
        1.8
    

    
        
            org.springframework.boot
            spring-boot-starter-web
        
        
            org.springframework.boot
            spring-boot-starter-websocket
        

        
            org.projectlombok
            lombok
            true
        
        
            org.springframework.boot
            spring-boot-starter-test
            test
        

        
            com.alibaba
            fastjson
            1.2.58
        
    

    
        
            
                org.springframework.boot
                spring-boot-maven-plugin

在websocketupload包下创建config包，然后建一个配置类WebsocketConfig

然后实现ServletContextInitializer接口
代码如下

package com.ccsert.websocketupload.config;

import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.web.servlet.ServletContextInitializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
import org.springframework.web.util.WebAppRootListener;

import javax.servlet.ServletContext;
import javax.servlet.ServletException;

/**
 * ClassName: WebsocketConfig 

 * Description: 开启websocket支持 

 * date: 2020/2/4 10:58

 *
 * @author ccsert

 * @since JDK 1.8
 */
@Configuration
@ComponentScan
@EnableAutoConfiguration
public class WebsocketConfig implements ServletContextInitializer {
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }

    /**
     * 配置websocket文件接受的文件最大容量
     * @param servletContext    context域对象
     * @throws ServletException 抛出异常
     */
    @Override
    public void onStartup(ServletContext servletContext) throws ServletException {
        servletContext.addListener(WebAppRootListener.class);
        servletContext.setInitParameter("org.apache.tomcat.websocket.textBufferSize","51200000");
        servletContext.setInitParameter("org.apache.tomcat.websocket.binaryBufferSize","51200000");
    }
}

在websocketupload包下创建一个web包然后建立一个WebSocketServer类

我们需要实现它的三个方法：OnOpen，OnClose，OnMessage

他们都会自动调用，类似于事件触发，含义分别是，连接建立成功时调用的方法，连接关闭时调用的方法，最后一个是接收客户端发来的消息

其中OnMessage我们后面会使用多种实现

我们先来玩一个例子，讲websocket自然是喜闻乐见聊天室,贴一下WebSocketServer的代码

package com.ccsert.websocketupload.web;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.websocket.OnClose;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * ClassName: webSocketServer 

 * Description: websocket服务处理类 

 * date: 2020/2/4 11:05

 *
 * @author ccsert

 * @since JDK 1.8
 */
@ServerEndpoint("/websocket/{sid}")
@Component
public class WebSocketServer {

    private static final Logger LOG = LoggerFactory.getLogger(WebSocketServer.class);
    /**
     * 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
     */
    private static int onlineCount = 0;

    /**
     * concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
     */
    private static CopyOnWriteArraySet webSocketSet = new CopyOnWriteArraySet<>();

    /**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    private Session session;

    /**
     * 连接建立成功时调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("sid") String sid) {
        this.session = session;
        //加入set中
        webSocketSet.add(this);
        //在线人数加1
        addOnlineCount();
        LOG.info(sid + "连接成功" + "----当前在线人数为：" + onlineCount);
    }

    /**
     * 连接关闭时调用的方法
     */
    @OnClose
    public void onClose(@PathParam("sid") String sid) {
        //在线人数减1
        subOnlineCount();
        //从set中删除
        webSocketSet.remove(this);
        LOG.info(sid + "已关闭连接" + "----剩余在线人数为：" + onlineCount);
    }

    /**
     * 接收客户端发送的消息时调用的方法
     *
     * @param message 接收的字符串消息
     */
    @OnMessage
    public void onMessage(String message, @PathParam("sid") String sid) {
        LOG.info(sid + "发送消息为:" + message);
        sendInfo(message, sid);
    }

    /**
     * 服务器主动提推送消息
     *
     * @param message 消息内容
     * @throws IOException io异常抛出
     */
    public void sendMessage(String message) throws IOException {

        this.session.getBasicRemote().sendText(message);
    }

    /**
     * 群发消息功能
     *
     * @param message 消息内容
     * @param sid     房间号
     */
    public static void sendInfo(String message, @PathParam("sid") String sid) {
        LOG.info("推送消息到窗口" + sid + "，推送内容:" + message);
        for (WebSocketServer item : webSocketSet) {
            try {
                //这里可以设定只推送给这个sid的，为null则全部推送
                item.sendMessage(message);
            } catch (IOException e) {
                LOG.error("消息发送失败" + e.getMessage(), e);
                return;
            }
        }
    }

    /**
     * 原子性的++操作
     */
    public static synchronized void addOnlineCount() {
        WebSocketServer.onlineCount++;
    }

    /**
     * 原子性的--操作
     */
    public static synchronized void subOnlineCount() {
        WebSocketServer.onlineCount--;
    }
}

这里的代码挺简单没什么好讲的

然后在建立一个html
在static目录下建立一个websocketDemo.html
代码如下




    
    chat room websocket
    
    




    聊天室
    


    ?户姓名  
    

    房间名  
    
    
    


    群发消息

这个也挺简单，这里引用了bootstrap.css样式和jq

然后我们就可以通过这样的一个小demo测试下聊天室的功能了，文末附上代码地址

使用websocket实现文件上传功能

我们仿造刚才的WebSocketServer在写一个websocket类

还是在web包下建立一个类，类名为WebSocketUploadServer

可以将原来WebSocketServer的代码复制过来，然后稍微改造一下，其实我们实现文件上传也可以直接在原来WebSocketServer的代码里直接实现，但是现在为了让代码更清晰一些，我们先将就一下

原本聊天室的情况下，一个房间里是可以有多个客户端连接的，但是文件上传我们是不允许的，假如有多个人在同一个房间，那么消息就会传到每个客户端，因为我们要做分块上传，所以这里控制每个房间只能有一个人

我们把websocketDemo.html也复制一份，改名就叫uploadFileDemo.html

我们原先的房间号是手动输入的，现在，我们保证每次都是不同的房间号所以，这次房间号就用随机数

当然方法有很多种，我只是提供一种简单的实现方式，不同的业务场景当然也需要不同的实现方式。不用太过死板

后台的核心代码主要是接收字节流和json消息，我们将字符串消息格式化成了json消息

下面是接收字符串的OnMessage

	/**
     * 接收客户端发送的消息时调用的方法
     *
     * @param message 接收的字符串消息。该消息应当为json字符串
     */
    @OnMessage
    public void onMessage(String message, @PathParam("sid") String sid) {
        //前端传过来的消息都是一个json
        JSONObject jsonObject = JSON.parseObject(message);
        //消息类型
        String type = jsonObject.getString("type");
        //消息内容
        String data = jsonObject.getString("data");
        //判断类型是否为文件名
        if ("fileName".equals(type)) {
            LOG.info("传输文件为:" + data);
            //此处的 “.”需要进行转义
            /*String[] split = data.split("\\.");*/
            try {
                Map map = saveFileI.docPath(data);
                docUrl = (HashMap) map;
                this.sendMessage("ok");
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        else if ("fileCount".equals(type)){
            LOG.info("传输第"+data+"份");
        }
        //判断是否结束
        else if (endupload.equals(type)) {
            LOG.info("===============>传输成功");
            //返回一个文件下载地址
            String path = (String) docUrl.get("nginxPath");
            //返回客户端文件地址
            try {
                this.sendMessage(path);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

然后是接收字节流的

    /**
     * 该方法用于接收字节流数组
     *
     * @param message 文件字节流数组
     * @param session 会话
     */
    @OnMessage
    public void onMessage(byte[] message, Session session) {
        //群发消息
        try {
            //将流写入文件
            saveFileI.saveFileFromBytes(message,docUrl);
            //文件写入成功，返回一个ok
            this.sendMessage("ok");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

单独看肯定是看不懂的
业务逻辑我们放在了service层进行处理

在websocketupload包下建立service包
然后建立SaveFileI接口，接口里有两个方法，一个是创建文件路径的，一个是将流数据写入文件的

package com.ccsert.websocketupload.service;

import java.util.Map;

/**
 * ClassName: saveFileI 

 * Description: 保存文件接口 

 * date: 2020/2/4 17:39

 *
 * @author ccsert

 * @since JDK 1.8
 */
public interface SaveFileI {
    /**
     * 生成文件路径
     * @param fileName  接收文件名
     * @return  返回文件路径
     */
    Map docPath(String fileName);

    /**
     * 将字节流写入文件
     * @param b 字节流数组
     * @param map  文件路径
     * @return  返回是否成功
     */
    boolean saveFileFromBytes(byte[] b, Map map);
}

具体的实现我直接贴出来

package com.ccsert.websocketupload.service.impl;

import com.ccsert.websocketupload.service.SaveFileI;
import org.springframework.stereotype.Service;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
 * ClassName: SaveFileImpl 

 * Description: 

 * date: 2020/2/4 19:01

 *
 * @author ccsert

 * @since JDK 1.8
 */
@Service
public class SaveFileImpl implements SaveFileI {
    @Override
    public Map docPath(String fileName) {
        HashMap map = new HashMap<>();
        //根据时间生成文件夹路径
        Date date = new Date();
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy/MM/dd");
        String docUrl = simpleDateFormat.format(date);
        //文件保存地址
        String path = "/data/images/" + docUrl;
        //创建文件
        File dest = new File(path+"/" + fileName);
        //如果文件已经存在就先删除掉
        if (dest.getParentFile().exists()) {
            dest.delete();
        }
        map.put("dest", dest);
        map.put("path", path+"/" + fileName);
        map.put("nginxPath","/"+docUrl+"/"+fileName);
        return map;
    }

    @Override
    public boolean saveFileFromBytes(byte[] b, Map map) {
        //创建文件流对象
        FileOutputStream fstream = null;
        //从map中获取file对象
        File file = (File) map.get("dest");
        //判断路径是否存在，不存在就创建
        if (!file.getParentFile().exists()) {
            file.getParentFile().mkdirs();
        }
        try {
            fstream = new FileOutputStream(file, true);
            fstream.write(b);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        } finally {
            if (fstream != null) {
                try {
                    fstream.close();
                } catch (IOException e1) {
                    e1.printStackTrace();
                }
            }
        }
        return true;
    }
}

我们在websocket服务里注入接口的时候要注意一点，因为spring是单例的，websocket在初始化的时候就实例化了spring的bean，但是当websocket创建一个新的连接的时候spring的bean会出现null的问题，也就是它只注入了一次。这里我们这样注入可以解决这个问题。

    /**
     * 注入文件保存的接口
     */
    private static SaveFileI saveFileI;

    @Autowired
    public void setSaveFileI(SaveFileI saveFileI) {
        WebSocketUploadServer.saveFileI = saveFileI;
    }

我们贴一下WebSocketUploadServer完整的代码

package com.ccsert.websocketupload.web;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ccsert.websocketupload.service.SaveFileI;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.websocket.OnClose;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * ClassName: WebSocketUploadServer 

 * Description: 

 * date: 2020/2/4 14:51

 *
 * @author ccsert

 * @since JDK 1.8
 */
@ServerEndpoint("/upload/{sid}")
@Component
public class WebSocketUploadServer {
    private static final Logger LOG = LoggerFactory.getLogger(WebSocketUploadServer.class);

    /**
     * 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
     */
    private static int onlineCount = 0;

    /**
     * concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
     */
    private static CopyOnWriteArraySet webSocketSet = new CopyOnWriteArraySet<>();

    /**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    private Session session;

    /**
     * 注入文件保存的接口
     */
    private static SaveFileI saveFileI;

    @Autowired
    public void setSaveFileI(SaveFileI saveFileI) {
        WebSocketUploadServer.saveFileI = saveFileI;
    }

    /**
     * 保证文件对象和文件路径的唯一性
     */
    private HashMap docUrl;

    /**
     * 结束标识判断
     */
    private String endupload = "over";

    /**
     * 连接建立成功时调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("sid") String sid) {
        this.session = session;
        //加入set中
        webSocketSet.add(this);
        //在线人数加1
        addOnlineCount();
        LOG.info(sid + "连接成功" + "----当前在线人数为：" + onlineCount);
    }

    /**
     * 连接关闭时调用的方法
     */
    @OnClose
    public void onClose(@PathParam("sid") String sid) {
        //在线人数减1
        subOnlineCount();
        //从set中删除
        webSocketSet.remove(this);
        LOG.info(sid + "已关闭连接" + "----剩余在线人数为：" + onlineCount);
    }

    /**
     * 接收客户端发送的消息时调用的方法
     *
     * @param message 接收的字符串消息。该消息应当为json字符串
     */
    @OnMessage
    public void onMessage(String message, @PathParam("sid") String sid) {
        //前端传过来的消息都是一个json
        JSONObject jsonObject = JSON.parseObject(message);
        //消息类型
        String type = jsonObject.getString("type");
        //消息内容
        String data = jsonObject.getString("data");
        //判断类型是否为文件名
        if ("fileName".equals(type)) {
            LOG.info("传输文件为:" + data);
            //此处的 “.”需要进行转义
            /*String[] split = data.split("\\.");*/
            try {
                Map map = saveFileI.docPath(data);
                docUrl = (HashMap) map;
                this.sendMessage("ok");
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        else if ("fileCount".equals(type)){
            LOG.info("传输第"+data+"份");
        }
        //判断是否结束
        else if (endupload.equals(type)) {
            LOG.info("===============>传输成功");
            //返回一个文件下载地址
            String path = (String) docUrl.get("nginxPath");
            //返回客户端文件地址
            try {
                this.sendMessage(path);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 该方法用于接收字节流数组
     *
     * @param message 文件字节流数组
     * @param session 会话
     */
    @OnMessage
    public void onMessage(byte[] message, Session session) {
        //群发消息
        try {
            //将流写入文件
            saveFileI.saveFileFromBytes(message,docUrl);
            //文件写入成功，返回一个ok
            this.sendMessage("ok");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 服务器主动提推送消息
     *
     * @param message 消息内容
     * @throws IOException io异常抛出
     */
    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }

    /**
     * 群发消息功能
     *
     * @param message 消息内容
     * @param sid     房间号
     */
    public static void sendInfo(String message, @PathParam("sid") String sid) {
        LOG.info("推送消息到窗口" + sid + "，推送内容:" + message);
        for (WebSocketUploadServer item : webSocketSet) {
            try {
                //这里可以设定只推送给这个sid的，为null则全部推送
                item.sendMessage(message);
            } catch (IOException e) {
                LOG.error("消息发送失败" + e.getMessage(), e);
                return;
            }
        }
    }

    /**
     * 原子性的++操作
     */
    public static synchronized void addOnlineCount() {
        WebSocketUploadServer.onlineCount++;
    }

    /**
     * 原子性的--操作
     */
    public static synchronized void subOnlineCount() {
        WebSocketUploadServer.onlineCount--;
    }

}

接着就是前端的一些处理了

前端我使用了很多打标记的思想

我就直接贴代码了




    
    chat room websocket
    
    




    选择文件
         
    
    
        上传
    


    
    




    传输信息

加了点样式加了个进度条，大部分代码都有注释所以也不多做解释了

最后返回的这个文件地址，你可以在后台存放到文件服务器后返回给前端展示，这里我就不做过多的操作了，我没加网络地址那么文件就存在项目所在盘符的根路径下的日期格式目录下。

最后附上源码
GitHub地址：
git@github.com:ccsert/websocketUpload.git
gitee地址：
https://gitee.com/ccsert/websocketUpload.git

WebSocket springboot

springboot集成websocket实现大文件分块上传

创建springboot项目并集成websocket

聊天室

使用websocket实现文件上传功能

传输信息

相关

微信小程序+SpringBoot(Shiro)登录后，调用接口还是跑到了login。

eclipse搭建springboot开发环境

springboot学习之路31（整合minio服务）

Springboot配置tk-mybatis

springboot配置信息

nginx反向代理websocket wss或ws端口

springboot配置国际化功能

SpringBoot AOP+自定义注解实现日志功能

SpringBoot学习- 1、SpringSuit创建项目

SpringBoot学习- 3、整合MyBatis

SpringBoot+Mybatis加载Mapper.xml文件的两种方式

初学SpringBoot

标签