socket集成springboot项目

引入依赖：

        <dependency>
            <groupId>com.corundumstudio.socketio</groupId>
            <artifactId>netty-socketio</artifactId>
            <version>1.7.7</version>
        </dependency>

配置类：

package cn.netinnet.cbt.config;

import com.corundumstudio.socketio.SocketConfig;
import com.corundumstudio.socketio.SocketIOServer;
import com.corundumstudio.socketio.Transport;
import com.corundumstudio.socketio.annotation.SpringAnnotationScanner;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class SocketIOConfig {

    @Value("${socketio.host:172.30.6.68}")
    private String host;

    @Value("${socketio.port:9098}")
    private Integer port;

    @Value("${socketio.bossCount:1}")
    private int bossCount;

    @Value("${socketio.workCount:100}")
    private int workCount;

    @Value("${socketio.allowCustomRequests:true}")
    private boolean allowCustomRequests;

    @Value("${socketio.upgradeTimeout:1000000}")
    private int upgradeTimeout;

    @Value("${socketio.pingTimeout:6000000}")
    private int pingTimeout;

    @Value("${socketio.pingInterval:25000}")
    private int pingInterval;

    @Value("${socketio.maxFramePayloadLength:1048576}")
    private int maxFramePayloadLength;

    /**
     * 以下配置在上面的application.properties中已经注明
     * @return
     */
    @Bean
    public SocketIOServer socketIOServer() {
        SocketConfig socketConfig = new SocketConfig();
        socketConfig.setTcpNoDelay(true);
        socketConfig.setSoLinger(0);
        com.corundumstudio.socketio.Configuration config = new com.corundumstudio.socketio.Configuration();
        config.setSocketConfig(socketConfig);
        config.setHostname(host);
        config.setPort(port);
        config.setBossThreads(bossCount);
        config.setWorkerThreads(workCount);
        config.setAllowCustomRequests(allowCustomRequests);
        config.setUpgradeTimeout(upgradeTimeout);
        config.setPingTimeout(pingTimeout);
        config.setPingInterval(pingInterval);
        //设置最大每帧处理数据的长度，防止他人利用大数据来攻击服务器
        config.setMaxFramePayloadLength(maxFramePayloadLength);
        config.setTransports(Transport.POLLING, Transport.WEBSOCKET);
        return new SocketIOServer(config);
    }

    /**
     * 开启SocketIOServer注解支持
     */
    @Bean
    public SpringAnnotationScanner springAnnotationScanner(SocketIOServer socketServer) {
        return new SpringAnnotationScanner(socketServer);
    }
}

业务接口类：

public interface WebSocketService{
    void receiveMessage(Message message, MultipartFile file);

}

实现类：

@Service
public class WebSocketServiceImpl implements WebSocketService {
    @Autowired
    private SocketIOService socketIOService;

    @Autowired
    private NinCbtStuScoreService ninCbtStuScoreService;

    @Override
    public void receiveMessage(Message message, MultipartFile file) {
        StuScoreBO bo = new StuScoreBO();
        BeanUtils.copyProperties(message,bo);
        // 直播结束的时候
        if (BusinessConstant.LIVE_ROOM_END == message.getMessageType()) {
            if (null == file) {
                throw new BaseServiceException(BusinessConstant.SERVICE_ERRO_CODE, BusinessConstant.FILE_EPT);
            }
            String fileUrl = ninCbtStuScoreService.uploadStuFileForApp(bo, file);
            bo.setFileUrl(fileUrl);
        }
        ninCbtStuScoreService.initStuScore(bo,message);
        socketIOService.pushMessageToUser(message);
    }

}

socket逻辑接口：

package cn.netinnet.cbt.service;


import cn.netinnet.cbt.bo.Message;
import cn.netinnet.cbt.constant.WsMsgTypeEnum;

public interface SocketIOService {

    //推送的事件
    String PUSH_EVENT = "push_event";

    String CONNECT_EVENT = "connect_event";

    // 启动服务
    void start() throws Exception;

    // 停止服务
    void stop();

    // 推送信息
    void pushMessageToUser(Message message);

}

实现类：

package cn.netinnet.cbt.service.impl;

import cn.netinnet.cbt.bo.Message;
import cn.netinnet.cbt.service.SocketIOService;
import com.corundumstudio.socketio.SocketIOClient;
import com.corundumstudio.socketio.SocketIOServer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Service
public class SocketIOServiceImpl implements SocketIOService {

    // 用来存已连接的客户端
    private static Map<String, SocketIOClient> clientMap = new ConcurrentHashMap<>();

    @Autowired
    private SocketIOServer socketIOServer;


    /**
     * Spring IoC容器创建之后，在加载SocketIOServiceImpl Bean之后启动
     * @throws Exception
     */
    @PostConstruct
    private void autoStartup() throws Exception {
        start();
    }

    /**
     * Spring IoC容器在销毁SocketIOServiceImpl Bean之前关闭,避免重启项目服务端口占用问题
     * @throws Exception
     */
    @PreDestroy
    private void autoStop() throws Exception  {
        stop();
    }

