[网络协议]springboot（三）websocket实现服务端，客户端的视频传输

1.服务端

1.1pom.xml引入库

 <!--引入websocket依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
        </dependency>

        <dependency>
            <groupId>org.java-websocket</groupId>
            <artifactId>Java-WebSocket</artifactId>
            <version>1.3.5</version>
        </dependency>

1.2新建配置类，?开启WebSocket支持：

发现问题

经过了测试，发现几个问题，
一.连接断开，而且不报任何错
解决方式：是因为缓冲区过小，而传输的视频过大，websocket的默认缓冲区应该是8k，我们传输的视频有10几兆了，所以我们应该去改动缓冲区，查询资料，这里的缓冲区是独立的，有几个websocket连接就有几个缓冲区。

package com.example.demo.config;

import org.springframework.boot.web.servlet.ServletContextInitializer;
import org.springframework.context.annotation.Bean;
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;

/**
 * 开启WebSocket支持
 **/

@Configuration
public class WebSocketConfig implements ServletContextInitializer {
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
    //设置websocket发送内容长度
    @Override
    public void onStartup(ServletContext servletContext) throws ServletException {
        servletContext.addListener(WebAppRootListener.class);
        //这里设置了30兆的缓冲区
        //Tomcat每次请求过来时在创建session时都会把这个webSocketContainer作为参数传进去所以对所有的session都生效了
        servletContext.setInitParameter("org.apache.tomcat.websocket.textBufferSize","30000000");
        servletContext.setInitParameter("org.apache.tomcat.websocket.binaryBufferSize","30000000");
    }
}

1.3新建一个websocket处理类,这里我创建一个类为WebSocketServer

package dzftxt.web.controller;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import dzftxt.data.dataobject.Videos;
import dzftxt.data.form.VideoForm;
import dzftxt.service.SaveFileI;
import dzftxt.service.VideoService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
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;


/**
 * @author ：Siyuan Gao
 * @date ：Created in 2020/9/12 10:24
 * @description：用于接受处理客户端向服务端传递的信息
 * @modified By：
 * @version: $
 */
@Component
@ServerEndpoint("/upload/{sid}")
public class WebSocketUploadServer {

    private static final Logger LOG =  LoggerFactory.getLogger(WebSocketUploadServer.class);
    //用来记录当前在线连接数，
    private static int onlineCount = 0;
    //线程安全的set，用来存放每个客户端对应的websocket对象
    private static CopyOnWriteArraySet<WebSocketUploadServer> webSocketSet = new CopyOnWriteArraySet<>();

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

    private static SaveFileI saveFileI;
    @Autowired
    public void setSaveFileI(SaveFileI saveFileI){
        WebSocketUploadServer.saveFileI=saveFileI;
    }
    //用hashmap存储文件对象和文件路径
    private HashMap docUrl;
    //结束标志
    private String endupload = "over";

