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

基于Netty实现websocket集群部署实现方案_springboot netty websocket集群

作者：IT小白 | 2024-05-27 23:35:10

赞

踩

springboot netty websocket集群

基于Netty实现websocket集群部署实现方案

每天多学一点点~
话不多说，这就开始吧…

文章目录

1.前言

最近公司在做saas平台，其中涉及到重构一个无人机项目。无人机推流拉流用了腾讯云直播、点播功能。安卓端集成了大疆的sdk，需要在飞无人机的时候一直推送飞行信息(比如飞行高度，飞行路线、风向什么的)。
之前用的tomcat自带的websocket,spring-boot-starter-websocke集成，但是性能可能有点问题。这次重构，打算换成netty。因为是saas化服务，领导还要求可集群部署，这里提供一种解决方案----redis的pub、sub（当然大型项目最好还是zk）。

在这里插入图片描述

2. 整体思路

  1. 跨服务之间案例采用redis的发布和订阅进行传递消息。多个实例监听同一个channel。客户端上线存入redis，下线清除redis。
  2. 用户A在发送消息给用户B时候，需要传递B的channeId，以用于服务端进行查找channeId所属是否自己的服务内。
  3. 若本次channel在A实例的本地缓存中能找到，说明属于自己服务，直接发送。若不在，则通过redis的pub到其他服务实例，其他实例监听到消息之后，判断这个channel是否在本地，在就发送。

当然，redis 的pub sub本身也是有缺陷的，比如： 数据可靠性无法保证、扩展性差、资源消耗高，如果是正式比较大型的可以换成zk，思路差不多。

3. 代码demo

nettywebsocket
├─ main
│ 	├─ java
│ 	│ 	└─ com
│ 	│ 		└─ example
│ 	│ 			└─ demo
│ 	│ 				├─ advice
│ 	│ 				│ 	├─ AdviceController.java
│ 	│ 				│ 	├─ BusinessException.java
│ 	│ 				│ 	├─ ErrorConstant.java
│ 	│ 				│ 	└─ NotFoundController.java
│ 	│ 				├─ domain
│ 	│ 				│ 	├─ CommonResult.java
│ 	│ 				│ 	├─ IErrorCode.java
│ 	│ 				│ 	├─ MyMessage.java
│ 	│ 				│ 	├─ ResultCode.java
│ 	│ 				│ 	├─ ServerInfo.java
│ 	│ 				│ 	└─ UserChannelInfo.java
│ 	│ 				├─ NettywebsocketApplication.java
│ 	│ 				├─ redis
│ 	│ 				│ 	├─ config
│ 	│ 				│ 	│ 	└─ RedisConfig.java
│ 	│ 				│ 	├─ RedisChannelListener.java
│ 	│ 				│ 	└─ RedisUtil.java
│ 	│ 				├─ server
│ 	│ 				│ 	├─ HttpRequestHandler.java
│ 	│ 				│ 	├─ NioWebSocketChannelInitializer.java
│ 	│ 				│ 	├─ WebsocketServer.java
│ 	│ 				│ 	└─ WebSocketServerHandler.java
│ 	│ 				├─ util
│ 	│ 				│ 	├─ CacheUtil.java
│ 	│ 				│ 	├─ DirectoryTreeV1.java 
│ 	│ 				│ 	├─ MsgUtil.java
│ 	│ 				│ 	└─ NetUtil.java
│ 	│ 				└─ web
│ 	│ 					└─ WebSocketController.java
│ 	└─ resources
│ 		├─ application.yml
│ 		└─ static
│ 			├─ index.html
│ 			├─ index2.html
│ 			└─ index3.html
└─ test
	└─ java
		└─ com
			└─ example
				└─ demo
					└─ NettywebsocketApplicationTests.java

  • 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

这里列出一些关键代码
实体类

/* ━━━━━━佛祖保佑━━━━━━
 *                  ,;,,;
 *                ,;;'(    社
 *      __      ,;;' ' \   会
 *   /'  '\'~~'~' \ /'\.)  主
 * ,;(      )    /  |.     义
 *,;' \    /-.,,(   ) \    码
 *     ) /       ) / )|    农
 *     ||        ||  \)
 *     (_\       (_\
 * ━━━━━━永无BUG━━━━━━
 * @author ：zjq
 * @date ：2021/4/7 0:38
 * @description: TODO 定义信息传输协议，这个看似简单但非常重要，每一个通信的根本就是定义传输协议信息
 * @version: V1.0
 * @slogan: 天下风云出我辈，一入代码岁月催
 */
