netty的核心架构

官网给出的底层示意图：

项目结构

• 代码结构（简单的springboot + maven项目）

• pom文件（netty版本 - 4.1.43.Final）

• 代码展示

WebSocketServer服务类

package com.netty.websocket.netty;import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
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.stream.ChunkedWriteHandler;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.TimeUnit;/**
* @ClassName: WebSocketServer
* @Description: WebSocketServer
* @author yuyehong
* @date 2020年4月29日 上午1:22:32
* @version v1.3.0
* @since 1.8
*/
@Component
@Slf4j
public class WebSocketServer {/*** webSocket协议名*/private static final String WEBSOCKET_PROTOCOL = "WebSocket";/*** 端口号*/@Value("${netty.server.port:9005}")private int port;@Value("${netty.server.path:/home}")private String path;private EventLoopGroup bossGroup;private EventLoopGroup workGroup;/*** 启动* @throws InterruptedException*/private void start() throws InterruptedException {// bossGroup就是parentGroup，是负责处理TCP/IP连接的bossGroup = new NioEventLoopGroup();// workerGroup就是childGroup,是负责处理Channel(通道)的I/O事件workGroup = new NioEventLoopGroup();ServerBootstrap sb = new ServerBootstrap();sb.option(ChannelOption.SO_BACKLOG, 1024);sb.group(workGroup, bossGroup).channel(NioServerSocketChannel.class).localAddress(this.port).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) {log.info("收到新连接：" + ch.localAddress());// HttpServerCodec：将请求和应答消息解码为HTTP消息ch.pipeline().addLast(new HttpServerCodec());// ChunkedWriteHandler：向客户端发送HTML5文件ch.pipeline().addLast(new ChunkedWriteHandler());// HttpObjectAggregator：将HTTP消息的多个部分合成一条完整的HTTP消息ch.pipeline().addLast(new HttpObjectAggregator(8192));ch.pipeline().addLast(new WebSocketServerProtocolHandler(path, WEBSOCKET_PROTOCOL, true, 65536 * 10));// 进行设置心跳检测ch.pipeline().addLast(new IdleStateHandler(60, 30, 60 * 30, TimeUnit.SECONDS));// 配置通道处理 来进行业务处理ch.pipeline().addLast(new WebSocketHandler());}});// 配置完成，开始绑定server，通过调用sync同步方法阻塞直到绑定成功ChannelFuture channelFuture = sb.bind(port).sync();log.info("【Netty服务启动成功========端口：" + port + "】");log.info("Server started and listen on:{}", channelFuture.channel().localAddress());// 成功绑定到端口之后,给channel增加一个 管道关闭的监听器并同步阻塞,直到channel关闭,线程才会往下执行,结束进程。channelFuture.channel().closeFuture().sync();}/*** 释放资源* @throws InterruptedException*/@PreDestroypublic 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

个人使用redis，将连接用户信息缓存

package com.netty.websocket.netty;import com.alibaba.fastjson.JSON;
import com.netty.websocket.common.enums.CommonConstant;
import com.netty.websocket.common.enums.PushTypeEnums;
import com.netty.websocket.common.enums.WsSubTypeEnum;
import com.netty.websocket.common.javabean.WsRespVo;
import com.netty.websocket.common.util.RedisUtil;
import com.netty.websocket.common.util.SpringUtil;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.UUID;/*** @ClassName: WebSocketHandler* @Description: TextWebSocketFrame类型， 表示一个文本帧* @author yuyehong* @date 2020年4月29日 上午1:22:16* @version v1.3.0* @since 1.8*/
@Component
@Slf4j
public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {/*** 客户端与服务器建立连接时触发*/@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {log.info("与客户端建立连接，通道开启！channelActive 被调用" + ctx.channel().id().asLongText());// 添加到channelGroup 通道组NettyConfig.addChannel(ctx.channel());}/*** 客户端与服务器关闭连接时触发*/@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {String channelLongId = ctx.channel().id().asLongText();log.info("channelInactive 被调用" + channelLongId);// 删除通道NettyConfig.removeChannel(ctx.channel());AttributeKey<String> key = AttributeKey.valueOf("wsSubType");String wsSubType = ctx.channel().attr(key).get();WsSubTypeEnum type = WsSubTypeEnum.getByKey(Integer.parseInt(wsSubType));switch (type) {case ONE:hdelRedisCache("user", channelLongId);break;default:log.info("未知来源");break;}}/*** 服务器接收客户端的数据消息*/@Overrideprotected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {log.info("服务器收到消息：{}", msg.text());String channelLongId = ctx.channel().id().asLongText();ReceiveDataContent receive = JSON.parseObject(msg.text(), ReceiveDataContent.class);// 用户IDString userId = receive.getUserId();if (StringUtils.isBlank(userId)) {log.info("用户ID为空，不记录连接信息");return;}Integer subType = receive.getSubType();// 订阅类型WsSubTypeEnum type = WsSubTypeEnum.getByKey(subType);switch (type) {case ONE:// 总首页存用户IDhSetRedisCache("user", channelLongId, userId);log.info("通道ID：{}", channelLongId);break;default:log.info("未知来源");break;}// 将socket订阅类型作为自定义属性加入到channel中，方便随时channel中获取//也可以将用户ID绑定到属性中去AttributeKey<String> key = AttributeKey.valueOf("wsSubType");ctx.channel().attr(key).setIfAbsent(String.valueOf(subType));// 回复消息ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(buildWsResp("连接成功", PushTypeEnums.USER_IDENTIFICATION))));}/*** 删除redis缓存* @Title: hdelRedisCache* @author yuyehong* @date 2020年4月30日 上午10:42:09* @param key  key* @param item item*/private void hdelRedisCache(String key, Object... item) {RedisUtil redisUtil = SpringUtil.getBean(RedisUtil.class);redisUtil.hdel(key, item);}/*** 缓存redis* @Title: hSetRedisCache* @author yuyehong@cloudwalk.cn* @date 2020年4月30日 上午10:43:38* @param key   key* @param item  item* @param value value*/private void hSetRedisCache(String key, String item, Object value) {RedisUtil redisUtil = SpringUtil.getBean(RedisUtil.class);redisUtil.hset(key, item, value);}private WsRespVo buildWsResp(Object msg, PushTypeEnums type) {WsRespVo wsRespVo = WsRespVo.builder().id(UUID.randomUUID().toString()).date(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").format(LocalDateTime.now())).typeCode(type.getKey()).type(type.getValue()).message(type.getRemark()).code(CommonConstant.CW_STATUS_CODE_SUCCESS).data(msg).build();return wsRespVo;}
}

自定义信息接受类ReceiveDataContent

package com.netty.websocket.netty;import lombok.Data;import java.io.Serializable;/*** @ClassName: ReceiveDataContent* @Description: netty用于接受的信息* @author yuyehong* @date 2020/4/2818:31* @version V1.3.0* @since 1.8*/
@Data
public class ReceiveDataContent implements Serializable {private static final long serialVersionUID = 1155902834663674240L;/*** 用户ID*/private String userId;/*** 操作类型(默认为 1 用户认证)*/private Integer operaType = 1;/*** 消息*/private String message;/*** 订阅类型*/private Integer subType = 1;
}

NettyConfig常量类

package com.netty.websocket.netty;import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;import java.util.concurrent.ConcurrentHashMap;/*** @ClassName: NettyConfig* @Description:netty常量类* @author yuyehong* @date 2020年4月29日 上午1:21:52* @version v1.3.0* @since 1.8*/
public class NettyConfig {/*** 定义一个channel组，管理所有的channel GlobalEventExecutor.INSTANCE 是全局的事件执行器，是一个单例*/private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);/*** 存放通道和通道对应信息，用户寻找指定通道信息*/private static ConcurrentHashMap<String, Channel> channelMap = new ConcurrentHashMap<>();private NettyConfig() {}/*** 获取channel组* @return*/public static ChannelGroup getChannelGroup() {return channelGroup;}public static void addChannel(Channel channel) {channelGroup.add(channel);channelMap.put(channel.id().asLongText(), channel);}public static void removeChannel(Channel channel) {channelGroup.remove(channel);channelMap.remove(channel.id().asLongText());}public static Channel findChannel(String id) {return channelMap.get(id);}public static void send2All(TextWebSocketFrame tws) {channelGroup.writeAndFlush(tws);}}

消息推送示例：