阅读量:0
1.添加pom文件,引入Netty依赖
<dependency> <groupId>io.netty</groupId> <artifactId>netty-all</artifactId> <version>4.1.65.Final</version> </dependency> 2.添加ChannelHandlerContextMap类，用来存储通道的标识和通道对象
在长连接中，我们可以给每个通道附带一些属性，也可以说是标识，当你拿到这个通道对象时，可以获取这些属性，不过这里的前提是，需要先拿到通道对象，所以，我们无法通过通道标识去拿到通道对象，这样非常的不友好，例如，我们需要获取某个用户的通道，然后向这个通道推送消息，这个时候我们就只能循环每个通道，然后判断当前循环的通道是不是对应用户的，加了这个类，可以让我们快速的找到对应用户的通道，这里使用ConcurrentHashMap类，ConcurrentHashMap类是线程安全的。
import io.netty.channel.Channel; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; public class ChannelHandlerContextMap { private static ChannelHandlerContextMap instance; private final Map<String, Channel> channelMap; private ChannelHandlerContextMap() { channelMap = new ConcurrentHashMap<>(); } public static synchronized ChannelHandlerContextMap getInstance() { if (instance == null) { instance = new ChannelHandlerContextMap(); } return instance; } public void put(String identifier, Channel ctx) { channelMap.put(identifier, ctx); } public Channel get(String identifier) { return channelMap.get(identifier); } public Channel remove(String identifier) { return channelMap.remove(identifier); } public Boolean containsKey(String identifier) { return channelMap.containsKey(identifier); } public Set<Map.Entry<String, Channel>> entrySet() { return channelMap.entrySet(); } public int size() { return channelMap.size(); } } 3.添加HeartbeatHandler类，用来处理通道在一定时间内，和服务器没有消息往来的处理方案
import io.netty.channel.ChannelHandlerContext; import io.netty.handler.timeout.IdleState; import io.netty.handler.timeout.IdleStateEvent; import io.netty.handler.timeout.IdleStateHandler; public class HeartbeatHandler extends IdleStateHandler { public HeartbeatHandler(int readerIdleTimeSeconds, int writerIdleTimeSeconds, int allIdleTimeSeconds) { super(readerIdleTimeSeconds, writerIdleTimeSeconds, allIdleTimeSeconds); } @Override protected void channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt) throws Exception { String deviceNameAttr = ctx.channel().attr("deviceName").get(); if(!"Server".equals(deviceNameAttr)){ //LogUtil.LOGI(HeartbeatHandler.class.getName(),"客户端超时类型："+evt.state()+";通道标识："+deviceNameAttr); if (evt.state() == IdleState.READER_IDLE) { // 一段时间内没有收到过消息 LogUtil.LOGI(HeartbeatHandler.class.getName(),"客户端读取超时，关闭通道:"+deviceNameAttr); ctx.close(); } // if (evt.state() == IdleState.WRITER_IDLE) { // //一段时间内没有向外发送过消息 // LogUtil.LOGI(HeartbeatHandler.class.getName(),"客户端写入超时，关闭通道:"+deviceNameAttr); // //ctx.close(); // } // if (evt.state() == IdleState.ALL_IDLE) { // //一段时间内既没收到过消息也没发送过消息 // LogUtil.LOGI(HeartbeatHandler.class.getName(),"客户端超时，关闭通道:"+deviceNameAttr); // //ctx.close(); // } } } } 当一定时间内，通道和服务器没有消息往来时，会进入channelIdle方法，这里的一定时间，后面的代码中会设置；在channelIdle方法中，我们可以通过ctx.channel()来获取触发条件的通道，通过evt.state()来获取触发的类型，触发类型一共有三种：
IdleState.READER_IDLE：服务端一段时间内，没有收到客户端的消息，触发
IdleState.WRITER_IDLE：服务端一段时间内，没有向客户端发送消息，触发
IdleState.ALL_IDLE：服务端一段时间内，既没有收到客户端的消息，也没有向客户端发送消息，触发
可以在这里做自己的逻辑，我这边是直接关闭通道，防止通道占用过多
3.添加NettyHandler类，处理服务器收到的消息
import com.alibaba.fastjson.JSONArray; import com.alibaba.fastjson.JSONObject; import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.channel.group.ChannelGroup; import io.netty.channel.group.DefaultChannelGroup; import io.netty.util.Attribute; import io.netty.util.concurrent.GlobalEventExecutor; import org.springframework.stereotype.Component; import java.util.Map; import java.util.Set; @Component public class NettyHandler extends ChannelInboundHandlerAdapter { public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); @Override public void channelActive(ChannelHandlerContext ctx) { LogUtil.LOGI(NettyHandler.class.getName(),"客户端["+ctx.channel().remoteAddress()+"]已连接"); } @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { String deviceName = ctx.channel().attr("deviceName").get(); if(deviceName!=null && ChannelHandlerContextMap.getInstance().containsKey(deviceName)){ ChannelHandlerContextMap.getInstance().remove(deviceName); LogUtil.LOGI(NettyHandler.class.getName(),"客户端["+deviceName+"]断开连接"); }else{ LogUtil.LOGI(NettyHandler.class.getName(),"客户端["+ctx.channel().remoteAddress()+"]断开连接"); } } @Override public void channelRead(ChannelHandlerContext ctx, Object msg){ LogUtil.LOGI(NettyHandler.class.getName(),"源消息:"+message); } public void sendMessgae(Channel ctx,String message){ ctx.writeAndFlush(message); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); ctx.close(); } public String buildReturnPackage(int type,Object content){ JSONObject jsonObject = new JSONObject(); jsonObject.put("type",type); jsonObject.put("typeName",ConnectCodeUtil.MESSAGE_TYPE.get(type)); jsonObject.put("content",content); return jsonObject+"##@##"; } } channelActive:当客户端连接上服务端，会触发这个方法
channelInactive：当客户端断开服务端，会触发这个方法
exceptionCaught：当出现异常时，会触发这个方法
channelRead：当收到消息，会触发这个方法
我们需要在channelInactive、channelRead这两个方法中，管理好我们第二步创建的Map集合
在channelInactive方法中，可以看到，我并不是直接在Map中操作remove，是需要先判断该通道是否包含标识，存在标识我才remove，这和我们的业务逻辑相关，我们的业务逻辑存在客户端连上了服务端，不过还没有认证（就是认证时，会给通道打上标识，并且添加到Map中）的情况。这里根据你们的业务逻辑进行处理。
因为我这边的逻辑是，当存在同一个标识的通道出现时，我会将之前的通道给关闭，这个场景应该很常见，关于这个场景有一个问题，当这个场景出现，我会在channelRead方法中将Map中的通道覆盖，然后在channelInactive方法中移除Map中的通道数据，这样就会造成存在挤号时，Map中的数据是小于channelGroup中的数据的，这个你们根据自己的业务需求解决就好了
buildReturnPackage方法，是我构建发送消息的一个通用方法，消息最后的##@##是分隔符，后面会讲到。
4.添加NettyChannelInitializer类
import io.netty.buffer.Unpooled; import io.netty.channel.ChannelInitializer; import io.netty.channel.socket.SocketChannel; import io.netty.handler.codec.DelimiterBasedFrameDecoder; import io.netty.handler.codec.string.StringDecoder; import io.netty.handler.codec.string.StringEncoder; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; import io.netty.util.CharsetUtil; public class NettyChannelInitializer extends ChannelInitializer<SocketChannel> { @Override protected void initChannel(SocketChannel channel) throws Exception { //channel.pipeline().addLast(new LineBasedFrameDecoder(1024)); String delimiter = "##@##"; // 对服务端返回的消息通过_$进行分隔，并且每次查找的最大大小为1024字节 channel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, Unpooled.wrappedBuffer(delimiter.getBytes()))); // 将分隔之后的字节数据转换为字符串 channel.pipeline().addLast(new StringDecoder()); channel.pipeline().addLast(new StringEncoder(CharsetUtil.UTF_8)); channel.pipeline().addLast("heartbeatHandler", new HeartbeatHandler(90, 90, 90)); // 添加心跳处理器 channel.pipeline().addLast(new NettyHandler()); //channel.pipeline().addLast(new LoggingHandler(LogLevel.INFO)); } } 这个类就是将自定义的NettyHandler和HeartbeatHandler都添加进来，HeartbeatHandler的参数就是HeartbeatHandler类的构造方法的入参，分别是：
读取超时时间：IdleState.READER_IDLE
写入超时时间：IdleState.WRITER_IDLE
读写超时时间：IdleState.ALL_IDLE
单位秒
这里的##@##的定义分割符解码器，因为在Netty传输消息中，可能会存在粘包和拆包的问题，定义分隔符解码器就是，让Netty服务器将两个分隔符中间的文本，当做一条消息，从而解决粘包和拆包的问题。
解码器还有多种类型：
LineBasedFrameDecoder：定义一条消息已\r\n结尾
FixedLengthFrameDecoder：定义一条消息的固定长度
DelimiterBasedFrameDecoder：自定义分隔符
5.添加配置文件
server.port=8473 netty.port = 5555 netty.url= 127.0.0.1 我这里是.properties文件
6.添加启动类
import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioServerSocketChannel; import org.springframework.stereotype.Component; import java.net.InetSocketAddress; @Component public class NettyServer { // 开启服务端口 public static void start(InetSocketAddress address) { EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { ServerBootstrap bootstrap = new ServerBootstrap() .group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .localAddress(address) .childHandler(new NettyChannelInitializer()) .option(ChannelOption.SO_BACKLOG, 102400) .childOption(ChannelOption.SO_KEEPALIVE, true); // 绑定端口，开始接收进来的连接 ChannelFuture future = bootstrap.bind(address).sync(); LogUtil.LOGI(NettyServer.class.getName(),"开启Netty服务端口：" + address.getPort()); future.channel().closeFuture().sync(); } catch (Exception e) { e.printStackTrace(); bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } } 然后在springboot的启动类，添加下面代码，就能启动了
InetSocketAddress address = new InetSocketAddress(URL, PORT); server.start(address); 