@Data
@AllArgsConstructor
public class MyMessage {

    //发送给某人，某人channelId
    private String toChannelId;

    //消息内容
    private String content;
}
服务端信息
@Data
@AllArgsConstructor
public class ServerInfo {

    //IP
    private String ip;

    //端口
    private int port;

    //启动时间
    private Date openDate;
}

# 用户管道信息；记录某个用户分配到某个服务端
@Data
@AllArgsConstructor
public class UserChannelInfo {

    //服务端：IP
    private String ip;

    //服务端：port
    private int port;

    //channelId
    private String channelId;

    //链接时间
    private Date linkDate;

}

  • 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

RedisConfig.java 配置redis 序列化 以及 监听

/* ━━━━━━佛祖保佑━━━━━━
 *                  ,;,,;
 *                ,;;'(    社
 *      __      ,;;' ' \   会
 *   /'  '\'~~'~' \ /'\.)  主
 * ,;(      )    /  |.     义
 *,;' \    /-.,,(   ) \    码
 *     ) /       ) / )|    农
 *     ||        ||  \)
 *     (_\       (_\
 * ━━━━━━永无BUG━━━━━━
 * @author ：zjq
 * @date ：2021/4/7 0:42
 * @description: TODO           redis 配置
 * @version: V1.0
 * @slogan: 天下风云出我辈，一入代码岁月催
 */
@Configuration
public class RedisConfig {

    @Autowired
    private RedisConnectionFactory connectionFactory;


    /**
     *  序列化 配置
     * @param redisConnectionFactory
     * @return
     */
    @Bean
    public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {

        RedisTemplate<Object, Object> template = new RedisTemplate<>();
        //Jackson2JsonRedisSerializer 序列化方式，取出来的是map
        template.setDefaultSerializer(new Jackson2JsonRedisSerializer<Object>(Object.class));
        //设置key的序列化，让其没有 ""
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        template.setKeySerializer(stringRedisSerializer);
        template.setHashKeySerializer(stringRedisSerializer);

        template.setConnectionFactory(redisConnectionFactory);
        return template;
    }
    
    /**
     * 监听器配置  pus sub
     */

    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer() {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(messageListenerAdapter(), channelTopic());
        return container;
    }

    @Bean
    public MessageListenerAdapter messageListenerAdapter() {
        return new MessageListenerAdapter(redisChannelListener());
    }

    @Bean
    public  RedisChannelListener redisChannelListener() {
        return new RedisChannelListener();
    }

    @Bean
    ChannelTopic channelTopic() {
        return new ChannelTopic("uav-flight-message");
    }


}

  • 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

RedisChannelListener.java 监听业务逻辑

/* ━━━━━━佛祖保佑━━━━━━
 *                  ,;,,;
 *                ,;;'(    社
 *      __      ,;;' ' \   会
 *   /'  '\'~~'~' \ /'\.)  主
 * ,;(      )    /  |.     义
 *,;' \    /-.,,(   ) \    码
 *     ) /       ) / )|    农
 *     ||        ||  \)
 *     (_\       (_\
 * ━━━━━━永无BUG━━━━━━
 * @author ：zjq
 * @date ：2021/4/7 0:42
 * @description: TODO           redis监听
 * @version: V1.0
 * @slogan: 天下风云出我辈，一入代码岁月催
 */
@Slf4j
public class RedisChannelListener implements MessageListener {


    @Override
    public void onMessage(Message message, @Nullable byte[] pattern) {
        log.info("sub message :) channel[cleanNoStockCache] !");

        log.info("接收到PUSH消息：{}", message);
        MyMessage msgAgreement = JSON.parseObject(message.getBody(), MyMessage.class);
        String toChannelId = msgAgreement.getToChannelId();
        Channel channel = CacheUtil.cacheChannel.get(toChannelId);
        if (null == channel) {
            return;
        }
        // 发送消息
        channel.writeAndFlush(new TextWebSocketFrame(MsgUtil.obj2Json(msgAgreement) + "  redis listener lalalalala "));
    }

}

  • 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

RedisUtil.java redis工具类

/* ━━━━━━佛祖保佑━━━━━━
 *                  ,;,,;
 *                ,;;'(    社
 *      __      ,;;' ' \   会
 *   /'  '\'~~'~' \ /'\.)  主
 * ,;(      )    /  |.     义
 *,;' \    /-.,,(   ) \    码
 *     ) /       ) / )|    农
 *     ||        ||  \)
 *     (_\       (_\
 * ━━━━━━永无BUG━━━━━━
 * @author ：zjq
 * @date ：2021/4/7 0:42
 * @description: TODO       redis工具类
 * @version: V1.0
 * @slogan: 天下风云出我辈，一入代码岁月催
 */
@Component
public class RedisUtil {

    @Autowired
    private StringRedisTemplate redisTemplate;

    public void pushObj(UserChannelInfo userChannelInfo) {
        redisTemplate.opsForHash().put("uav-flight-user",
                userChannelInfo.getChannelId(), JSON.toJSONString(userChannelInfo));
    }

    public List<UserChannelInfo> popList() {
        List<Object> values = redisTemplate.opsForHash().values("uav-flight-user");
        if (null == values) {
            return new ArrayList<>();
        }

        List<UserChannelInfo> userChannelInfoList = new ArrayList<>();

        for (Object strJson : values) {
            userChannelInfoList.add(JSON.parseObject(strJson.toString(), UserChannelInfo.class));
        }
        return userChannelInfoList;
    }

    public void remove(String channelId) {
        redisTemplate.opsForHash().delete("uav-flight-user", channelId);
    }

    public void clear() {
        redisTemplate.delete("uav-flight-user");
    }


    public void push(String channel, String message) {
        redisTemplate.convertAndSend(channel, message);

    }

}


  • 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

CacheUtil.java 缓存必要信息，用于业务流程处理

/* ━━━━━━佛祖保佑━━━━━━
 *                  ,;,,;
 *                ,;;'(    社
 *      __      ,;;' ' \   会
 *   /'  '\'~~'~' \ /'\.)  主
 * ,;(      )    /  |.     义
 *,;' \    /-.,,(   ) \    码
 *     ) /       ) / )|    农
 *     ||        ||  \)
 *     (_\       (_\
 * ━━━━━━永无BUG━━━━━━
 * @author ：zjq
 * @date ：2021/4/7 0:38
 * @description: TODO
 * @version: V1.0
 * @slogan: 天下风云出我辈，一入代码岁月催
 */
public class CacheUtil {

    // 缓存channel
    public static Map<String, Channel> cacheChannel = Collections.synchronizedMap(new HashMap<String, Channel>());

    // 缓存服务信息
    public static Map<Integer, ServerInfo> serverInfoMap = Collections.synchronizedMap(new HashMap<Integer, ServerInfo>());

    // 缓存 websocket 服务端
    public static Map<Integer, WebsocketServer> serverMap = Collections.synchronizedMap(new HashMap<Integer, WebsocketServer>());

}

  • 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

WebsocketServer.java 基于netty的websocket服务端

/* ━━━━━━佛祖保佑━━━━━━
 *                  ,;,,;
 *                ,;;'(    社
 *      __      ,;;' ' \   会
 *   /'  '\'~~'~' \ /'\.)  主
 * ,;(      )    /  |.     义
 *,;' \    /-.,,(   ) \    码
 *     ) /       ) / )|    农
 *     ||        ||  \)
 *     (_\       (_\
 * ━━━━━━永无BUG━━━━━━
 * @author ：zjq
 * @date ：2021/4/7 0:42
 * @description: TODO
 * @version: V1.0
 * @slogan: 天下风云出我辈，一入代码岁月催
 */
@Slf4j
@Service
public class WebsocketServer implements Callable<Channel> {


    @Value("${netty.port}")
    private int port;

    //配置服务端NIO线程组
    private final EventLoopGroup bossGroup = new NioEventLoopGroup();

    private final EventLoopGroup workerGroup = new NioEventLoopGroup();


