点对点websocket实现

zengh

2021-04-07 f3d3ab1267f35a6d4b22b7a34f5f94bb733cd513

点对点websocket实现

 blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/nettyServer/NettyConfig.java

@@ -1,7 +1,10 @@
package org.springblade.jfpt.nettyServer;
import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

import java.util.concurrent.ConcurrentHashMap;


public class NettyConfig {
@@ -9,4 +12,34 @@
     * 存储每一个客户端接入进来时的channel对象
     */
    public static ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    /**
     * 定义一个channel组，管理所有channel
     * GlobalEventExecutor.INSTANCE 是全局的事件执行器，是一个单例
     */
    private static ChannelGroup channelGroup = new DefaultChannelGroup("用户管理组",GlobalEventExecutor.INSTANCE);

    /**
     * 存放用户与chanel 的对应的信息，用于给指定用户发送信息
     */
    private static ConcurrentHashMap<String, Channel> userChannelMap = new ConcurrentHashMap<>();

    public NettyConfig() {
    }

    /**
     * 获取用户channel 组
     * @return
     */
    public static ChannelGroup getChannelGroup() {
        return channelGroup;
    }

    /**
     * 获取用户channel map
     * @return
     */
    public static ConcurrentHashMap<String, Channel> getUserChannelMap() {
        return userChannelMap;
    }
}

 blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/WebSocketHandler.java

@@ -14,9 +14,11 @@
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import io.netty.util.AttributeKey;
import io.netty.util.CharsetUtil;
import java.util.Date;

import org.springblade.jfpt.nettyServer.NettyConfig;
import org.springblade.jfpt.suser.service.ISuserService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@@ -69,7 +71,10 @@
            //工作状态(0闲置，1工作中)
            String workSt = "0";
            webSocketHandler.suserService.updateUser(num,name,workSt);
            ChannelSupervise.removeChannel(ctx.channel());
            //ChannelSupervise.removeChannel(ctx.channel());

            NettyConfig.getChannelGroup().remove(ctx.channel());
            removeUserId(ctx);
        }
    }

@@ -100,6 +105,13 @@
        String request = ((TextWebSocketFrame) frame).text();

        if (!request.equals("ping")){

            NettyConfig.getUserChannelMap().put(request,ctx.channel());

            //将用户id作为自定义属性加入到channel 中，方便随时channel中获取用户id
            AttributeKey<String> key = AttributeKey.valueOf("userId");
            ctx.channel().attr(key).setIfAbsent(request);

            //把用户信息添加到通道里
            ChannelSupervise.addChannel(ctx.channel(),request);
            //用户在线状态
@@ -169,4 +181,14 @@
        ctx.close();
    }

    /**
     * 删除用户与channel 对应关系
     * @param ctx
     */
    private void removeUserId(ChannelHandlerContext ctx){
        AttributeKey<String> key = AttributeKey.valueOf("userId");
        String userId = ctx.channel().attr(key).get();
        NettyConfig.getUserChannelMap().remove(userId);
    }

}

 blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/controller/PushMsgController.java

New file
@@ -0,0 +1,30 @@
package org.springblade.jfpt.webscoket.controller;

import org.springblade.jfpt.webscoket.service.IPushMsgService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author lq
 * @date 2020/4/1 11:22
 */
@RestController
public class PushMsgController {

    @Autowired
    private IPushMsgService pushMsgService;

    @PostMapping("/pushUser")
    public String pushUser(String userId,String msg){
        pushMsgService.pushMsg(userId, msg);
        return "消息发送成功："+msg;
    }

    @PostMapping("/pushAll")
    public String pushAll(String msg){
        pushMsgService.pushMsg(msg);
        return "消息发送成功："+msg;
    }

}

 blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/service/IPushMsgService.java

New file
@@ -0,0 +1,21 @@
package org.springblade.jfpt.webscoket.service;

/**
 * @author 123456
 */
public interface IPushMsgService {

    /**
     * 给指定用户发送消息
     * @param userId
     * @param msg
     */
    void pushMsg(String userId,String msg);

    /**
     * 给所有用户发送消息
     * @param msg
     */
    void pushMsg(String msg);

}

 blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/webscoket/service/impl/PushMsgServiceImpl.java

New file
@@ -0,0 +1,29 @@
package org.springblade.jfpt.webscoket.service.impl;

import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.springblade.jfpt.nettyServer.NettyConfig;
import org.springblade.jfpt.webscoket.service.IPushMsgService;
import org.springframework.stereotype.Service;

/**
 * @author lq
 * @date 2020/4/1 11:20
 */
@Service
public class PushMsgServiceImpl implements IPushMsgService {
    @Override
    public void pushMsg(String userId, String msg) {
        Channel channel = NettyConfig.getUserChannelMap().get(userId);
        channel.writeAndFlush(new TextWebSocketFrame(msg));
    }

    @Override
    public void pushMsg(String msg) {
        ChannelGroup group = NettyConfig.getChannelGroup();
        String name = group.name();
        System.out.println("空间大小："+group.size()+",名字："+name);
        group.writeAndFlush(new TextWebSocketFrame(msg));
    }
}