devbox
寸_铁
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Springboot 、Netty-SocketIO、Redission 集群实现web消息通讯

作者：寸_铁 | 2024-08-13 08:06:53

赞

踩

netty-socketio

说明

最近因需要研究了一下消息推送，也是在网上参考其他大佬的文章以后进行的尝试，特此记录探索的小实践，也是为了后面需要时参考参考， 如果大家发现不妥之处欢迎指正。

目标

  • 需要负载均衡，部署多服务，业务场景触发处理消息提醒，实现业务端出发消息推送
    1. 后台给指定用户发送消息
    2. 后台给指定群体进行发送消息

关于

  • 使用默认namespace，后面有时间再研究自定义namespace
  • 关于room, 目前打算将用户id作为room进行管理，方便做点对点消息推送
    1. 在创建连接时，创建用户id房间并加入
    2. 如果有群发的业务场景，如给某个部门发消息，可以继续加入部门编号命名的房间

实现方案介绍

以2台机器部署服务端为例

一、 服务端

1. 技术框架

  • Springboot
  • Redission（实现集群）
  • Netty-SocketIO

2. 代码实现

2.1 maven配置
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>websocket-server</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.4.2</version>
    </parent>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>2.0.25</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.26</version>
        </dependency>
        <dependency>
            <groupId>com.corundumstudio.socketio</groupId>
            <artifactId>netty-socketio</artifactId>
            <version>2.0.2</version>
        </dependency>
        <dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson</artifactId>
            <version>3.21.3</version>
        </dependency>
   </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
2.2 配置文件
  • application.yml
server:
  port: 8080
websocket:
  socket-port: 8090
  maxFramePayloadLength: 1048576
  maxHttpContentLength: 1048576
  upgradeTimeout: 1000000
  pingTimeout: 6000000
  pingInterval: 25000

redisson:
  address: redis://127.0.0.1:6379
  password:
  database: 0

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • application-s1.yml
server:
  port: 8080
websocket:
  socket-port: 8090

  • 1
  • 2
  • 3
  • 4
  • application-s2.yml
server:
  port: 9080
websocket:
  socket-port: 9090

  • 1
  • 2
  • 3
  • 4
2.3 启动调试
  • IDEA 在启动时添加启动参数 --spring.profiles.active=s1 或--spring.profiles.active=s1
2.4 关键代码
  • springboot 跨域配置 CorsConfig.java，允许跨域，可以自行指定跨域路径，socketIO, 路径前缀为/socket.io
package com.ddw.chat.config;

import org.springframework.context.annotation.Configuration;
import org.springframework.web.servlet.config.annotation.CorsRegistry;
import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;