    @Autowired
    private NioWebSocketChannelInitializer nioWebSocketChannelInitializer;

    private Channel channel = null;

    public Channel getChannel() {
        return this.channel;
    }

    public void setChannel(Channel channel) {
        this.channel = channel;
    }

    /**
     *  这里有点问题，换句话，应该直接 在系统启动的时候初始化一次更好，可以优化下
     * @return
     * @throws Exception
     */
    @Override
    public Channel call() throws Exception {
        ChannelFuture channelFuture = null;
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        try {
            //boss辅助客户端的tcp连接请求  worker负责与客户端之前的读写操作
            serverBootstrap.group(bossGroup, workerGroup)
                    //配置客户端的channel类型
                    .channel(NioServerSocketChannel.class)
                    //配置TCP参数，握手字符串长度设置
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    //TCP_NODELAY算法，尽可能发送大块数据，减少充斥的小块数据
                    .option(ChannelOption.TCP_NODELAY, true)
                    //开启心跳包活机制，就是客户端、服务端建立连接处于ESTABLISHED状态，超过2小时没有交流，机制会被启动
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    //配置固定长度接收缓存区分配器
                    .childOption(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(592048))
                    // websocket 初始化 handler
                    .childHandler(nioWebSocketChannelInitializer);

            log.info("Netty Websocket服务器启动完成，已绑定端口 " + port + " 阻塞式等候客户端连接");

            channelFuture = serverBootstrap.bind(port).sync();
            this.channel = channelFuture.channel();
            // 因为是通过接口调用所以这里不能异步
//            channel.closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e.getMessage());
        }

        return channel;
    }


    /**
     * 摧毁
     */
    public void destroy() {
        if (null == channel) {
            return;
        }
        channel.close();
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }

}

  • 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

NioWebSocketChannelInitializer.java 初始化channel

@Service
public class NioWebSocketChannelInitializer extends ChannelInitializer<SocketChannel> {

    @Autowired
    private WebSocketServerHandler webSocketServerHandler;

    @Autowired
    private HttpRequestHandler httpRequestHandler;

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        //设置log监听器，并且日志级别为debug，方便观察运行流程
        ch.pipeline().addLast("logging", new LoggingHandler("DEBUG"));
        //设置解码器
        ch.pipeline().addLast("http-codec", new HttpServerCodec());
        //聚合器，使用websocket会用到 把HTTP头、HTTP体拼成完整的HTTP请求
        ch.pipeline().addLast("aggregator", new HttpObjectAggregator(65536));
        //用于大数据的分区传输
        ch.pipeline().addLast("http-chunked", new ChunkedWriteHandler());
        // 用于http 升级成 websocket
        ch.pipeline().addLast("http-handler", httpRequestHandler);
        //自定义的业务handler 处理websocket
        ch.pipeline().addLast("handler", webSocketServerHandler);
    }
}

  • 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

HttpRequestHandler.java 用于对http 协议 升级 的 handler

/* ━━━━━━佛祖保佑━━━━━━
 *                  ,;,,;
 *                ,;;'(    社
 *      __      ,;;' ' \   会
 *   /'  '\'~~'~' \ /'\.)  主
 * ,;(      )    /  |.     义
 *,;' \    /-.,,(   ) \    码
 *     ) /       ) / )|    农
 *     ||        ||  \)
 *     (_\       (_\
 * ━━━━━━永无BUG━━━━━━
 * @author ：zjq
 * @date ：2021/4/7 1:45
 * @description: TODO            用于对http 协议 升级 的 handler
 * @version: V1.0
 * @slogan: 天下风云出我辈，一入代码岁月催
 */
@Component
@Sharable
public class HttpRequestHandler extends SimpleChannelInboundHandler<Object> {

    /**
     * 读取 数据
     * 描述：读取完连接的消息后，对消息进行处理。
     * 这里仅处理HTTP请求，WebSocket请求交给下一个处理器。
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof FullHttpRequest) {
            handleHttpRequest(ctx, (FullHttpRequest) msg);
        } else if (msg instanceof WebSocketFrame) {
            ctx.fireChannelRead(((WebSocketFrame) msg).retain());
        } 
    }

    /**
     * 描述：处理Http请求，主要是完成HTTP协议到Websocket协议的升级
     * @param ctx
     * @param req
     */
    private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
        if (!req.decoderResult().isSuccess()) {
            sendHttpResponse(ctx, req,
                    new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            return;
        }

        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
                "ws:/" + ctx.channel() + "/websocket", null, false);
        WebSocketServerHandshaker handshaker = wsFactory.newHandshaker(req);