    private static VideoService videoService;
    @Autowired
    public void setVideoService(VideoService videoService){
        WebSocketUploadServer.videoService=videoService;
    }
    //连接成功时，服务端会自动调用这个方法
    @OnOpen
    public void onOpen(Session session, @PathParam("sid") String sid){
       //这一句是注入session，否则this无法发送消息
        this.session=session;
        webSocketSet.add(this);
        addOnlineCount();
        LOG.info(sid + "------连接成功---当前在线人数为"+onlineCount);
    }
    /**
     * 连接关闭时调用的方法
     */
    @OnClose
    public void onClose(@PathParam("sid") String sid) {
        //在线人数减1
        subOnlineCount();
        //从set中删除
        webSocketSet.remove(this);
        LOG.info(sid + "已关闭连接" + "----剩余在线人数为：" + onlineCount);
    }
    /*
    下面这个onmessage只负责处理客户端传来的字符串类型的文件，主要用于和前端的配合，
    */
    /*@OnMessage
    public void onMessage(String message, @PathParam("sid") String sid) {
        //把前端传来的json转为对象‘
        JSONObject jsonObject = JSON.parseObject(message);
        //消息类型 比如类型有文件名称filename，传的文件的第几份filecount，是否结束over
        String type = jsonObject.getString("type");
        //得到消息内容
        String data = jsonObject.getString("data");
        //一开始文件名传进来，服务端接受后生成有具体地址的文件对象，并向客户端传递ok进行下一步的传输
        if ("fileName".equals(type)) {
            LOG.info("传输的文件名为：" + data);
            try {
                Map<String, Object> 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();
            }
        }


    }*/
     /*
    下面这个onmessage只负责处理客户端传来的byte[]的文件，主要用于和前端的配合，进行分段传输
    */
     /*@OnMessage
    public void onMessage(byte[] message,Session session){
        //
        try{
            //把传来的字节流数组写入到上面创建的文件对象里去
             saveFileI.saveFileFromBytes(message,docUrl);
           // saveFileI.saveFileFromBytes(message);
            //只有客户端接受到ok才会传输下一段文件
            this.sendMessage("服务端已成功接受视频");
        }catch (IOException e){
            e.printStackTrace();
        }


    }*/
    /*
    * 下面的onmessage用于接受经过json转换的map
    * */
    @OnMessage
    public void onMessage(String message, @PathParam("sid") String sid)  {
        //把前端传来的json转为对象‘
        JSONObject jsonObject = JSON.parseObject(message);
        //消息类型 比如类型有文件名称filename，传的文件的第几份filecount，是否结束over
        String fileName= jsonObject.getString("fileName");
        VideoForm videos=JSON.toJavaObject(jsonObject,VideoForm.class);
        //得到消息内容

        if (videos.getContent()==null) System.out.println("传输的内容为空！");
        Map<String, Object> map = saveFileI.docPath(fileName);
        //
       try {
           saveFileI.saveFileFromBytes(videos.getContent(),map);
           this.sendMessage("服务端已成功接受视频:"+videos.toString());
           //保存视频信息到数据库
           Videos videos1=new Videos();
           BeanUtils.copyProperties(videos,videos1);
           videos1.setPath((String) map.get("path"));
           videoService.saveVideo(videos1);
           this.sendMessage("服务端已成功保存该视频");
           LOG.info(sid + "------视频传输成功---");
       }catch (IOException e){
           e.printStackTrace();
       }


    }
    //这个方法用于接受处理客户端传来的字节流数组类型的文件
    @OnMessage
    public void onMessage(byte[] message,Session session){
        //
        try{
            //把传来的字节流数组写入到上面创建的文件对象里去
          // saveFileI.saveFileFromBytes(message,docUrl);
           saveFileI.saveFileFromBytes(message);
            //只有客户端接受到ok才会传输下一段文件
            this.sendMessage("服务端已成功接受视频");
        }catch (IOException e){
            e.printStackTrace();
        }


    }

    //服务端向客户端发送消息
    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--;
    }
}

?savefilei.java

package dzftxt.service;

import java.util.Map;

public interface SaveFileI {
    /**
     * 生成文件路径
     * @param fileName  接收文件名
     * @return  返回一个map，里面包换文件路径，文件，相对路径
     */
    Map<String,Object> docPath(String fileName);

    /**
     * 将字节流写入文件
     * @param b 字节流数组，是前端传来的。把这个字节流数组添加到map中的file中去（先通过file生成文件流对象outputstream）
     * @param map  文件路径
     * @return  返回是否成功
     */
    boolean saveFileFromBytes(byte[] b, Map<String, Object> map);
    boolean saveFileFromBytes(byte[] b);
}

saveFileImpl.java

package dzftxt.service;

import org.springframework.stereotype.Service;

import javax.websocket.server.ServerEndpoint;
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;

/**
 * @author ：Siyuan Gao
 * @date ：Created in 2020/9/12 10:17
 * @description：
 * @modified By：
 * @version: $
 */
@Service
public class SaveFileImpl implements SaveFileI{

    public Map<String, Object> docPath(String fileName) {
        HashMap<String, Object> 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;
    }


    public boolean saveFileFromBytes(byte[] b, Map<String, Object> 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;
    }
    public boolean saveFileFromBytes(byte[] b) {
        //创建文件流对象
        FileOutputStream fstream = null;
        //从map中获取file对象
        File file = new File("E:\\Users\\Administrator.DESKTOP-K38H7QV\\Desktop\\videos\\test.mp4");
        //判断路径是否存在，不存在就创建
        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;
    }
}

2.客户端

WebSocketConfig.java

package com.nju.config;

import org.java_websocket.drafts.Draft_6455;
import org.java_websocket.handshake.ServerHandshake;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
import org.java_websocket.client.WebSocketClient;


import java.net.URI;

