SpringBoot+Netty+WebSocket,服务端、JS页面、客户端互发消息（附项目源码）

Netty简介

    Netty是基于Java NIO client-server的网络应用框架，使用Netty可以快速开发网络应用，例如服务器和客户端协议。Netty提供了一种新的方式来开发网络应用程序，这种新的方式使它很容易使用和具有很强的扩展性。Netty的内部实现是很复杂的，但是Netty提供了简单易用的API从网络处理代码中解耦业务逻辑。Netty是完全基于NIO实现的，所以整个Netty都是异步的。

项目源码：百度网盘 请输入提取码    提取码: 4e7u

1.服务端

步骤一：pom文件引入

io.netty netty-all 4.1.33.Final cn.hutool hutool-all 5.2.3 com.alibaba fastjson 1.2.76

步骤二：NettyConfig

package com.example.nettybackstage.config;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 { private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); private static ConcurrentHashMap userChannelMap = new ConcurrentHashMap<>(); private NettyConfig() {} public static ChannelGroup getChannelGroup() { return channelGroup; } public static ConcurrentHashMap getUserChannelMap(){ return userChannelMap; }}

步骤三：NettyServer

package com.example.nettybackstage.config;import io.netty.bootstrap.ServerBootstrap;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelInitializer;import io.netty.channel.EventLoopGroup;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioServerSocketChannel;import io.netty.handler.codec.http.HttpObjectAggregator;import io.netty.handler.codec.http.HttpServerCodec;import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;import io.netty.handler.codec.serialization.ObjectEncoder;import io.netty.handler.stream.ChunkedWriteHandler;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.beans.factory.annotation.Value;import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;import javax.annotation.PreDestroy;import java.net.InetSocketAddress;@Componentpublic class NettyServer{ private static final Logger log = LoggerFactory.getLogger(NettyServer.class); @Value("${webSocket.protocol}") private String webSocketProtocol; @Value("${webSocket.netty.port}") private int port; @Value("${webSocket.netty.path}") private String webSocketPath; @Autowired private WebSocketHandler webSocketHandler; private EventLoopGroup bossGroup; private EventLoopGroup workGroup; private void start() throws InterruptedException { //数据量上来时设置线程池 bossGroup = new NioEventLoopGroup(); workGroup = new NioEventLoopGroup(); ServerBootstrap bootstrap = new ServerBootstrap(); // bossGroup辅助客户端的tcp连接请求, workGroup负责与客户端之前的读写操作 bootstrap.group(bossGroup,workGroup); // 设置NIO类型的channel bootstrap.channel(NioServerSocketChannel.class); // 设置监听端口 bootstrap.localAddress(new InetSocketAddress(port)); // 连接到达时会创建一个通道 bootstrap.childHandler(new ChannelInitializer() { @Override protected void initChannel(SocketChannel ch) throws Exception { // 流水线管理通道中的处理程序（Handler），用来处理业务 // webSocket协议本身是基于http协议的，所以这边也要使用http编解码器 ch.pipeline().addLast(new HttpServerCodec()); ch.pipeline().addLast(new ObjectEncoder()); // 以块的方式来写的处理器 ch.pipeline().addLast(new ChunkedWriteHandler()); ch.pipeline().addLast(new HttpObjectAggregator(8192)); ch.pipeline().addLast(new WebSocketServerProtocolHandler(webSocketPath, webSocketProtocol, true, 65536 * 10)); // 自定义的handler，处理业务逻辑 ch.pipeline().addLast(webSocketHandler); } }); // 配置完成，开始绑定server，通过调用sync同步方法阻塞直到绑定成功 ChannelFuture channelFuture = bootstrap.bind().sync(); log.info("Server started and listen on:{}",channelFuture.channel().localAddress()); // 对关闭通道进行监听 channelFuture.channel().closeFuture().sync(); } @PreDestroy public void destroy() throws InterruptedException { if(bossGroup != null){ bossGroup.shutdownGracefully().sync(); } if(workGroup != null){ workGroup.shutdownGracefully().sync(); } } @PostConstruct() public void init() { //需要开启一个新的线程来执行netty server 服务器 new Thread(() -> { try { start(); } catch (InterruptedException e) { e.printStackTrace(); } }).start(); }}

步骤四：WebSocketHandler

package com.example.nettybackstage.config;import cn.hutool.json.JSONObject;import cn.hutool.json.JSONUtil;import io.netty.channel.ChannelHandler;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.SimpleChannelInboundHandler;import io.netty.handler.codec.http.websocketx.TextWebSocketframe;import io.netty.util.AttributeKey;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.stereotype.Component;import org.springframework.util.ObjectUtils;import java.util.HashMap;import java.util.Map;@Component@ChannelHandler.Sharablepublic class WebSocketHandler extends SimpleChannelInboundHandler { private static final Logger log = LoggerFactory.getLogger(WebSocketHandler.class); //标识 private static final String CHANNEL_TYPE = "netty_msg"; @Override public void handlerAdded(ChannelHandlerContext ctx) throws Exception { log.info("handlerAdded 被调用:{}",ctx.channel().id().asLongText()); // 添加到channelGroup 通道组 NettyConfig.getChannelGroup().add(ctx.channel()); } @Override public void channelRead(ChannelHandlerContext ctx, Object msgObj) throws Exception { // 1、获取客户端发送的消息 try { TextWebSocketframe msg = (TextWebSocketframe) msgObj; if (!ObjectUtils.isEmpty(msg.text())){ Map map = new HashMap<>(); map = com.alibaba.fastjson.JSONObject.parseObject(msg.text(),Map.class); String typeId = String.valueOf(map.get("typeId")); switch (typeId){ //netty_msg_first类型ID:首次建立连接管道信息 case "netty_msg_first": log.info("(类型ID:netty_msg_first)服务器收到消息连接成功：{}",msg.text()); // 获取CHANNEL_TYPE,关联channel(自定义发送类型) NettyConfig.getUserChannelMap().put(CHANNEL_TYPE,ctx.channel()); // 将用type作为自定义属性加入到channel中，方便随时channel中获取type AttributeKey key = AttributeKey.valueOf("type"); ctx.channel().attr(key).setIfAbsent(CHANNEL_TYPE); break; //netty_web_20220204类型ID:接收前端发送的消息类型 case "netty_web_20220204": log.info("(类型ID:netty_web_20220204)服务器收到前端发送消息：{}",msg.text()); break; //netty_web_20220204类型ID:接收客户端发送的消息类型 case "netty_client_20220204": log.info("(类型ID:netty_client_20220204)服务器收到前端发送消息：{}",msg.text()); break; } } }catch (Exception e){ log.error(e.getMessage()); e.printStackTrace(); } } @Override protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketframe textWebSocketframe) throws Exception { } @Override public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { log.info("handlerRemoved 被调用:{}",ctx.channel().id().asLongText()); // 删除通道 NettyConfig.getChannelGroup().remove(ctx.channel()); removeUserId(ctx); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { log.info("异常：{}",cause.getMessage()); // 删除通道 NettyConfig.getChannelGroup().remove(ctx.channel()); removeUserId(ctx); ctx.close(); } private void removeUserId(ChannelHandlerContext ctx){ AttributeKey key = AttributeKey.valueOf("type"); String userId = ctx.channel().attr(key).get(); NettyConfig.getUserChannelMap().remove(userId); }}

步骤五：MessageDto

package com.example.nettybackstage.dto;import java.io.Serializable;import java.util.Map;public class MessageDto implements Serializable { public MessageDto(){} public MessageDto(String nettyId,String type,String status,String msg,Map objectMap,String typeId,String date){ this.nettyId = nettyId; this.type = type; this.status = status; this.msg = msg; this.objectMap = objectMap; this.typeId = typeId; this.date = date; } private String nettyId; private String type; private String status; private String msg; private String typeId; private String date; public String getDate() { return date; } public void setDate(String date) { this.date = date; } public String getTypeId() { return typeId; } public void setTypeId(String typeId) { this.typeId = typeId; } private Map objectMap; public String getNettyId() { return nettyId; } public void setNettyId(String nettyId) { this.nettyId = nettyId; } public String getType() { return type; } public void setType(String type) { this.type = type; } public String getStatus() { return status; } public void setStatus(String status) { this.status = status; } public String getMsg() { return msg; } public void setMsg(String msg) { this.msg = msg; } public Map getObjectMap() { return objectMap; } public void setObjectMap(Map objectMap) { this.objectMap = objectMap; }}

步骤六：PushService

package com.example.nettybackstage.service;import com.example.nettybackstage.dto.MessageDto;public interface PushService { public void sendMsgToOne(String type, MessageDto msg); public void sendMsgToAll(MessageDto msg);}

步骤七：PushServiceImpl

package com.example.nettybackstage.serviceimpl;import com.alibaba.fastjson.JSONObject;import com.example.nettybackstage.config.NettyConfig;import com.example.nettybackstage.config.WebSocketHandler;import com.example.nettybackstage.dto.MessageDto;import com.example.nettybackstage.service.PushService;import io.netty.channel.Channel;import io.netty.handler.codec.http.websocketx.TextWebSocketframe;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.stereotype.Service;import java.util.concurrent.ConcurrentHashMap;@Servicepublic class PushServiceImpl implements PushService { private static final Logger log = LoggerFactory.getLogger(PushServiceImpl.class); @Override public void sendMsgToOne(String type, MessageDto msg){ ConcurrentHashMap userChannelMap = NettyConfig.getUserChannelMap(); Channel channel = userChannelMap.get(type); String jsonMsg = JSONObject.toJSonString(msg); channel.writeAndFlush(new TextWebSocketframe(jsonMsg)); log.info("pushMsgToOne消息发送成功={}",jsonMsg); } @Override public void sendMsgToAll(MessageDto msg){ String jsonMsg = JSONObject.toJSonString(msg); NettyConfig.getChannelGroup().writeAndFlush(new TextWebSocketframe(jsonMsg)); log.info("pushMsgToAll消息发送成功={}",msg); }}

步骤八：PushController

package com.example.nettybackstage.web;import com.example.nettybackstage.dto.MessageDto;import com.example.nettybackstage.serviceimpl.PushServiceImpl;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.web.bind.annotation.PostMapping;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RequestParam;import org.springframework.web.bind.annotation.RestController;import java.util.HashMap;import java.util.Map;import java.util.UUID;@RestController@RequestMapping("/push")public class PushController { @Autowired private PushServiceImpl pushService; @PostMapping("/sendMsgToAll") public void sendMsgToAll(@RequestParam("msg") String msg){ MessageDto messageDto = new MessageDto(); String uuid = String.valueOf(UUID.randomUUID()).replace("-",""); messageDto.setNettyId(uuid); messageDto.setMsg(msg); messageDto.setStatus("0000"); messageDto.setType("netty_msg"); messageDto.setTypeId(""); messageDto.setDate("2021-12-05 21:19:10"); Map map = new HashMap<>(); map.put("id",uuid); map.put("name","王五"); map.put("number","A-1257564246"); messageDto.setObjectMap(map); pushService.sendMsgToAll(messageDto); } @PostMapping("/sendMsgToOne") public void sendMsgToOne(@RequestParam("name") String name,@RequestParam("msg") String msg,@RequestParam("typeId") String typeId){ MessageDto messageDto = new MessageDto(); String uuid = String.valueOf(UUID.randomUUID()).replace("-",""); messageDto.setNettyId(uuid); messageDto.setMsg(msg); messageDto.setStatus("0000"); messageDto.setType("netty_msg"); messageDto.setTypeId(typeId); messageDto.setDate("2022-02-05 15:09:00"); Map map = new HashMap<>(); map.put("id",uuid); map.put("name",name); map.put("number","1257564246"); messageDto.setObjectMap(map); pushService.sendMsgToOne(messageDto.getType(),messageDto); }}

步骤十：application.properties

server.port=8081#netty配置#连接端口webSocket.netty.port=58080#协定名称(可定义)webSocket.protocol=webSocket#路径webSocket.netty.path=/${webSocket.protocol}

2.页面客户端

步骤一

3.客户端

步骤一:pom

org.java-websocket Java-WebSocket 1.5.1 com.alibaba fastjson 1.2.76

步骤二：WebSocketListener

package com.example.nettyclient.listener;import com.alibaba.fastjson.JSONObject;import lombok.extern.slf4j.Slf4j;import org.java_websocket.client.WebSocketClient;import org.java_websocket.drafts.Draft_6455;import org.java_websocket.handshake.ServerHandshake;import org.springframework.context.annotation.Bean;import org.springframework.stereotype.Component;import org.springframework.util.ObjectUtils;import java.net.URI;import java.util.HashMap;import java.util.Map;@Slf4j@Componentpublic class WebSocketListener { @Bean public WebSocketClient webSocketClient() { try { WebSocketClient webSocketClient = new WebSocketClient(new URI("ws://127.0.0.1:58080/webSocket"),new Draft_6455()) { @Override public void onOpen(ServerHandshake handshakedata) { log.info("[websocket] 连接成功"); Map map = new HashMap<>(); map.put("typeId","netty_msg_first"); String jsonMsg = JSONObject.toJSonString(map); this.send(jsonMsg); log.info("[websocket] 回发连接成功消息!"); } @Override public void onMessage(String message) { log.info("[websocket] 收到消息={}",message); if (ObjectUtils.isEmpty(message)){ return; } Map map = JSONObject.parseObject(message,Map.class); String typeId = String.valueOf(map.get("typeId")); switch (typeId){ case "netty_client_20220204": log.info("[websocket] 收到类型ID为={},消息={}",typeId,message); break; } } @Override public void onClose(int code, String reason, boolean remote) { log.info("[websocket] 退出连接"); } @Override public void onError(Exception ex) { log.info("[websocket] 连接错误={}",ex.getMessage()); } }; webSocketClient.connect(); return webSocketClient; } catch (Exception e) { e.printStackTrace(); } return null; }}

4.服务端和页面客户端互发消息

启动服务端和页面客户端,启动后访问页面:http://127.0.0.1:8082/index.html,如下:

 

 用postman发送一条消息

页面成功接收到服务端即时发送的消息，同时页面也向服务端回发一条消息

5.服务端和客户端互相发送消息

启动服务

用postman调用服务端接口发送一条消息

 成功接收到服务端消息