        if (handshaker == null) {
            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
        } else {
            handshaker.handshake(ctx.channel(), req);
        }
    }
    
    private void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res) {
        // 返回应答给客户端
        if (res.status().code() != 200) {
            ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
            res.content().writeBytes(buf);
            buf.release();
        }
        // 如果是非Keep-Alive，关闭连接
        boolean keepAlive = HttpUtil.isKeepAlive(req);
        ChannelFuture f = ctx.channel().writeAndFlush(res);
        if (!keepAlive) {
            f.addListener(ChannelFutureListener.CLOSE);
        }
    }

    /**
     * 描述：异常处理，关闭channel
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

  • 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

WebSocketServerHandler.java 处理接收消息，上线下线等业务逻辑handler

/* ━━━━━━佛祖保佑━━━━━━
 *                  ,;,,;
 *                ,;;'(    社
 *      __      ,;;' ' \   会
 *   /'  '\'~~'~' \ /'\.)  主
 * ,;(      )    /  |.     义
 *,;' \    /-.,,(   ) \    码
 *     ) /       ) / )|    农
 *     ||        ||  \)
 *     (_\       (_\
 * ━━━━━━永无BUG━━━━━━
 * @author ：zjq
 * @date ：2021/4/7 1:37
 * @description: TODO
 * @version: V1.0
 * @slogan: 天下风云出我辈，一入代码岁月催
 */
@Slf4j
@Service
@ChannelHandler.Sharable       // 因是通过注入的方式，而不是通过new,共享channel
public class WebSocketServerHandler extends SimpleChannelInboundHandler<WebSocketFrame> {

    @Autowired
    private RedisUtil redisUtil;

    private WebSocketServerHandshaker handshaker;

