这篇文章起了一个很牛b的名字，实际要讲的内容很简单。但是还是发现很多人把这个功能写复杂了。netty的服务端网络编程，按照官方提供的demo，稍加修改即可，但是一些参数选项，需要自己去完善设置。而数据分发功能，就是面向所有连接分发数据，很多人的做法是使用java concurrent包下的相关容器保存连接，然后需要分发数据时，遍历集合中的元素，一个一个的调用writeAndFlush()将数据发出去。其实可以更简单一些，使用线程安全的ChannelGroup保存连接并分发数据。性能提高多少，我没有对比过，至少代码看着简化了。下面是相关实现代码，实际生产环境中，下面代码应该还有很多很多需要优化的地方，目前我还没有在超过10k并发连接情况下测试过下面的代码。
public class NetServer implements Runnable {
private static Logger LOGGER = LoggerFactory.getLogger(NetServer.class);
private final static int PORT = 9500;
private final static int MAX_MESSAGE_LENGTH = 8192;
private final ChannelGroup channels_ = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
public NetServer() {
}
// 分发数据
public void dispatcher(String message) {
channels_.writeAndFlush(message);
}
@Override
public void run() {
startup();
}
private void startup() {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup);
b.channel(NioServerSocketChannel.class);
b.option(ChannelOption.SO_BACKLOG, 128);
b.option(ChannelOption.TCP_NODELAY, true);
b.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(new LengthFieldBasedFrameDecoder(MAX_MESSAGE_LENGTH, 0, 4, 0 ,4));
p.addLast(new LengthFieldPrepender(4));
p.addLast(new StringDecoder(CharsetUtil.UTF_8));
p.addLast(new StringEncoder(CharsetUtil.UTF_8));
p.addLast(new NetServerHandler(channels_));
}
});
ChannelFuture future = b.bind(PORT).sync();
future.addListener(new GenericFutureListener<Future<? super Void>>() {
@Override
public void operationComplete(Future<? super Void> future) throws Exception {
if (future.isSuccess()) {
LOGGER.info("服务器启动成功...");
} else {
LOGGER.info("服务器启动失败...");
if (future.cause() != null) {
LOGGER.error("异常信息: " + future.cause().getMessage());
}
}
}
});
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
LOGGER.error("服务器启动出现异常..." + e.getMessage());
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
public class NetServerHandler extends SimpleChannelInboundHandler<String> {
private ChannelGroup channels_;
private static Logger LOGGER = LoggerFactory.getLogger(NetServerHandler.class);
public NetServerHandler(ChannelGroup channels) {
channels_ = channels;
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
channels_.add(ctx.channel());
LOGGER.info("{} is up...当前连接数量: {}", ctx.channel().remoteAddress(), channels_.size());
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
ctx.close();
LOGGER.info("{} is down...当前连接数量: {}", ctx.channel().remoteAddress(), channels_.size());
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
LOGGER.error("与客户端 {} 连接出现异常, 异常信息: " + cause.getMessage(), ctx.channel().remoteAddress());
}
}
---------------------
作者：grafx
来源：CSDN
原文：/grafx/article/details/56677667
版权声明：本文为博主原创文章，转载请附上博文链接！