    @Override
    public void start() {
        // 监听客户端连接
        socketIOServer.addConnectListener(client -> {
        });

        // 监听客户端断开连接
        socketIOServer.addDisconnectListener(client -> {
            client.disconnect();
        });

        // 处理自定义的事件，与连接监听类似， 连接加入批次房间
        socketIOServer.addEventListener(CONNECT_EVENT, Message.class, (client, data, ackSender) -> {
            client.joinRoom("ec_cbt_room_" + data.getTeaClassSessionId() +"_"+ data.getUserId());
        });
        socketIOServer.addEventListener(PUSH_EVENT, Message.class, (client, data, ackSender) -> {
            socketIOServer.getRoomOperations("ec_cbt_room_" + data.getTeaClassSessionId() +"_"+ data.getUserId()).sendEvent(PUSH_EVENT, "现在是这个ip连着" + data);
        });
        socketIOServer.start();
    }

    @Override
    public void stop() {
        if (socketIOServer != null) {
            socketIOServer.stop();
            socketIOServer = null;
        }
    }

    /**
     * 推送信息
     * @param message
     */
    @Override
    public void pushMessageToUser(Message message) {
        socketIOServer.getRoomOperations("ec_cbt_room_" + message.getTeaClassSessionId() +"_"+ message.getUserId()).sendEvent(PUSH_EVENT, message);
    }


    /**
     * 此方法为获取client连接中的参数，可根据需求更改
     * @param client
     * @return
     */
    private String getParamsByClient(SocketIOClient client) {
        // 从请求的连接中拿出参数（这里的loginUserNum必须是唯一标识）
        Map<String, List<String>> params = client.getHandshakeData().getUrlParams();
            List<String> list = params.get("batchId");
        if (list != null && list.size() > 0) {
            return list.get(0);
        }
        return null;
    }
}

前端仿客户端：

<html>
<head>
    <meta charset="utf-8"/>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/2.1.0/socket.io.js"></script>
<!--    <script src="socket.io.js"></script>-->
    <script>

        var socket = null;
        var i = 0;
        window.onload = function(){
            initSocket();
        }
        function getCookie(c_name)
        {
            if (document.cookie.length>0)
            {
                c_start=document.cookie.indexOf(c_name + "=")
                if (c_start!=-1)
                {
                    c_start=c_start + c_name.length+1
                    c_end=document.cookie.indexOf(";",c_start)
                    if (c_end==-1) c_end=document.cookie.length
                    return unescape(document.cookie.substring(c_start,c_end))
                }
            }
            return ""
        }
        function initSocket(){
            socket = io.connect('http://172.30.6.68:9098', { "transports":["websocket"]} ); //正式发布环境
            // socket = io('http://localhost:8974/test'); //正式发布环境
            socket.on('connect', function () {
				socket.emit("connect_event", {"teaClassSessionId":1111,"userId":880785304192290816});
                console.log('socket连接成功');

            });

            socket.on('disconnect', function () {
                console.log('socket断开连接');
            });


            //==============以下使用命名空间test========================

            //监听广播消息
            socket.on('testNamespace', function (data) {
                console.log("接收到消息:" + data);
            });

            //监听点对点消息
            socket.on('bbbb', function (data) {
                //....收到消息后具体操作
                //  console.log(data);
                console.log(data);
            });

            socket.on('push_event', function (data) {
                //....收到消息后具体操作
                //  console.log(data);
                console.log(data);
            });

            //监听后端无限推送的点对点消息
            socket.on('testPush', function (data) {
                console.log("接收到消息的次数:" + ++i);
            });

            //监听加入房间的反馈
            socket.on('testJoinRoom', function (data) {
                console.log("接收到消息:" + data);
            });

            //监听房间消息
            socket.on('testRoom', function (data) {
                console.log("接收到消息:" + data);
            });
            //监听广播消息
            socket.on('testBroadcast', function (data) {
                console.log("接收到消息:" + data);
            });

        }

        //发送点对点消息
        function send(){
            socket.emit('push_event', {"teaClassId":1111,"userId":3333,"content":"直播结束"});
        }
        //触发无限推送
        function send2(){
            socket.emit('testPush', "begin");
        }
        //发送加入房间消息
        function send3(){
            socket.emit('joinRoom', "room1");
        }
        //发送房间消息
        function send4(){
            socket.emit('testRoom', "testRoomData");
        }
        //发送广播消息
        function send5(){
            socket.emit('testBroadcast', "testBroadCastData");
        }


    </script>

    <input type="button" value="发送点对点消息" onclick="send();">
    <br/><br/>
    <input type="button" value="开启无限推送测试" onclick="send2();">
    <br/><br/>
    <input type="button" value="测试加入房间" onclick="send3();">
    <br/><br/>
    <input type="button" value="测试房间内发消息" onclick="send4();">
    <br/><br/>
    <input type="button" value="测试发送广播消息" onclick="send5();">
    <br/><br/>

</head>
</html>