    /**
     * 读取 消息
     *
     * @param ctx
     * @param frame
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) throws Exception {

        // 关闭请求
        if (frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
            return;
        }
        // ping请求
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        // 只支持文本格式，不支持二进制消息
        if (!(frame instanceof TextWebSocketFrame)) {
            sendErrorMessage(ctx, "仅支持文本(Text)格式，不支持二进制消息");
        }

        // 客服端发送过来的消息
        String request = ((TextWebSocketFrame) frame).text();
        log.info("服务端收到新信息：" + request);
        try {
            //接收msg消息{此处已经不需要自己进行解码}
            System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + " 接收到消息内容：" + request);

            MyMessage msgAgreement = MsgUtil.json2Obj(request.toString());

            String toChannelId = msgAgreement.getToChannelId();
            //判断接收消息用户是否在本服务端
            Channel channel = CacheUtil.cacheChannel.get(toChannelId);
            if (null != channel) {
                channel.writeAndFlush(new TextWebSocketFrame(MsgUtil.obj2Json(msgAgreement) + " lalalalalalalalalalalalal"));
                return;
            }

            //如果为NULL则接收消息的用户不在本服务端，需要push消息给全局
            log.info("接收消息的用户不在本服务端，PUSH！");
            redisUtil.push("uav-flight-message", MsgUtil.obj2Json(msgAgreement));


        } catch (Exception e) {
            sendErrorMessage(ctx, "JSON字符串转换出错！");
            e.printStackTrace();
        }
    }

    /**
     * 抓住异常，当发生异常的时候，可以做一些相应的处理，比如打印日志、关闭链接
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
        // 1. 清除 redis
        redisUtil.remove(ctx.channel().id().toString());
        // 2. 清除缓存
        CacheUtil.cacheChannel.remove(ctx.channel().id().toString(), ctx.channel());
        System.out.println("异常信息：\r\n" + cause.getMessage());
    }

    /**
     * 当客户端主动链接服务端的链接后，这个通道就是活跃的了。也就是客户端与服务端建立了通信通道并且可以传输数据
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        SocketChannel channel = (SocketChannel) ctx.channel();
        System.out.println("客户端上线");
        System.out.println("客户端信息：有一客户端链接到本服务端。channelId：" + channel.id());
        System.out.println("客户端IP:" + channel.localAddress().getHostString());
        System.out.println("客户端Port:" + channel.localAddress().getPort());
        System.out.println("客户端信息完毕");

        //保存用户信息
        UserChannelInfo userChannelInfo = new UserChannelInfo(channel.localAddress().getHostString(),
                channel.localAddress().getPort(), channel.id().toString(), new Date());
        // 放入 redis 和 缓存
        redisUtil.pushObj(userChannelInfo);
        CacheUtil.cacheChannel.put(channel.id().toString(), channel);
        //通知客户端链接建立成功
        String str = "通知客户端链接建立成功" + " " + new Date() + " " + channel.localAddress().getHostString() + "\r\n";
        ctx.writeAndFlush(MsgUtil.buildMsg(channel.id().toString(), str));
    }


    /**
     * 当客户端主动断开服务端的链接后，这个通道就是不活跃的。也就是说客户端与服务端的关闭了通信通道并且不可以传输数据
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("客户端断开链接" + ctx.channel().localAddress().toString());
        // 移除 redis  和 缓存
        redisUtil.remove(ctx.channel().id().toString());
        CacheUtil.cacheChannel.remove(ctx.channel().id().toString(), ctx.channel());
    }


    private void sendErrorMessage(ChannelHandlerContext ctx, String errorMsg) {
        String responseJson = "不支持二进制消息";
        ctx.channel().writeAndFlush(new TextWebSocketFrame(responseJson));
    }

}

  • 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

WebSocketController.java 控制层，开启、关闭netty，查看服务端、用户管道列表(其实这里可以根据业务逻辑，项目启动时候就启动netty，也不一定非要调接口开启)

/* ━━━━━━佛祖保佑━━━━━━
 *                  ,;,,;
 *                ,;;'(    社
 *      __      ,;;' ' \   会
 *   /'  '\'~~'~' \ /'\.)  主
 * ,;(      )    /  |.     义
 *,;' \    /-.,,(   ) \    码
 *     ) /       ) / )|    农
 *     ||        ||  \)
 *     (_\       (_\
 * ━━━━━━永无BUG━━━━━━
 * @author ：zjq
 * @date ：2021/4/7 1:20
 * @description: TODO     http://www.websocket-test.com    在线测试
 *                        {"content":"xxxx","toChannelId":"db3abfed"}   测试数据  toChannelId 在 redis 中
 * @version: V1.0
 * @slogan: 天下风云出我辈，一入代码岁月催
 */
@RestController
@RequestMapping
@Slf4j
public class WebSocketController {


    @Value("${netty.port}")
    private int port;

    @Autowired
    private RedisUtil redisUtil;

    @Autowired
    private WebsocketServer websocketServer;

    //默认线程池 先意思意思 这么用
    private static ExecutorService executorService = Executors.newCachedThreadPool();


    /**
     * 开启 netty
     *
     * @return
     */
    @RequestMapping("/openNettyServer")
    public CommonResult openNettyServer() {

        try {

            log.info("启动Netty服务，获取可用端口：{}", port);
            Future<Channel> future = executorService.submit(websocketServer);
            Channel channel = future.get();
            if (null == channel) {
                throw new RuntimeException("netty server open error channel is null");
            }
            while (!channel.isActive()) {
                log.info("启动Netty服务，循环等待启动...");
                Thread.sleep(500);
            }
            // 放入 缓存
            CacheUtil.serverInfoMap.put(port, new ServerInfo(NetUtil.getHost(), port, new Date()));
            CacheUtil.serverMap.put(port, websocketServer);

            log.info("启动Netty服务，完成：{}", channel.localAddress());
            return CommonResult.success();
        } catch (Exception e) {
            log.error("启动Netty服务失败", e);
            return CommonResult.failed(e.getMessage());
        }
    }