@Configuration
public class CorsConfig implements WebMvcConfigurer {
    @Override
    public void addCorsMappings(CorsRegistry registry) {
        registry.addMapping("/**")//项目中的所有接口都支持跨域
                .allowedOriginPatterns("*")//所有地址都可以访问，也可以配置具体地址
                .allowCredentials(true)
                .allowedMethods("*")//"GET", "HEAD", "POST", "PUT", "DELETE", "OPTIONS"
                .maxAge(3600);// 跨域允许时间
    }
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • Properties配置类 RedissonProperties.java，WebSocketProperties读取socketIO、和redisson配置
package com.ddw.chat.controller.socket.properties;

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

@Data
@Component
@ConfigurationProperties(prefix = "redisson")
public class RedissonProperties {
    private int timeout = 3000;
    private String address;
    private String password;
    private int connectionPoolSize = 5;
    private int connectionMinimumIdleSize = 2;
    private int slaveConnectionPoolSize = 250;
    private int masterConnectionPoolSize = 250;
    private String[] sentinelAddresses;
    private String masterName;
    private int database = 1;
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
package com.ddw.chat.controller.socket.properties;

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

@Data
@Component
@ConfigurationProperties(prefix = "websocket")
public class WebSocketProperties {
    /**
     * socket 地址
     */
    private String host;
    /**
     * socket 端口
     */
    private Integer socketPort;
    /**
     * 最大每帧处理数据的长度
     */
    private String maxFramePayloadLength;
    /**
     * http交互最大内容长度
     */
    private String maxHttpContentLength;
    /**
     * Ping 心跳间隔（毫秒）
     */
    private Integer pingInterval;
    /**
     * Ping消息超时时间（毫秒），默认60秒，这个时间间隔内没有接收到心跳消息就会发送超时事件
     */
    private Integer pingTimeout;
    /**
     * 协议升级超时时间（毫秒），默认10秒。HTTP握手升级为ws协议超时时间
     */
    private Integer upgradeTimeout;
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • Configuration配置：RedissonConfig.java，SocketIoConfig.java
package com.ddw.chat.controller.socket.config;

import com.ddw.chat.controller.socket.properties.RedissonProperties;
import jodd.util.StringUtil;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.redisson.config.SingleServerConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RedissonConfig {
    @Autowired
    private RedissonProperties conf;

    @Bean(name="redission",destroyMethod="shutdown")
    public RedissonClient redission() {
        Config config = new Config();
        config.setCodec(new org.redisson.client.codec.StringCodec());
        if(conf.getSentinelAddresses()!=null && conf.getSentinelAddresses().length>0){
            config.useSentinelServers()
                    .setMasterName(conf.getMasterName()).addSentinelAddress(conf.getSentinelAddresses())
                    .setPassword(conf.getPassword()).setDatabase(conf.getDatabase());
        }else{
            SingleServerConfig serverConfig = config.useSingleServer()
                    .setAddress(conf.getAddress())
                    .setTimeout(conf.getTimeout())
                    .setConnectionPoolSize(conf.getConnectionPoolSize())
                    .setConnectionMinimumIdleSize(conf.getConnectionMinimumIdleSize())
                    .setDatabase(conf.getDatabase());
            if(StringUtil.isNotBlank(conf.getPassword())) {
                serverConfig.setPassword(conf.getPassword());
            }
        }
        return Redisson.create(config);
    }
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
package com.ddw.chat.controller.socket.config;

import com.corundumstudio.socketio.AckMode;
import com.corundumstudio.socketio.SocketConfig;
import com.corundumstudio.socketio.SocketIOServer;
import com.corundumstudio.socketio.annotation.SpringAnnotationScanner;
import com.corundumstudio.socketio.store.RedissonStoreFactory;
import com.corundumstudio.socketio.store.pubsub.PubSubStore;
import com.ddw.chat.controller.socket.CustomRedissonStoreFactory;
import com.ddw.chat.controller.socket.properties.WebSocketProperties;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@Slf4j
public class SocketIoConfig {
    @Autowired
    private RedissonClient redisson;

    @Autowired
    private WebSocketProperties webSocketProperties;

    private RedissonStoreFactory createRedissonStoreFactory(){
        log.info("创建 RedissonStoreFactory 开始");
//        RedissonStoreFactory redissonStoreFactory = new RedissonStoreFactory(redisson);
        RedissonStoreFactory redissonStoreFactory = new CustomRedissonStoreFactory(redisson);
        log.info("创建 RedissonStoreFactory 结束");
        return redissonStoreFactory;
    }

    @Bean
    public SocketIOServer getSocketIOServer(){
        log.info("创建 SocketIOServer 开始");
        //Sokcket配置 参考 jdk
        SocketConfig socketConfig = new SocketConfig();
        socketConfig.setTcpNoDelay(true);
        //在默认情况下,当调用close关闭socke的使用,close会立即返回,
        // 但是,如果send buffer中还有数据,系统会试着先把send buffer中的数据发送出去,然后close才返回.
        socketConfig.setSoLinger(0);
        com.corundumstudio.socketio.Configuration config = new com.corundumstudio.socketio.Configuration();
        // 设置监听端口
        config.setPort(webSocketProperties.getSocketPort());
        // 协议升级超时时间（毫秒），默认10000。HTTP握手升级为ws协议超时时间
        config.setUpgradeTimeout(webSocketProperties.getUpgradeTimeout());
        // Ping消息间隔（毫秒），默认25000。客户端向服务器发送一条心跳消息间隔
        config.setPingInterval(webSocketProperties.getPingInterval());
        // Ping消息超时时间（毫秒），默认60000，这个时间间隔内没有接收到心跳消息就会发送超时事件
        config.setPingTimeout(webSocketProperties.getPingTimeout());
        // 推荐使用redisson
        config.setStoreFactory(createRedissonStoreFactory());
        //异常处理
//        config.setExceptionListener(nettyExceptionListener);
        //手动确认
        config.setAckMode(AckMode.MANUAL);
        // 握手协议参数使用JWT的Token认证方案 认证方案
        config.setAuthorizationListener(data -> {
           /* HttpHeaders httpHeaders = data.getHttpHeaders();
            String token = httpHeaders.get("Authorization");*/
            return  true;
        });
        socketConfig.setTcpKeepAlive(true);
        config.setSocketConfig(socketConfig);
        log.info("创建 SocketIOServer 结束");
        return new SocketIOServer(config);
    }

    /**
     * spring
     * @param socketServer
     * @return
     */
    @Bean
    public SpringAnnotationScanner springAnnotationScanner(SocketIOServer socketServer) {
        return new SpringAnnotationScanner(socketServer);
    }

    @Bean
    public PubSubStore pubSubStore(SocketIOServer socketServer) {
        return socketServer.getConfiguration().getStoreFactory().pubSubStore();
    }
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 事件处理器：NettySocketEventHandler.java 处理客户端发送事件
package com.ddw.chat.controller.socket;

import com.corundumstudio.socketio.AckRequest;
import com.corundumstudio.socketio.HandshakeData;
import com.corundumstudio.socketio.SocketIOClient;
import com.corundumstudio.socketio.annotation.OnConnect;
import com.corundumstudio.socketio.annotation.OnDisconnect;
import com.corundumstudio.socketio.annotation.OnEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class NettySocketEventHandler {

    @OnDisconnect
    public void onDisconnect(SocketIOClient client) {
        log.info("--------------------客户端已断开连接--------------------");
        client.disconnect();
    }

    @OnConnect
    public void onConnect(SocketIOClient client) {
        HandshakeData handshakeData = client.getHandshakeData();
        String room = handshakeData.getSingleUrlParam("room");
        client.joinRoom(room);
        //存储SocketIOClient，用于向不同客户端发送消息
        log.info("-客户端[{}]连接成功", room);
    }

    @OnEvent(value = "msg_event")
    public void onMessage(SocketIOClient client, AckRequest ackRequest, String data) throws Exception {
        client.sendEvent("msg_event", "收到消息了！,你发送的内容为：" + data);
    }
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 启动类 SocketServerRunner.java
package com.ddw.chat.controller.socket;

import com.corundumstudio.socketio.SocketIOServer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

@Component
@Order(1)
@Slf4j
public class SocketServerRunner implements CommandLineRunner {
    @Autowired
    private SocketIOServer socketIOServer;

    @Override
    public void run(String... args) throws Exception {
        log.info("socketIOServer 启动");
        socketIOServer.start();
    }
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 自定义StoreFactory：CustomRedissonStoreFactory.java ,此处代码可以根据自己需求自行修改，这里是从 源码里粘贴过来的
package com.ddw.chat.controller.socket;

import com.corundumstudio.socketio.handler.AuthorizeHandler;
import com.corundumstudio.socketio.namespace.Namespace;
import com.corundumstudio.socketio.namespace.NamespacesHub;
import com.corundumstudio.socketio.protocol.JsonSupport;
import com.corundumstudio.socketio.store.RedissonStore;
import com.corundumstudio.socketio.store.RedissonStoreFactory;
import com.corundumstudio.socketio.store.Store;
import com.corundumstudio.socketio.store.pubsub.*;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonClient;

import java.util.Set;
import java.util.UUID;

@Slf4j
public class CustomRedissonStoreFactory extends RedissonStoreFactory {
    private final RedissonClient redisClient;
    private final RedissonClient redisPub;
    private final RedissonClient redisSub;
    private final PubSubStore pubSubStore;

    public CustomRedissonStoreFactory(RedissonClient redisson) {
        this.redisClient = redisson;
        this.redisPub = redisson;
        this.redisSub = redisson;
        this.pubSubStore = new CustomRedissonPubSubStore(this.redisPub, this.redisSub, this.getNodeId());
    }

    public Store createStore(UUID sessionId) {
        return new RedissonStore(sessionId, this.redisClient);
    }

    public PubSubStore pubSubStore() {
        return this.pubSubStore;
    }

    public void shutdown() {
        this.redisClient.shutdown();
        this.redisPub.shutdown();
        this.redisSub.shutdown();
    }

    @Override
    public void init(NamespacesHub namespacesHub, AuthorizeHandler authorizeHandler, JsonSupport jsonSupport) {
        this.pubSubStore().subscribe(PubSubType.DISCONNECT, msg -> log.debug("{} sessionId: {}", PubSubType.DISCONNECT, msg.getSessionId()), DisconnectMessage.class);
        this.pubSubStore().subscribe(PubSubType.CONNECT, msg -> {
            authorizeHandler.connect(msg.getSessionId());
            log.debug("{} sessionId: {}", PubSubType.CONNECT, msg.getSessionId());
        }, ConnectMessage.class);
        this.pubSubStore().subscribe(PubSubType.DISPATCH, msg -> {
			//TODO 重点关注，订阅消息分发逻辑
            String name = msg.getRoom();
            Namespace n = namespacesHub.get(msg.getNamespace());
            if (n != null) {
                n.dispatch(name, msg.getPacket());
            }
            log.debug("{} packet: {}", PubSubType.DISPATCH, msg.getPacket());
        }, DispatchMessage.class);
        this.pubSubStore().subscribe(PubSubType.JOIN, msg -> {
            String name = msg.getRoom();
            Namespace n = namespacesHub.get(msg.getNamespace());
            if (n != null) {
                n.join(name, msg.getSessionId());
            }
            log.debug("{} sessionId: {}", PubSubType.JOIN, msg.getSessionId());
        }, JoinLeaveMessage.class);
        this.pubSubStore().subscribe(PubSubType.BULK_JOIN, msg -> {
            Set<String> rooms = msg.getRooms();

            for (String room : rooms) {
                Namespace n = namespacesHub.get(msg.getNamespace());
                if (n != null) {
                    n.join(room, msg.getSessionId());
                }
            }
            log.debug("{} sessionId: {}", PubSubType.BULK_JOIN, msg.getSessionId());
        }, BulkJoinLeaveMessage.class);
        this.pubSubStore().subscribe(PubSubType.LEAVE, new PubSubListener<JoinLeaveMessage>() {
            public void onMessage(JoinLeaveMessage msg) {
                String name = msg.getRoom();
                Namespace n = namespacesHub.get(msg.getNamespace());
                if (n != null) {
                    n.leave(name, msg.getSessionId());
                }
                log.debug("{} sessionId: {}", PubSubType.LEAVE, msg.getSessionId());
            }
        }, JoinLeaveMessage.class);
        this.pubSubStore().subscribe(PubSubType.BULK_LEAVE, new PubSubListener<BulkJoinLeaveMessage>() {
            public void onMessage(BulkJoinLeaveMessage msg) {
                Set<String> rooms = msg.getRooms();

                for (String room : rooms) {
                    Namespace n = namespacesHub.get(msg.getNamespace());
                    if (n != null) {
                        n.leave(room, msg.getSessionId());
                    }
                }
                log.debug("{} sessionId: {}", PubSubType.BULK_LEAVE, msg.getSessionId());
            }
        }, BulkJoinLeaveMessage.class);

    }
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 自定义 PubSubStore：CustomRedissonPubSubStore.java,这里只改动部分代码，其余代码也是从源码中粘贴的，同样根据自己的需求进行修改
package com.ddw.chat.controller.socket;

import com.corundumstudio.socketio.store.RedissonPubSubStore;
import com.corundumstudio.socketio.store.pubsub.PubSubListener;
import com.corundumstudio.socketio.store.pubsub.PubSubMessage;
import com.corundumstudio.socketio.store.pubsub.PubSubType;
import io.netty.util.internal.PlatformDependent;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.redisson.codec.SerializationCodec;

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;

public class CustomRedissonPubSubStore extends RedissonPubSubStore {
    private final RedissonClient redissonPub;
    private final RedissonClient redissonSub;
    private final Long nodeId;
    private final ConcurrentMap<String, Queue<Integer>> map = PlatformDependent.newConcurrentHashMap();

    public CustomRedissonPubSubStore(RedissonClient redissonPub, RedissonClient redissonSub, Long nodeId) {
        super(redissonPub, redissonSub, nodeId);
        this.redissonPub = redissonPub;
        this.redissonSub = redissonSub;
        this.nodeId = nodeId;
    }

    public void publish(PubSubType type, PubSubMessage msg) {
        msg.setNodeId(this.nodeId);
        this.redissonPub.getTopic(type.toString(), new SerializationCodec()).publish(msg);
    }

    public <T extends PubSubMessage> void subscribe(PubSubType type, final PubSubListener<T> listener, Class<T> clazz) {
        String name = type.toString();
        RTopic topic = this.redissonSub.getTopic(name, new SerializationCodec());
        int regId = topic.addListener(clazz, (channel, msg) -> {
        	 //TODO 重点关注，可以根据具体业务场景进行修改
//            if (!CustomRedissonPubSubStore.this.nodeId.equals(msg.getNodeId())) {
                listener.onMessage(msg);
//            }
        });
        Queue<Integer> list = this.map.get(name);
        if (list == null) {
            list = new ConcurrentLinkedQueue<>();
            Queue<Integer> oldList = this.map.putIfAbsent(name, list);
            if (oldList != null) {
                list = oldList;
            }
        }
        list.add(regId);
    }

    public void unsubscribe(PubSubType type) {
        String name = type.toString();
        Queue<Integer> regIds = this.map.remove(name);
        RTopic topic = this.redissonSub.getTopic(name, new SerializationCodec());

        for (Integer id : regIds) {
            topic.removeListener(id);
        }

    }

    public void shutdown() {
    }
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 消息实体类：SocketMessage<T> 根据需要自行修改
package com.ddw.chat.controller.socket;

import lombok.Data;

@Data
public class SocketMessage<T> {
    private String room;
    private String namespace;
    private String fromUserId;
    private String toUserId;
    private String eventName;
    private T message;
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 消息触发controller：SocketPushController.java
package com.ddw.chat.controller;

import com.corundumstudio.socketio.SocketIOServer;
import com.corundumstudio.socketio.protocol.EngineIOVersion;
import com.corundumstudio.socketio.protocol.Packet;
import com.corundumstudio.socketio.protocol.PacketType;
import com.corundumstudio.socketio.store.pubsub.DispatchMessage;
import com.corundumstudio.socketio.store.pubsub.PubSubStore;
import com.corundumstudio.socketio.store.pubsub.PubSubType;
import com.ddw.chat.controller.socket.SocketMessage;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

import java.util.Collections;

@RestController
@Slf4j
public class SocketPushController {

    @Autowired
    private PubSubStore pubSubStore;
    @Autowired
    private RedissonClient redisson;
    @Autowired
    private SocketIOServer socketIOServer;

    @PostMapping("/push")
    public String push(@RequestBody SocketMessage<String> socketMessage) {
        Packet packet = new Packet(PacketType.MESSAGE, EngineIOVersion.V3);
        packet.setSubType(PacketType.EVENT);
        packet.setName(socketMessage.getEventName());
        packet.setData(Collections.singletonList(socketMessage.getMessage()));
        DispatchMessage dispatchMessage = new DispatchMessage(socketMessage.getRoom(), packet, socketMessage.getNamespace());
        pubSubStore.publish(PubSubType.DISPATCH,dispatchMessage);
        return "ok";

    }

}


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44

二、前端

1. 技术框架
  • bootstrap v3
  • Socket.IO v2.3.0
<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/2.3.0/socket.io.js"></script>

<link rel="stylesheet" href="//cdn.bootcss.com/bootstrap/3.3.5/css/bootstrap.min.css"> 
<link rel="stylesheet" href="//cdn.bootcss.com/bootstrap/3.3.5/css/bootstrap-theme.min.css">
<script src="//cdn.bootcss.com/jquery/1.11.3/jquery.min.js"></script>
<script src="//cdn.bootcss.com/bootstrap/3.3.5/js/bootstrap.min.js"></script>

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
2.前端页面代码
  • index.html
<!DOCTYPE html>
<html lang="en" xmlns="http://www.w3.org/1999/xhtml">
  <head>
    <meta charset="utf-8" />
    <title>webSocket测试</title>
    <link rel="stylesheet" href="lib/bootstrap.min.css" />
    <link rel="stylesheet" href="lib/bootstrap-theme.min.css" />
    <link rel="stylesheet" href="lib/app.css" />
    <script src="lib/socket.io.js"></script>
    <script src="lib/jquery.min.js"></script>
    <script src="lib/bootstrap.min.js"></script>
  </head>

  <body>
    <div class="main-page">
      <div class="col-md-3 part-page">
        <div class="user-box">
          <div class="user-part">
            <img src="img/head-icon.png" class="img-circle head-icon" />
          </div>
          <div class="user-part user-ipart-info">
            <div>用户ID:<span id="userId">86888</span></div>
            <div id="onlineState">
              <span class="label label-default">离线</span>
            </div>
          </div>
        </div>
        <div class="server-config">
          <div class="form-group">
            <label for="ip" class="control-label">IP</label>
            <input class="form-control" id="ip" />
          </div>
          <div class="form-group">
            <label for="port" class="control-label">端口</label>
            <input class="form-control" id="port" />
          </div>
          <div class="form-group">
            <label for="namespace" class="control-label">namespace</label>
            <input class="form-control" id="namespace" value="/websocket" />
          </div>
          <div class="form-group">
            <label for="room" class="control-label">room</label>
            <input class="form-control" id="room" value="def_pub_room" />
          </div>
          <div class="form-group">
            <button type="submit" class="btn btn-danger" id="connect">
              连接
            </button>
          </div>
        </div>
      </div>
      <div class="col-md-9 part-page">
        <div class="panel panel-default" style="height: 100%">
          <div class="panel-heading">消息记录</div>
          <div class="panel-body" style="height: calc(100% - 60px)">
            <div class="well" id="msg-box"></div>
            <div class="col-lg">
              <div class="input-group">
                <input
                  type="text"
                  class="form-control"
                  placeholder="发送信息..."
                  id="message"
                />
                <span class="input-group-btn">
                  <button class="btn btn-default" type="button" id="send">
                    发送
                  </button>
                </span>
              </div>
            </div>
          </div>
        </div>
      </div>
    </div>
  </body>
  <script type="text/javascript">
    $(function () {
      var socket = null
      var userId = ''

      init()

      $('#connect').bind('click', function () {
        var namespace = $('#namespace').val()
        namespace=""
        // var room = $('#room').val()
        var port = $('#port').val()
        var url =
          'http://localhost:' +
          port +
          namespace +
          '?room=' +
          userId +
          '&userId=' +
          userId
        socket = io.connect(url, {
          'reconnection delay': 2000,
          'force new connection': true
        })
        socketInit(socket)
      })

      $('#send').bind('click', function () {
        send()
      })

      function init () {
        userId = 'U' + Math.ceil(Math.random() * 10000)
        $('#userId').html(userId)
        var host = window.location.host
        var ip = host.split(':')[0]
        $('#ip').val(ip)
        $('#port').val('90')
      }

      function socketInit (socket) {
        //监听服务器连接事件
        socket.on('connect', function () {
          $('#msg-box').html(
            $('#msg-box').html() + '<br/>【系统消息】：连接服务器成功！'
          )
          $('#onlineState').html(
            '<span class="label label-success">在线</span>'
          )
        })
        //监听服务器关闭服务事件
        socket.on('disconnect', function () {
          $('#msg-box').html(
            $('#msg-box').html() + '<br/>【系统消息】：与服务器断开了连接！'
          )
          $('#onlineState').html(
            '<span class="label label-default">离线</span>'
          )
        })
        //监听服务器端发送消息事件
        socket.on('msg_event', function (data) {
          $('#msg-box').html(
            $('#msg-box').html() + '<br/>【收到消息】：' + data
          )
        })
      }

      function send () {
        if (socket != null) {
          console.log("发送消息");
          var message = document.getElementById('message').value
          var title = 'message'
          var obj = { message: message, title: title }
          var str = JSON.stringify(obj)
          socket.emit('msg_event', str)
        } else {
          alert('未与服务器链接.')
        }
      }
    })
  </script>
</html>


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 106
  • 107
  • 108
  • 109
  • 110
  • 111
  • 112
  • 113
  • 114
  • 115
  • 116
  • 117
  • 118
  • 119
  • 120
  • 121
  • 122
  • 123
  • 124
  • 125
  • 126
  • 127
  • 128
  • 129
  • 130
  • 131
  • 132
  • 133
  • 134
  • 135
  • 136
  • 137
  • 138
  • 139
  • 140
  • 141
  • 142
  • 143
  • 144
  • 145
  • 146
  • 147
  • 148
  • 149
  • 150
  • 151
  • 152
  • 153
  • 154
  • 155
  • 156
  • 157
  • 158
  • 159
  • app.css
html,
body,
.main-page,
.part-page {
    height: 100%;
}

.part-page {
    padding: 40px 20px;
}

.part-page:nth-child(1) {
    background-image: linear-gradient(to top, #a8edea 0%, #fed6e3 100%);
}

.part-page:nth-child(2) {
    border-right: 1px solid #eee;
}


.user-box {
    border-bottom: 1px solid #bebebe;
    margin-bottom: 10px;
    padding-bottom: 10px;
}

.user-box::after {
    content: "";
    clear: both;
    display: block;
}

.user-part {
    float: left;
}

.user-part:nth-child(2) {
    padding: 10px;
}

.user-part span:nth-child(1) {
    margin: 0 5px;
}

.user-part div {
    margin-bottom: 10px;
    font-weight: 600;
}

.user-ipart-info {
    margin-top: 15px;
}

.user-ipart-info #userId {
    color: brown;
}

.head-icon {
    height: 100px;
    padding: 5px;
    background-color: #fff;
}

.btn-settings-box .btn {
    margin-right: 5px;
}

.server-config {
    margin-top: 20px;
    padding: 10px;
    border: 1px solid #bebebe;
    border-radius: 5px;
}

#msg-box  {
    height: calc(100% - 50px);
    overflow: auto;
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78

三、Nginx配置

  • conf/nginx.cnf 全量配置

worker_processes  1;

events {
    worker_connections  1024;
}

http {
    include       mime.types;
    default_type  application/octet-stream;
    sendfile        on;
    keepalive_requests 8192;
	keepalive_timeout 180s 180s;
	
	map $http_upgrade $connection_upgrade {
        default upgrade;
        '' close;
	}
       
    upstream web-server{
	   server localhost:8080 weight=1;
	   server localhost:9080 weight=1;
	}
	
	upstream ws-server {
		hash $remote_addr consistent;
		server localhost:8090 weight=1; 
		server localhost:9090 weight=1;
	}

    server {
        listen       80;
        server_name  localhost;   
        
        location / {			
            proxy_pass http://web-server/;           
        }
    }
    
    server {
        listen       90;
        server_name  localhost;  
        
		location ~/socket.io/(.*) {
			proxy_pass http://ws-server;
			proxy_http_version 1.1;
			proxy_set_header Upgrade $http_upgrade;
			proxy_set_header Connection "upgrade";
			proxy_set_header Host $http_host;
			proxy_set_header X-NginX-Proxy true;
			proxy_redirect off;
		} 
    }  
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/寸_铁/article/detail/973920
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号