/**
 * @author ：Siyuan Gao
 * @date ：Created in 2020/9/8 14:08
 * @description：websocket的bean
 * @modified By：
 * @version: $
 */
@Component
public class WebSocketConfig {
    @Bean
    public WebSocketClient webSocketClient(){

        try{
            WebSocketClient webSocketClient=new WebSocketClient(new URI("ws://localhost:8090/dzftxt/upload/25477fes1"),new Draft_6455()) {

                @Override
                public void onOpen(ServerHandshake serverHandshake) {
                    System.out.println("客户端建立连接");
                }

                @Override
                public void onMessage(String s) {

                    System.out.println("客户端收到消息----"+s);
                }

                @Override
                public void onClose(int i, String s, boolean b) {
                    System.out.println("客户端关闭连接");
                }

                @Override
                public void onError(Exception e) {

                }
            };
            webSocketClient.connect();
            return webSocketClient;

        }catch (Exception e){
            e.printStackTrace();
        }


        return null;
    }
}

package com.nju.service;

import org.java_websocket.client.WebSocketClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @author ：Siyuan Gao
 * @date ：Created in 2020/9/8 11:19
 * @description：用于websocket传输视频的接口
 * @modified By：
 * @version: $
 */
@Component
public class WebSocketService {
    @Autowired
    private WebSocketClient webSocketClient;
    public void sendMessageToServe(String s){
        webSocketClient.send(s);
    }
    public void sendByteMessage(byte[] bytes){
        webSocketClient.send(bytes);
    }


}

package com.nju.controller;

import com.alibaba.fastjson.JSON;
import com.nju.Utils.FileUtil;
import com.nju.service.WebSocketService;

import com.alibaba.fastjson.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;

import javax.websocket.server.ServerEndpoint;
import java.io.*;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * @author ：Siyuan Gao
 * @date ：Created in 2020/9/9 10:39
 * @description：测试文件上传
 * @modified By：
 * @version: $
 */
@Controller
@ServerEndpoint("/clientWs")
public class UploadFileController {


    private static WebSocketService webSocketService;
    @Autowired
    public void setWebSocketService(WebSocketService webSocketService){
        UploadFileController.webSocketService=webSocketService;
    }
    @RequestMapping("/upload")
    public String uploadFile(){
        return "fileUpload";
    }
    /*
    * 下面这个wsupload用于直接传输byte类型
    * */
    @RequestMapping("/wsUpload")
    public String wsUpload() throws FileNotFoundException {
//        File file=new File("E:\\Users\\Administrator.DESKTOP-K38H7QV\\Desktop\\test.mp4");
        File file=new File("D:\\BaiduYunDownload\\14test.mp4");
        String name=file.getName();
        byte[] bytes = FileUtil.fileToBinArray(file);
        //如果直接传入会不会出问题呢测试一下
        webSocketService.sendByteMessage(bytes);
       /* int start_size=0;
        int end_size=0;
        //当服务端处理完了
        while(end_size<bytes.length){
            byte[] temp= Arrays.copyOfRange(bytes,start_size,end_size);
            webSocketService.sendByteMessage(temp);

        }*/


        System.out.println("客户端发送了视频");
        return "success";
    }
    /*
    * 下面这个wsupload用于测试传输含有byte类型的map转化为string
    * */
    @RequestMapping("/wsUpload1")
    public String wsUpload1() throws FileNotFoundException {
//        File file=new File("E:\\Users\\Administrator.DESKTOP-K38H7QV\\Desktop\\test.mp4");
        File file=new File("D:\\BaiduYunDownload\\02vuetest.avi");
        byte[] bytes = FileUtil.fileToBinArray(file);
        HashMap<String,Object> map=new HashMap();
        map.put("fileName",file.getName());
        map.put("content",bytes);
        map.put("ah","案号test");
        map.put("dzftId","锁id");
        map.put("fydm","法院代码test");
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        map.put("createTime",(df.format(new Date())).toString());

        webSocketService.sendMessageToServe(JSON.toJSONString(map));
       /* int start_size=0;
        int end_size=0;
        //当服务端处理完了
        while(end_size<bytes.length){
            byte[] temp= Arrays.copyOfRange(bytes,start_size,end_size);
            webSocketService.sendByteMessage(temp);

        }*/


        System.out.println("客户端发送了视频");
        return "success";
    }


}

关注博主即可阅读全文