    /**
     * 关闭 netty
     *
     * @return
     */
    @RequestMapping("/closeNettyServer")
    public CommonResult closeNettyServer() {
        try {
            log.info("关闭Netty服务开始，端口：{}", port);
            WebsocketServer websocketServer = CacheUtil.serverMap.get(port);
            if (null == websocketServer) {
                CacheUtil.serverMap.remove(port);
                return CommonResult.success();
            }
            websocketServer.destroy();
            CacheUtil.serverMap.remove(port);
            CacheUtil.serverInfoMap.remove(port);
            log.info("关闭Netty服务完成，端口：{}", port);
            return CommonResult.success();
        } catch (Exception e) {
            log.error("关闭Netty服务失败，端口：{}", port, e);
            return CommonResult.failed();
        }
    }

    /**
     *  查 服务端 列表
     * @return
     */
    @RequestMapping("/queryNettyServerList")
    public Collection<ServerInfo> queryNettyServerList() {
        try {
            Collection<ServerInfo> serverInfos = CacheUtil.serverInfoMap.values();
            log.info("查询服务端列表。{}", JSON.toJSONString(serverInfos));
            return serverInfos;
        } catch (Exception e) {
            log.info("查询服务端列表失败。", e);
            return null;
        }
    }

    /**
     *  从 redis 查 用户管道
     * @return
     */
    @RequestMapping("/queryUserChannelInfoList")
    public List<UserChannelInfo> queryUserChannelInfoList() {
        try {
            log.info("查询用户列表信息开始");
            List<UserChannelInfo> userChannelInfoList = redisUtil.popList();
            log.info("查询用户列表信息完成。list：{}", JSON.toJSONString(userChannelInfoList));
            return userChannelInfoList;
        } catch (Exception e) {
            log.error("查询用户列表信息失败", e);
            return null;
        }
    }

}

  • 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

application.yml 配置文件

server:
  port: 8080
netty:
  port: 9000

spring:
  application:
    name: nettywebsocket
  redis:
    database: 1
    timeout: 6000   #超时时间
    host: 127.0.0.1
    password: 123456
    port: 6379
    lettuce:        # 这里标明使用lettuce配置
      pool:
        max‐idle: 50              # 连接池中的最大空闲连接
        min‐idle: 10              # 连接池中的最小空闲连接
        max‐active: 100           # 连接池最大连接数（使用负值表示没有限制）
        max‐wait: 1000            #  连接池最大阻塞等待时间（使用负值表示没有限制）


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21

4. 测试

可以用index页面，也可以直接 在线测试 websocket在线测试

index.html

<!DOCTYPE HTML>
<html>
<head>
    <title>My WebSocket</title>
</head>

<body>
<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://127.0.0.1:9000/ws");    // netty 端口
    } else {
        alert('Not support websocket')
    }

    //连接发生错误的回调方法
    websocket.onerror = function () {
        setMessageInnerHTML("error");
    };

    //连接成功建立的回调方法
    websocket.onopen = function (event) {
        //setMessageInnerHTML("open");
    }

    //接收到消息的回调方法
    websocket.onmessage = function (event) {
        console.log("ok i am receive " + event.data);
        setMessageInnerHTML(event.data);
    }

    //连接关闭的回调方法
    websocket.onclose = function () {
        setMessageInnerHTML("close");
    }

    //监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。
    window.onbeforeunload = function () {
        websocket.close();
    }

    //将消息显示在网页上
    function setMessageInnerHTML(innerHTML) {
        document.getElementById('message').innerHTML += 'receive message is : ' + innerHTML + '<br/>';
    }

    //关闭连接
    function closeWebSocket() {
        websocket.close();
    }

    //发送消息
    function send() {
        var message = document.getElementById('text').value;
        websocket.send(message);
    }
</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

启动2个实例,，分别调用接口开启netty服务端，连接上线，触发上线事件，便会在redis和缓存中存入信息。然后在线测试

消息体，json格式，根据toChannelId判断是否需要pub
{"content":"xxxx","toChannelId":"58f5ef52"}

  • 1
  • 2

在这里插入图片描述

http://localhost:8080/index.html
在这里插入图片描述

5.结语

世上无难事，只
怕有心人，每天积累一点点，fighting！！！
2021，加油，fighting，希望可以少些crud啦！

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/IT小白/article/detail/634592
推荐阅读
相关标签

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

  

闽ICP备14008679号