Netty是建立在NIO基础之上,Netty在NIO之上又提供了更高层次的抽象。

在Netty里面,Accept连接可以使用单独的线程池去处理,读写操作又是另外的线程池来处理。

Accept连接和读写操作也可以使用同一个线程池来进行处理。而请求处理逻辑既可以使用单独的线程池进行处理,也可以跟放在读写线程一块处理。线程池中的每一个线程都是NIO线程。用户可以根据实际情况进行组装,构造出满足系统需求的并发模型。

Netty提供了内置的常用编解码器,包括行编解码器[一行一个请求],前缀长度编解码器[前N个字节定义请求的字节长度],可重放解码器[记录半包消息的状态],HTTP编解码器,WebSocket消息编解码器等等

Netty提供了一些列生命周期回调接口,当一个完整的请求到达时,当一个连接关闭时,当一个连接建立时,用户都会收到回调事件,然后进行逻辑处理。

Netty可以同时管理多个端口,可以使用NIO客户端模型,这些对于RPC服务是很有必要的。

Netty除了可以处理TCP Socket之外,还可以处理UDP Socket。

在消息读写过程中,需要大量使用ByteBuffer,Netty对ByteBuffer在性能和使用的便捷性上都进行了优化和抽象。

服务端:

package com.kenson.netty.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel; public class NettyServer {
/**
* 端口
*/
private int port; public NettyServer(int port) {
this.port = port;
} public void run() {
//EventLoopGroup是用来处理IO操作的多线程事件循环器
//负责接收客户端连接线程
EventLoopGroup bossGroup = new NioEventLoopGroup();
//负责处理客户端i/o事件、task任务、监听任务组
EventLoopGroup workerGroup = new NioEventLoopGroup();
//启动 NIO 服务的辅助启动类
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup);
//配置 Channel
bootstrap.channel(NioServerSocketChannel.class);
bootstrap.childHandler(new ServerIniterHandler());
//BACKLOG用于构造服务端套接字ServerSocket对象,
// 标识当服务器请求处理线程全满时,用于临时存放已完成三次握手的请求的队列的最大长度
bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
//是否启用心跳保活机制
bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
try {
//绑定服务端口监听
Channel channel = bootstrap.bind(port).sync().channel();
System.out.println("server run in port " + port);
//服务器关闭监听
/*channel.closeFuture().sync()实际是如何工作:
channel.closeFuture()不做任何操作,只是简单的返回channel对象中的closeFuture对象,对于每个Channel对象,都会有唯一的一个CloseFuture,用来表示关闭的Future,
所有执行channel.closeFuture().sync()就是执行的CloseFuturn的sync方法,从上面的解释可以知道,这步是会将当前线程阻塞在CloseFuture上*/
channel.closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
//关闭事件流组
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
} public static void main(String[] args) {
new NettyServer(8899).run();
}
}

服务端业务逻辑处理:

package com.kenson.netty.server;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor; public class ServerHandler extends SimpleChannelInboundHandler<String> { /**
* 所有的活动用户
*/
public static final ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); /**
* 读取消息通道
*
* @param context
* @param s
* @throws Exception
*/
@Override
protected void channelRead0(ChannelHandlerContext context, String s)
throws Exception {
Channel channel = context.channel();
//当有用户发送消息的时候,对其他的用户发送消息
for (Channel ch : group) {
if (ch == channel) {
ch.writeAndFlush("[you]: " + s + "\n");
} else {
ch.writeAndFlush("[" + channel.remoteAddress() + "]: " + s + "\n");
}
}
System.out.println("[" + channel.remoteAddress() + "]: " + s + "\n");
} /**
* 处理新加的消息通道
*
* @param ctx
* @throws Exception
*/
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
for (Channel ch : group) {
if (ch == channel) {
ch.writeAndFlush("[" + channel.remoteAddress() + "] coming");
}
}
group.add(channel);
} /**
* 处理退出消息通道
*
* @param ctx
* @throws Exception
*/
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
for (Channel ch : group) {
if (ch == channel) {
ch.writeAndFlush("[" + channel.remoteAddress() + "] leaving");
}
}
group.remove(channel);
} /**
* 在建立连接时发送消息
*
* @param ctx
* @throws Exception
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
boolean active = channel.isActive();
if (active) {
System.out.println("[" + channel.remoteAddress() + "] is online");
} else {
System.out.println("[" + channel.remoteAddress() + "] is offline");
}
ctx.writeAndFlush("[server]: welcome");
} /**
* 退出时发送消息
*
* @param ctx
* @throws Exception
*/
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
if (!channel.isActive()) {
System.out.println("[" + channel.remoteAddress() + "] is offline");
} else {
System.out.println("[" + channel.remoteAddress() + "] is online");
}
} /**
* 异常捕获
*
* @param ctx
* @param e
* @throws Exception
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) throws Exception {
Channel channel = ctx.channel();
System.out.println("[" + channel.remoteAddress() + "] leave the room");
ctx.close().sync();
} }

服务端处理器注册:

package com.kenson.netty.server;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder; public class ServerIniterHandler extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
//管道注册handler
ChannelPipeline pipeline = socketChannel.pipeline();
//编码通道处理
pipeline.addLast("decode", new StringDecoder());
//转码通道处理
pipeline.addLast("encode", new StringEncoder());
//聊天服务通道处理
pipeline.addLast("chat", new ServerHandler());
}
}

客户端:

package com.kenson.netty.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.commons.lang3.StringUtils; import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader; public class NettyClient { private String ip; private int port; private boolean stop = false; public NettyClient(String ip, int port) {
this.ip = ip;
this.port = port;
} public void run() throws IOException {
//设置一个多线程循环器
EventLoopGroup workerGroup = new NioEventLoopGroup();
//启动附注类
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(workerGroup);
//指定所使用的NIO传输channel
bootstrap.channel(NioSocketChannel.class);
//指定客户端初始化处理
bootstrap.handler(new ClientIniterHandler());
try {
//连接服务
Channel channel = bootstrap.connect(ip, port).sync().channel();
while (true) {
//向服务端发送内容
BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
String content = reader.readLine();
if (StringUtils.isNotEmpty(content)) {
if (StringUtils.equalsIgnoreCase(content, "q")) {
System.exit(1);
}
channel.writeAndFlush(content);
}
}
} catch (InterruptedException e) {
e.printStackTrace();
System.exit(1);
} finally {
workerGroup.shutdownGracefully();
}
} public static void main(String[] args) throws Exception {
new NettyClient("127.0.0.1", 8899).run();
}
}

客户端逻辑处理:

package com.kenson.netty.client;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler; public class ClientHandler extends SimpleChannelInboundHandler<String> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, String s) throws Exception {
//打印服务端的发送数据
System.out.println(s);
}
}

客户端处理器注册:

package com.kenson.netty.client;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder; public class ClientIniterHandler extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
//注册管道
ChannelPipeline pipeline = socketChannel.pipeline();
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
pipeline.addLast("http", new HttpClientCodec());
pipeline.addLast("chat", new ClientHandler());
}
}

测试时先启动服务端,再启动客户端。。。

声明: 本文借鉴网络资源,来自:https://www.cnblogs.com/kingsonfu/p/8635064.html

如果侵犯您的利益, 请联系本人删除

05-17 23:19
查看更多