当前位置: 首页 > >

springboot netty 搭建

发布时间:

maven


io.netty
netty-all
4.1.36.Final


废话不说代码奉上(感觉有用的点个赞可好)



import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
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 lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import javax.annotation.PreDestroy;
import java.net.InetSocketAddress;

/**
*


* 服务启动监听器
**/
@Slf4j
@Component//当成组件处理
@Order(value = 1)//这里表示启动顺序
public class NettyServer implements CommandLineRunner {

/**
* 配置服务端的NIO线程组
* NioEventLoopGroup 是用来处理I/O操作的Reactor线程组
* bossGroup:用来接收进来的连接,workerGroup:用来处理已经被接收的连接,进行socketChannel的网络读写,
* bossGroup接收到连接后就会把连接信息注册到workerGroup
* workerGroup的EventLoopGroup默认的线程数是CPU核数的二倍
*/
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
//new 一个工作线程组 专门负责网络读写操作
EventLoopGroup workGroup = new NioEventLoopGroup(200);

private Channel channel;

public void start(InetSocketAddress socketAddress) {
ChannelFuture future = null;
ServerBootstrap bootstrap = new ServerBootstrap()
.group(bossGroup, workGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ServerChannelInitializer())
.localAddress(socketAddress)
//设置队列大小
.option(ChannelOption.SO_BACKLOG, 2048)
//关闭延迟发送
.childOption(ChannelOption.TCP_NODELAY, true)
// 两小时内没有数据的通信时,TCP会自动发送一个活动探测数据报文
.childOption(ChannelOption.SO_KEEPALIVE, true);
try {
//绑定端口,开始接收进来的连接
future = bootstrap.bind(socketAddress).sync();
log.info("服务器启动开始监听端口: {}", socketAddress.getPort());
channel = future.channel();
} catch (InterruptedException e) {
e.printStackTrace();
}
}

@PreDestroy
public void destroy() {
if (channel != null) {
log.info("Netty Server close");
channel.close();
workGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
log.info("关闭成功");
}

public Channel getChannel() {
return channel;
}

@Async//注意这里,组件启动时会执行run,这个注解是让线程异步执行,这样不影响主线程
@Override
public void run(String... args) throws Exception {
start(new InetSocketAddress(6789));
}
}



import io.netty.buffer.ByteBuf;
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.timeout.IdleStateHandler;
import io.netty.util.CharsetUtil;


/**
*
* netty服务初始化器
**/
public class ServerChannelInitializer extends ChannelInitializer {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
// 这里使用的是自定义分隔符作为 ,防止 数据粘包问题 分包
ByteBuf delimiter = Unpooled.copiedBuffer(">###".getBytes());
socketChannel.pipeline()
.addLast(new DelimiterBasedFrameDecoder(1024*1024,delimiter))
//三十秒没有收到消息 将IdleStateHandler 添加到 ChannelPipeline 中
.addLast(new IdleStateHandler(30, 0, 0))//(8)
// 往pipeline链中添加一个编码器,字符串->二进制
.addLast("encoder", new StringEncoder(CharsetUtil.US_ASCII))
// 往pipeline链中添加一个解码器
.addLast("decoder", new StringDecoder(CharsetUtil.US_ASCII))
.addLast(new NettyServerHandler())
;
}
}


import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelId;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import java.net.InetSocketAddress;
import java.util.concurrent.ConcurrentHashMap;

import static io.netty.handler.codec.stomp.StompHeaders.HEART_BEAT;

/**
*


* netty服务端处理器
**/

@Slf4j
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

/**
* 管理一个全局map,保存连接进服务端的通道数量
*/
private static final ConcurrentHashMap CHANNEL_MAP = new ConcurrentHashMap<>();



public void sendOne(){

}

/**
* 客户端连接会触发
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
String clientIp = insocket.getAddress().getHostAddress();
int clientPort = insocket.getPort();
//获取连接通道唯一标识
ChannelId channelId = ctx.channel().id();
System.out.println();
//如果map中不包含此连接,就保存连接
if (CHANNEL_MAP.containsKey(channelId)) {
log.info("客户端【" + channelId + "】是连接状态,连接通道数量: " + CHANNEL_MAP.size());
} else {
//保存连接
CHANNEL_MAP.put(channelId, ctx);
log.info("客户端【" + channelId + "】连接netty服务器[IP:" + clientIp + "--->PORT:" + clientPort + "]");
log.info("连接通道数量: " + CHANNEL_MAP.size());
}

}


@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
delectChannel(ctx);
}



/**
* 客户端发消息会触发
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println();
log.info("加载客户端报文......");
log.info("【" + ctx.channel().id() + "】" + " :" + msg);

//响应客户端
this.channelWrite(ctx.channel().id(), msg);
}



/**
* @param msg 需要发送的消息内容
* @param channelId 连接通道唯一id
* @author xiongchuan on 2019/4/28 16:10
* @DESCRIPTION: 服务端给客户端发送消息
* @return: void
*/
public void channelWrite(ChannelId channelId, Object msg) throws Exception {
ChannelHandlerContext ctx = CHANNEL_MAP.get(channelId);
if (ctx == null) {
log.info("通道【" + channelId + "】不存在");
return;
}
//将客户端的信息直接返回写入ctx
ctx.write(msg);
//刷新缓存区
ctx.flush();
}



/**
* 超时处理
*
* @param ctx
* @param evt
* @throws Exception
*/
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
String socketString = ctx.channel().remoteAddress().toString();
if (evt instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent) evt;
if (event.state() == IdleState.READER_IDLE) {
log.info("Client: " + socketString + " READER_IDLE 读超时");
this.channelWrite(ctx.channel().id(), "超时服务器断开".getBytes());
ctx.disconnect();
ctx.writeAndFlush(HEART_BEAT).addListener(ChannelFutureListener.CLOSE_ON_FAILURE) ;
} else if (event.state() == IdleState.WRITER_IDLE) {
log.info("Client: " + socketString + " WRITER_IDLE 写超时");
this.channelWrite(ctx.channel().id(), "超时服务器断开".getBytes());
ctx.disconnect();
} else if (event.state() == IdleState.ALL_IDLE) {
log.info("Client: " + socketString + " ALL_IDLE 总超时");
this.channelWrite(ctx.channel().id(), "超时服务器断开".getBytes());
ctx.disconnect();
}
}
}


/**
* 发生异常触发
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
// 当出现异常就关闭连接
log.info(ctx.channel().id() + " 发生了错误,此连接被关闭" + "此时连通数量: " + CHANNEL_MAP.size());
delectChannel(ctx);
ctx.close();
}

/**
* 断开移除
* @param ctx
*/
public void delectChannel(ChannelHandlerContext ctx){
InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
String clientIp = insocket.getAddress().getHostAddress();
ChannelId channelId = ctx.channel().id();
//包含此客户端才去删除
if (CHANNEL_MAP.containsKey(channelId)) {
//删除连接
CHANNEL_MAP.remove(channelId);
log.info("客户端【" + channelId + "】退出netty服务器[IP:" + clientIp + "--->PORT:" + insocket.getPort() + "]");
log.info("连接通道数量: " + CHANNEL_MAP.size());
}
}

}



友情链接: