导航：首页 > 互联网科技 >

Netty中怎么整合WebSocket

发表于：2024-11-27 作者：千家信息网编辑

千家信息网最后更新 2024年11月27日，Netty中怎么整合WebSocket，很多新手对此不是很清楚，为了帮助大家解决这个难题，下面小编将为大家详细讲解，有这方面需求的人可以来学习下，希望你能有所收获。WebSocket协议是基于TCP的

千家信息网最后更新 2024年11月27日Netty中怎么整合WebSocket

Netty中怎么整合WebSocket，很多新手对此不是很清楚，为了帮助大家解决这个难题，下面小编将为大家详细讲解，有这方面需求的人可以来学习下，希望你能有所收获。

WebSocket协议是基于

TCP的一种新的网络协议。它实现了浏览器与服务器全双工(full-duplex)通信--允许服务器主动发送信息给客户端，它是先进行一次Http的连接，连接成功后转为TCP连接。

现在我们来做一个WebSocket HelloWorld,意思为接收一条WebSocket客户端发送过来的消息，然后刷到所有连接上的客户端，大家都可以看到这条消息。

@Slf4j@AllArgsConstructorpublic class WebSocketServer {private int port;    public void run() throws InterruptedException {        EventLoopGroup boss = new NioEventLoopGroup();        EventLoopGroup worker = new NioEventLoopGroup();        try {            ServerBootstrap bootstrap = new ServerBootstrap();            bootstrap.group(boss,worker)                    .channel(NioServerSocketChannel.class)                    .option(ChannelOption.SO_BACKLOG,1024)                    .childOption(ChannelOption.TCP_NODELAY,true)                    .childHandler(new ChannelInitializer() {@Override                        protected void initChannel(SocketChannel ch) throws Exception {//web基于http协议的解码器                            ch.pipeline().addLast(new HttpServerCodec());                            //对大数据流的支持                            ch.pipeline().addLast(new ChunkedWriteHandler());                            //对http message进行聚合，聚合成FullHttpRequest或FullHttpResponse                            ch.pipeline().addLast(new HttpObjectAggregator(1024 * 64));                            //websocket服务器处理对协议，用于指定给客户端连接访问的路径                            //该handler会帮你处理一些繁重的复杂的事                            //会帮你处理握手动作：handshaking(close,ping,pong) ping + pong = 心跳                            //对于websocket来讲，都是以frames进行传输的，不同的数据类型对应的frames也不同                            ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws"));                            //添加我们的自定义channel处理器                            ch.pipeline().addLast(new WebSocketHandler());                        }                    });            log.info("服务器启动中");            ChannelFuture future = bootstrap.bind(port).sync();            future.channel().closeFuture().sync();        } finally {            worker.shutdownGracefully();            boss.shutdownGracefully();        }    }}

channel处理器

/** * TextWebSocketFrame: 在netty中，用于为websocket专门处理文本的对象，frame是消息的载体 */@Slf4jpublic class WebSocketHandler extends SimpleChannelInboundHandler {//用于记录和管理所有客户端的channel    private static ChannelGroup clients = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);    @Override    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {//获取客户端传输过来的消息        String content = msg.text();        log.info("接收到的数据" + content);        clients.stream().forEach(channel -> channel.writeAndFlush(new TextWebSocketFrame("[服务器在]" + LocalDateTime.now() + "接收到消息:" + content)        ));        //下面的方法与上面一致//        clients.writeAndFlush(new TextWebSocketFrame("[服务器在]" + LocalDateTime.now() +//                "接收到消息:" + content));    }/**     * 当客户端连接服务端之后(打开连接)     * 获取客户端的channel,并且放到ChannelGroup中去进行管理     * @param ctx     * @throws Exception     */    @Override    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {clients.add(ctx.channel());    }@Override    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {//当触发handlerRemoved,ChannelGroup会自动移除对应的客户端的channel        //所以下面这条语句可不写//        clients.remove(ctx.channel());        log.info("客户端断开，channel对应的长id为:" + ctx.channel().id().asLongText());        log.info("客户端断开，channel对应的短id为:" + ctx.channel().id().asShortText());    }}

服务器启动

@SpringBootApplicationpublic class WebsocketApplication {   public static void main(String[] args) throws InterruptedException {      SpringApplication.run(WebsocketApplication.class, args);      new WebSocketServer(10101).run();   }}

客户端写一个html文件，代码如下

                                    发送消息:
                                接受消息:

打开浏览器如下

此时我们输入一段话，发送

我们可以看到所有打开的连接，都会收到相同的消息。

此时服务器的日志

2019-10-16 22:31:16.066 INFO 1376 --- [ntLoopGroup-3-5] c.g.w.netty.websocket.WebSocketHandler : 接收到的数据helloworld
2019-10-16 22:31:33.131 INFO 1376 --- [ntLoopGroup-3-5] c.g.w.netty.websocket.WebSocketHandler : 接收到的数据你好，中国

如果我们关闭一个页面，服务器的日志为

2019-10-16 22:36:39.390 INFO 1376 --- [ntLoopGroup-3-7] c.g.w.netty.websocket.WebSocketHandler : 客户端断开，channel对应的长id为:acde48fffe001122-00000560-00000007-9ca78ac7e2b907ab-0b15dfcb
2019-10-16 22:36:39.390 INFO 1376 --- [ntLoopGroup-3-7] c.g.w.netty.websocket.WebSocketHandler : 客户端断开，channel对应的短id为:0b15dfcb

现在我们来做一个点对点的聊天功能。

首先WebSocketServer添加空闲处理器

@Slf4j@AllArgsConstructorpublic class WebSocketServer {private int port;    public void run() throws InterruptedException {        EventLoopGroup boss = new NioEventLoopGroup();        EventLoopGroup worker = new NioEventLoopGroup();        try {            ServerBootstrap bootstrap = new ServerBootstrap();            bootstrap.group(boss,worker)                    .channel(NioServerSocketChannel.class)                    .option(ChannelOption.SO_BACKLOG,1024)                    .childOption(ChannelOption.TCP_NODELAY,true)                    .childHandler(new ChannelInitializer() {@Override                        protected void initChannel(SocketChannel ch) throws Exception {//web基于http协议的解码器                            ch.pipeline().addLast(new HttpServerCodec());                            //对大数据流的支持                            ch.pipeline().addLast(new ChunkedWriteHandler());                            //对http message进行聚合，聚合成FullHttpRequest或FullHttpResponse                            ch.pipeline().addLast(new HttpObjectAggregator(1024 * 64));                            //websocket服务器处理对协议，用于指定给客户端连接访问的路径                            //该handler会帮你处理一些繁重的复杂的事                            //会帮你处理握手动作：handshaking(close,ping,pong) ping + pong = 心跳                            //对于websocket来讲，都是以frames进行传输的，不同的数据类型对应的frames也不同                            ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws"));                            //针对客户端，如果1分钟时没有向服务端发送读写心跳(All),则主动断开                            //如果是读空闲或者是写空闲，不处理                            ch.pipeline().addLast(new IdleStateHandler(20,40,60));                            //添加我们的自定义channel处理器                            ch.pipeline().addLast(new WebSocketHandler());                        }                    });            log.info("服务器启动中");            ChannelFuture future = bootstrap.bind(port).sync();            future.channel().closeFuture().sync();        } finally {            worker.shutdownGracefully();            boss.shutdownGracefully();        }    }}

然后我们需要增加两个实体类--消息类和聊天类

首先添加一个消息的接口

public interface Chat {/**     * 保存消息到数据库     * @param chatMsg     */    public void save(Chat chatMsg);    /**     * 签名消息     * @param msgIdList     */    public void updateMsgSigned(List msgIdList);    /**     * 查找未读未签名消息     * @param acceptUserId     * @return     */    public List findUnReadChat(Long acceptUserId);}

添加一个消息的类来实现该接口

@ToString@Datapublic class ChatMsg implements Serializable,Chat{@JSONField(serializeUsing = ToStringSerializer.class)private Long senderId; //发送者的用户id    @JSONField(serializeUsing = ToStringSerializer.class)private Long receiverId; //接收者的用户id    private String msg;    @JSONField(serializeUsing = ToStringSerializer.class)private Long msgId; //用于消息的签收    private MsgSignFlagEnum signed; //消息签收状态    private LocalDateTime createDate;    @Override    @Transactional    public void save(Chat chatMsg) {        ChatDao chatDao = SpringBootUtil.getBean(ChatDao.class);        IdService idService = SpringBootUtil.getBean(IdService.class);        ((ChatMsg)chatMsg).setMsgId(idService.genId());        ((ChatMsg)chatMsg).setCreateDate(LocalDateTime.now());        chatDao.saveChat((ChatMsg) chatMsg);    }@Transactional    @Override    public void updateMsgSigned(List msgIdList) {        ChatDao chatDao = SpringBootUtil.getBean(ChatDao.class);        chatDao.updateMsgSigned(msgIdList);    }@Transactional    @Override    public List findUnReadChat(Long acceptUserId) {        ChatDao chatDao = SpringBootUtil.getBean(ChatDao.class);        return chatDao.findUnReadMsg(acceptUserId);    }}

其中MsgSignFlagEnum为一个枚举类型，代码如下

public enum MsgSignFlagEnum implements LocalisableAll {unsign(0,"未签收"),    signed(1,"已签收");    public final int type;    public final String value;    private MsgSignFlagEnum(int type,String value) {this.type = type;        this.value = value;    }@Override    public int getType() {return type;    }@Override    public String getValue() {return value;    }}

建立一个消息的工厂

public class ChatMsgFactory {public static Chat createChatMsgService() {return new ChatMsg();    }}

然后是聊天类

@Datapublic class DataContent implements Serializable {private Integer action; //动作类型    private ChatMsg chatMsg; //用户的聊天内容entity    private String extand; //扩展字段}

我们根据动作类型来定义一个枚举

public enum MsgActionEnum {CONNECT(1,"第一次(或重连)初始化连接"),    CHAT(2,"聊天消息"),    SIGNED(3,"消息签收"),    KEEPALIVE(4,"客户端保持心跳");    public final Integer type;    public final String content;    private MsgActionEnum(Integer type,String content) {this.type = type;        this.content = content;    }}

在写WebSocketHandler之前，我们需要将用户Id跟Channel做一个绑定

/** * 用户id和channel的关联关系的处理 */@Slf4jpublic class UserChannelRel {private static Map manager = new HashMap<>();    public static void put(Long senderId,Channel channel) {manager.put(senderId,channel);    }public static Channel get(Long senderId) {return manager.get(senderId);    }public static void output() {manager.entrySet().stream().forEach(entry ->log.info("UserId:" + entry.getKey() + ",ChannelId:" +                    entry.getValue().id().asLongText())        );    }}

最后WebSocketHandler改写如下

/** * TextWebSocketFrame: 在netty中，用于为websocket专门处理文本的对象，frame是消息的载体 */@Slf4jpublic class WebSocketHandler extends SimpleChannelInboundHandler {//用于记录和管理所有客户端的channel    private static ChannelGroup users = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);    private Chat chatMsgService = ChatMsgFactory.createChatMsgService();    @Override    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {//获取客户端传输过来的消息        String content = msg.text();        log.info("content为:" + content);        Channel currentChannel = ctx.channel();        //解析传输过来的消息转成聊天对象        DataContent dataContent = JSONObject.parseObject(content,DataContent.class);        //获取聊天对象的动作        Integer action = dataContent.getAction();        if (action == MsgActionEnum.CONNECT.type) {//当websocket第一次open的时候，初始化channel,把用的channel和userId关联起来            Long senderId = dataContent.getChatMsg().getSenderId();            UserChannelRel.put(senderId,currentChannel);            //测试            users.stream().forEach(channel -> log.info(channel.id().asLongText()));            UserChannelRel.output();        }else if (action == MsgActionEnum.CHAT.type) {//聊天类型的消息，把聊天记录保存到数据库，同时标记消息的签收状态[未签收]            ChatMsg chatMsg = dataContent.getChatMsg();            String msgText = chatMsg.getMsg();            Long receiverId = chatMsg.getReceiverId();            Long senderId = chatMsg.getSenderId();            //保存数据库            chatMsgService.save(chatMsg);            Channel receiverChannel = UserChannelRel.get(receiverId);            if (receiverChannel == null) {//接收方离线状态，此处无需处理            }else {                Channel findChannel = users.find(receiverChannel.id());                if (findChannel != null) {                    findChannel.writeAndFlush(new TextWebSocketFrame(                            JSONObject.toJSONString(chatMsg)                    ));                }else {//接收方离线，此处无需处理                }            }        }else if (action == MsgActionEnum.SIGNED.type) {//签收消息类型，针对具体的消息进行签收，修改数据库中对应消息的签收状态[已签收]            //扩展字段在signed类型的消息中，代表需要去签收的消息id,逗号间隔            String msgIdsStr = dataContent.getExtand();            log.info("extand为:" + msgIdsStr);            String[] msgIds = msgIdsStr.split(",");            List msgIdList = new ArrayList<>();            for (String mId : msgIds) {if (!StringUtils.isEmpty(mId)) {                    msgIdList.add(Long.valueOf(mId));                }            }log.info(msgIdList.toString());            if (!CollectionUtils.isEmpty(msgIdList)) {//批量签收                chatMsgService.updateMsgSigned(msgIdList);            }        }else if (action == MsgActionEnum.KEEPALIVE.type) {//心跳类型的消息            log.info("收到来自channel为[" + currentChannel + "]的心跳包");        }    }/**     * 当客户端连接服务端之后(打开连接)     * 获取客户端的channel,并且放到ChannelGroup中去进行管理     * @param ctx     * @throws Exception     */    @Override    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {users.add(ctx.channel());    }@Override    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {//当触发handlerRemoved,ChannelGroup会自动移除对应的客户端的channel        //所以下面这条语句可不写//        clients.remove(ctx.channel());        log.info("客户端断开，channel对应的长id为:" + ctx.channel().id().asLongText());        log.info("客户端断开，channel对应的短id为:" + ctx.channel().id().asShortText());    }@Override    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {        cause.printStackTrace();        ctx.channel().close();        users.remove(ctx.channel());    }@Override    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {//IdleStateEvent是一个用户事件，包含读空闲/写空闲/读写空闲        if (evt instanceof IdleStateEvent) {            IdleStateEvent event = (IdleStateEvent) evt;            if (event.state() == IdleState.READER_IDLE) {log.info("进入读空闲");            }else if (event.state() == IdleState.WRITER_IDLE) {log.info("进入写空闲");            }else if (event.state() == IdleState.ALL_IDLE) {log.info("channel关闭前，用户数量为:" + users.size());                //关闭无用的channel，以防资源浪费                ctx.channel().close();                log.info("channel关闭后，用户数量为:" + users.size());            }        }    }}

此处我们可以看到有两个地方注释了//接收方离线状态，此处无需处理。所以我们需要在用户上线的时候获取未签名的消息，只需要通过Controller从数据库获取就好

@RestControllerpublic class UnReadMessageController {   private Chat chatService = ChatMsgFactory.createChatMsgService();   @SuppressWarnings("unchecked")   @PostMapping("/notification-anon/getunreadmeg")   public Result> getUnReadMessage(@RequestParam("receiverid") Long receiverId) {      return Result.success(chatService.findUnReadChat(receiverId));   }}

由于我们登录用的是OAuth框架，在user模块添加

@PostMapping("/users-anon/finduser")public LoginAppUser findUserByName(@RequestParam("username") String username) {return appUserService.findByUsername(username);}

修改网关登录，登录后可以获取用户的信息，首先添加User模块的feign

@FeignClient("user-center")public interface UserClient {@PostMapping("/users-anon/finduser")    LoginAppUser findUserByName(@RequestParam("username") String username);}

/** * 登陆、刷新token、退出 * * @author 关键 */@Slf4j@RestControllerpublic class TokenController {@Autowired    private Oauth3Client oauth3Client;    @Autowired    private UserClient userClient;    /**     * 系统登陆
     * 根据用户名登录
     * 采用oauth3密码模式获取access_token和refresh_token     *     * @param username     * @param password     * @return     */    @SuppressWarnings("unchecked")@PostMapping("/sys/login")public Result login(@RequestParam String username,@RequestParam String password) {        Map parameters = new HashMap<>();        parameters.put(OAuth3Utils.GRANT_TYPE, "password");        parameters.put(OAuth3Utils.CLIENT_ID, "system");        parameters.put("client_secret", "system");        parameters.put(OAuth3Utils.SCOPE, "app");//    parameters.put("username", username);        // 为了支持多类型登录，这里在username后拼装上登录类型        parameters.put("username", username + "|" + CredentialType.USERNAME.name());        parameters.put("password", password);        Map tokenInfo = oauth3Client.postAccessToken(parameters);        AppUser user = userClient.findUserByName(username);        tokenInfo.put("user",user);        saveLoginLog(username, "用户名密码登陆");        return Result.success(tokenInfo);    }@Autowired    private LogClient logClient;    /**     * 登陆日志     *     * @param username     */    private void saveLoginLog(String username, String remark) {log.info("{}登陆", username);        // 异步        CompletableFuture.runAsync(() -> {try {                Log log = Log.builder().username(username).module(LogModule.LOGIN).remark(remark).createTime(new Date())                        .build();                logClient.save(log);            } catch (Exception e) {// do nothing            }        });    }/**     * 系统刷新refresh_token     *     * @param refresh_token     * @return     */    @PostMapping("/sys/refresh_token")public Map refresh_token(String refresh_token) {        Map parameters = new HashMap<>();        parameters.put(OAuth3Utils.GRANT_TYPE, "refresh_token");        parameters.put(OAuth3Utils.CLIENT_ID, "system");        parameters.put("client_secret", "system");        parameters.put(OAuth3Utils.SCOPE, "app");        parameters.put("refresh_token", refresh_token);        return oauth3Client.postAccessToken(parameters);    }/**     * 退出     *     * @param access_token     */    @GetMapping("/sys/logout")public void logout(String access_token, @RequestHeader(required = false, value = "Authorization") String token) {if (StringUtils.isBlank(access_token)) {if (StringUtils.isNoneBlank(token)) {                access_token = token.substring(OAuth3AccessToken.BEARER_TYPE.length() + 1);            }        }oauth3Client.removeToken(access_token);    }}

做如上修改后，我们登录后可以获得用户的Id,以上省略Dao,Mapper

这个时候服务端就全部完成了，我们再来看一下客户端

先建立一个全局变量app.js

window.app = {                /**         * 和后端的枚举对应         */        CONNECT: 1,                        //"第一次(或重连)初始化连接"),        CHAT: 2,                   //"聊天消息"),        SIGNED: 3,                 //"消息签收"),        KEEPALIVE: 4,              //"客户端保持心跳");        /**        * 和后端的ChatMsg聊天模型对象保持一致        * @param {Object} senderId        * @param {Object} receiverId        * @param {Object} msg        * @param {Object} msgId        */        ChatMsg: function(senderId,receiverId,msg,msgId) {                this.senderId = senderId;                this.receiverId = receiverId;                this.msg = msg;                this.msgId = msgId;        },        /**        * 构建消息模型对象        * @param {Object} action        * @param {Object} chatMsg        * @param {Object} extand        */        DataContent: function(action,chatMsg,extand) {                this.action = action;                this.chatMsg = chatMsg;                this.extand = extand;        }}

由于这里只是聊天样例，没有处理登录功能，我们现在就以1，2来代表两个用户的id

用户id为1的代码，文件名index.html

                                                                        发送消息:
                                                                接受消息:

用户id为2的代码,文件名receive.html

                                                                        发送消息:
                                                                接受消息:

这里都是通过Json来做数据的序列化的，之后会修改为ProtoBuffer来处理。

现在来增加发图片的功能。首先我们需要搭建好一个fastdfs服务器，具体可以参考分布式文件系统FastDFS安装配置以及Springboot 2.0+FastDFS开发配置。

这里只放入我们需要的Controller

@RestControllerpublic class FastDFSController {@Autowired    private FastDFSClientWrapper dfsClient;    @PostMapping("/files-anon/fdfsupload")@SuppressWarnings("unchecked")public Result upload(@RequestParam("file") MultipartFile file) throws Exception {        String imgUrl = dfsClient.uploadFile(file);        return Result.success(imgUrl);    }}

然后修改我们的用户id为1的前端代码

                                                                        发送消息:
                                                                                接受消息:

意思即为使用ajax访问我们的上传文件Controller，获取上传成功后的url，将该url拼接到的标签中，当称普通聊天信息发送出去即可。

用户id为2的前端代码相同。

                                                                        发送消息:
                                                                                接受消息:

看完上述内容是否对您有帮助呢？如果还想对相关知识有进一步的了解或阅读更多相关文章，请关注行业资讯频道，感谢您对的支持。

很赞哦！