沉淀再出发:关于netty的一些理解和使用

沉淀再出发:关于netty的一些理解和使用

一、前言

Netty是由JBOSS提供的一个java开源框架。Netty提供异步的、事件驱动的网络应用程序框架和工具,用以快速开发高性能、高可靠性的网络服务器和客户端程序。也就是说,Netty 是一个基于NIO的客户、服务器端编程框架,使用Netty 可以确保你快速和简单的开发出一个网络应用,例如实现了某种协议的客户、服务端应用。Netty相当于简化和流线化了网络应用的编程开发过程,例如:基于TCP和UDP的socket服务开发。“快速”和“简单”并不用产生维护性或性能上的问题。Netty 是一个吸收了多种协议(包括FTP、SMTP、HTTP等各种二进制文本协议)的实现经验,并经过相当精心设计的项目。最终,Netty 成功的找到了一种方式,在保证易于开发的同时还保证了其应用的性能,稳定性和伸缩性。netty在底层的数据通信和封装之中有着重要的作用,下面我们就来看看netty的简单使用过程,以及背后的原理。

二、netty的简单使用

2.1、netty的环境部署和使用

在这里我们使用myeclipse平台,maven管理工具进行开发,其实使用eclipse或者其他软件也可以。首先我们新建一个maven项目,项目名和包名自定:

沉淀再出发:关于netty的一些理解和使用

    之后我们修改pom.xml文件,增加netty依赖:

沉淀再出发:关于netty的一些理解和使用

沉淀再出发:关于netty的一些理解和使用

保存之后,系统就会自动为我们下载和安装了,非常的方便,这样,我们的环境就部署完毕了。

2.2、一个简单的案例

    下面我们看一个简单地案例:

    我们新建一个包,然后写入两个文件:

沉淀再出发:关于netty的一些理解和使用

    首先我们编写一个处理连接的类 HelloServerHandler :

 package com.coder.server;

 import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;
import io.netty.util.ReferenceCountUtil; public class HelloServerHandler extends ChannelInboundHandlerAdapter {
/**
* 收到数据时调用
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
try {
ByteBuf in = (ByteBuf)msg;
System.out.print(in.toString(CharsetUtil.UTF_8));
} finally {
// 抛弃收到的数据
ReferenceCountUtil.release(msg);
}
} /**
* 当Netty由于IO错误或者处理器在处理事件时抛出异常时调用
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
// 当出现异常就关闭连接
cause.printStackTrace();
ctx.close();
}
}

   其次,我们编写接收连接,并且派发和处理的类 HelloServer :

 package com.coder.server;

 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; public class HelloServer {
private int port; public HelloServer(int port) {
this.port = port;
} public void run() throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(); // 用来接收进来的连接
EventLoopGroup workerGroup = new NioEventLoopGroup(); // 用来处理已经被接收的连接
System.out.println("准备运行端口:" + port); try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class) // 这里告诉Channel如何接收新的连接
.childHandler( new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// 自定义处理类
ch.pipeline().addLast(new HelloServerHandler());
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true); // 绑定端口,开始接收进来的连接
ChannelFuture f = b.bind(port).sync(); // 等待服务器socket关闭
f.channel().closeFuture().sync();
} catch (Exception e) {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
} public static void main(String[] args) throws Exception {
int port = 12345;
new HelloServer(port).run();
}
}

     然后运行,等待连接就好了,那么问题来了,使用什么进行连接呢?在windows中,我们可以使用Telnet,这个比较方便和简单,但是我们需要打开控制面板的程序和功能模块,并且启动服务,之后最好重启一下电脑:

沉淀再出发:关于netty的一些理解和使用

   下面我们运行程序,并使用Telnet客户端测试一下:

沉淀再出发:关于netty的一些理解和使用

沉淀再出发:关于netty的一些理解和使用

   在telnet中‘ctrl+]’可以显示输入的文字,否则将看不到输入。

三、使用netty自定义时间服务器

本例中我们试图在服务器和客户端连接被创立时发送一个消息,然后在客户端解析收到的消息并输出。并且,在这个项目中使用 POJO 代替 ByteBuf 来作为传输对象。

沉淀再出发:关于netty的一些理解和使用

3.1、pojo对象创建

   Time 类:

 package com.coder.pojo;

 import java.util.Date;

 /**
* 自定义时间数据类
*
*/
public class Time {
private final long value; public Time() {
// 除以1000是为了使时间精确到秒
        //注意这里的this,其实就是调用了 public Time(long value) ,并且更加的方便和快捷。
this(System.currentTimeMillis() / 1000L);
} public Time(long value) {
this.value = value;
} public long value() {
return value;
} @Override
public String toString() {
return new Date((value()) * 1000L).toString();
}
}

3.2、服务器程序

   TimeEncoderPOJO类:

 package com.coder.server;

 import com.coder.pojo.Time;

 import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder; /**
* 服务器数据编码类
*
*/
public class TimeEncoderPOJO extends MessageToByteEncoder<Time> { // 发送数据时调用
@Override
protected void encode(ChannelHandlerContext ctx, Time msg, ByteBuf out) throws Exception {
// 只传输当前时间,精确到秒
out.writeInt((int)msg.value());
} }

   TimeServerHandlerPOJO类:连接建立并且准备通信的时候进行处理,发送当前时间,并增加监听。

 package com.coder.server;

 import com.coder.pojo.Time;

 import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter; /**
* 服务器解码器
* 连接建立时发送当前时间
*
*/
public class TimeServerHandlerPOJO extends ChannelInboundHandlerAdapter {
/**
* 连接建立的时候并且准备进行通信时被调用
*/
@Override
public void channelActive(final ChannelHandlerContext ctx) throws Exception {
// 发送当前时间信息
ChannelFuture f = ctx.writeAndFlush(new Time());
// 发送完毕之后关闭 Channel
f.addListener(ChannelFutureListener.CLOSE);
} @Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}

   TimeServerPOJO类:服务器的主程序

 package com.coder.server;

 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; public class TimeServerPOJO {
private int port; public TimeServerPOJO(int port) {
this.port = port;
} public void run() throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(); // 用来接收进来的连接
EventLoopGroup workerGroup = new NioEventLoopGroup(); // 用来处理已经被接收的连接
System.out.println("准备运行端口:" + port); try {
ServerBootstrap b = new ServerBootstrap(); // 启动NIO服务的辅助启动类
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class) // 这里告诉Channel如何接收新的连接
.childHandler( new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// 自定义处理类
// 注意添加顺序
ch.pipeline().addLast(new TimeEncoderPOJO(),new TimeServerHandlerPOJO());
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true); // 绑定端口,开始接收进来的连接
ChannelFuture f = b.bind(port).sync(); // 等待服务器socket关闭
f.channel().closeFuture().sync();
} catch (Exception e) {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
} public static void main(String[] args) throws Exception {
int port = 12345;
new TimeServerPOJO(port).run();
}
}

   其中ch.pipeline().addLast(new TimeEncoderPOJO(),new TimeServerHandlerPOJO());方法的含义为:Handles an I/O event or intercepts an I/O operation, and forwards it to its next handler in its ChannelPipeline.也就是说当我们添加一些处理的时候会按照管道的方式,一步步的处理,因此先后顺序非常重要。

3.3、客户端程序

   先来看看解码器(服务器端发送了编码后的时间信息,因此,这里客户端收到之后需要解码):

   TimeDecoderPOJO 类:

 package com.coder.client;

 import java.util.List;

 import com.coder.pojo.Time;

 import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder; public class TimeDecoderPOJO extends ByteToMessageDecoder {
/**
* 有新数据接收时调用
* 为防止分包现象,先将数据存入内部缓存,到达满足条件之后再进行解码
*/
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
if(in.readableBytes() < 4) {
return;
} // out添加对象则表示解码成功
out.add(new Time(in.readUnsignedInt()));
}
}

  再看看客户端数据处理类:

  TimeClientHandlerPOJO类:

 package com.coder.client;

 import com.coder.pojo.Time;

 import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter; /**
* 客户端数据处理类
*
*/
public class TimeClientHandlerPOJO extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// 直接将信息转换成Time类型输出即可
Time time = (Time)msg;
System.out.println(time);
ctx.close();
} @Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}

  最后是客户端的主程序:

   TimeClientPOJO类:

 package com.coder.client;

 import io.netty.bootstrap.Bootstrap;
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.NioSocketChannel; public class TimeClientPOJO {
public static void main(String[] args) throws Exception{
String host = "127.0.0.1"; // ip
int port = 12345; // 端口
EventLoopGroup workerGroup = new NioEventLoopGroup(); try {
Bootstrap b = new Bootstrap(); // 与ServerBootstrap类似
b.group(workerGroup); // 客户端不需要boss worker
b.channel(NioSocketChannel.class);
b.option(ChannelOption.SO_KEEPALIVE, true); // 客户端的socketChannel没有父亲
b.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// POJO
ch.pipeline().addLast(new TimeDecoderPOJO() ,new TimeClientHandlerPOJO());
}
}); // 启动客户端,客户端用connect连接
ChannelFuture f = b.connect(host, port).sync(); // 等待连接关闭
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
}
}
}

至此程序编写完毕,先运行服务器,再运行客户端程序,然后测试即可,我们会发现服务器一直等待着请求,当客户端连接上之后,服务器就会发出带着格式的时间,客户端接收到之后进行解码,然后显示出来并且退出。在同一个myeclipse之中可以运行多个程序,使用下图中的按钮可以进行切换。

沉淀再出发:关于netty的一些理解和使用

沉淀再出发:关于netty的一些理解和使用

四、netty的基本组成部分

4.1、Channel  

Channel 是 Java NIO 的一个基本构造。它代表一个到实体(如一个硬件设备、一个文件、一个网络套接字或者一个能够执行一个或者多个不同的I/O操作的程序组件)的开放连接,如读操作和写操作。目前,可以把 Channel 看作是传入(入站)或者传出(出站)数据的载体。因此,它可以被打开或者被关闭,连接或者断开连接。

4.2、Callback(回调)  

Netty 在内部使用了回调来处理事件;当一个回调被触发时,相关的事件可以被一个 interfaceChannelHandler 的实现处理。

4.3、Future

  Future 提供了另一种在操作完成时通知应用程序的方式。这个对象可以看作是一个异步操作的结果的占位符;它将在未来的某个时刻完成,并提供对其结果的访问。JDK 预置了 interface java.util.concurrent.Future,但是其所提供的实现,只允许手动检查对应的操作是否已经完成,或者一直阻塞直到它完成。这是非常繁琐的,所以 Netty 提供了它自己的实现ChannelFuture,用于在执行异步操作的时候使用。

4.4、Event 和 Handler

Netty 使用不同的事件来通知我们状态的改变或者是操作的状态。这使得我们能够基于已经发生的事件来触发适当的动作。这些动作可能是:记录日志、数据转换、流控制、应用程序逻辑。Netty 是一个网络编程框架,所以事件是按照它们与入站或出站数据流的相关性进行分类的。可能由入站数据或者相关的状态更改而触发的事件包括:连接已被激活或者连接失活、数据读取、用户事件、错误事件。出站事件是未来将会触发的某个动作的操作结果,这些动作包括:打开或者关闭到远程节点的连接、将数据写到或者冲刷到套接字。
    Netty 的 ChannelHandler 为处理器提供了基本的抽象,目前可以认为每个 ChannelHandler 的实例都类似于一种为了响应特定事件而被执行的回调。Netty 提供了大量预定义的可以开箱即用的 ChannelHandler 实现,包括用于各种协议(如 HTTP 和 SSL/TLS)的 ChannelHandler。在内部 ChannelHandler 自己也使用了事件和 Future。

五、netty聊天程序

5.1、服务器端

SimpleChatServerInitializer类:

 package com.coder.server;

 import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder; /**
* 服务器配置初始化
* 添加多个处理器
*/
public class SimpleChatServerInitializer extends ChannelInitializer<SocketChannel> { @Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
// 添加处理类
// 使用'\r''\n'分割帧
pipeline.addLast("framer",
new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
// 解码、编码器
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
// 处理器
pipeline.addLast("handler", new SimpleChatServerHandler()); System.out.println("SimpleChatClient: " + ch.remoteAddress() + "连接上");
} }

SimpleChatServerHandler类:

 package com.coder.server;

 import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor; /**
* 服务端处理器
*/
public class SimpleChatServerHandler extends SimpleChannelInboundHandler<String> { public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); /**
* 收到新的客户端连接时调用
* 将客户端channel存入列表,并广播消息
*/
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
Channel incoming = ctx.channel();
// 广播加入消息
channels.writeAndFlush("[SERVER] - " + incoming.remoteAddress() + " 加入\n");
channels.add(incoming); // 存入列表
} /**
* 客户端连接断开时调用
* 广播消息
*/
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
Channel incoming = ctx.channel();
// 广播离开消息
channels.writeAndFlush("[SERVER] - " + incoming.remoteAddress() + " 离开\n");
// channel会自动从ChannelGroup中删除
} /**
* 收到消息时调用
* 将消息转发给其他客户端
*/
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
Channel incoming = ctx.channel();
for(Channel channel : channels) { // 遍历所有连接的客户端
if(channel != incoming) { // 其他客户端
channel.writeAndFlush("[" + incoming.remoteAddress() + "] " + msg + "\n" );
} else { // 自己
channel.writeAndFlush("[you] " + msg + "\n" );
}
}
} /**
* 监听到客户端活动时调用
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
Channel incoming = ctx.channel();
System.out.println("SimpleChatClient: " + incoming.remoteAddress() + " 在线");
} /**
* 监听到客户端不活动时调用
*/
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
Channel incoming = ctx.channel();
System.out.println("SimpleChatClient: " + incoming.remoteAddress() + " 掉线");
} /**
* 当Netty由于IO错误或者处理器在处理事件抛出异常时调用
* 关闭连接
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
Channel incoming = ctx.channel();
System.out.println("SimpleChatClient: " + incoming.remoteAddress() + " 异常");
}
}

SimpleChatServer类:

 package com.coder.server;

 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; /**
* 服务端 main 启动
*/
public class SimpleChatServer {
private int port; // 端口 public SimpleChatServer(int port) {
this.port = port;
} // 配置并开启服务器
public void run() throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(); // 用来接收进来的连接
EventLoopGroup workerGroup = new NioEventLoopGroup(); // 用来处理已接收的连接 try {
ServerBootstrap sb = new ServerBootstrap(); // 启动NIO服务的辅助启动类
sb.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class) // 设置如何接受连接
.childHandler(new SimpleChatServerInitializer()) // 配置Channel
.option(ChannelOption.SO_BACKLOG, 128) // 设置缓冲区
.childOption(ChannelOption.SO_KEEPALIVE, true); // 启用心跳机制 System.out.println("SimpleChatServer 启动了");
ChannelFuture future = sb.bind(port).sync(); // 绑定端口,开始接收连接
future.channel().closeFuture().sync(); // 等待关闭服务器(不会发生)
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
System.out.println("SimpleChatServer 关闭了");
}
} public static void main(String[] args) throws Exception {
int port = 8080;
new SimpleChatServer(port).run(); // 开启服务器
}
}

5.2、客户端程序

SimpleChatClientInitializer类:

 package com.coder.client;

 import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder; /**
* 客户端配置初始化
* 与服务端类似
*/
public class SimpleChatClientInitializer extends ChannelInitializer<SocketChannel> { @Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
// 添加处理类
// 使用'\r''\n'分割帧
pipeline.addLast("framer",
new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
// 解码、编码器
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
// 处理器
pipeline.addLast("handler", new SimpleChatClientHandler());
} }

SimpleChatClientHandler类:

 package com.coder.client;

 import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler; /**
* 客户端处理类
* 直接输出收到的消息
*/
public class SimpleChatClientHandler extends SimpleChannelInboundHandler<String> { @Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
System.out.println(msg); // 直接输出消息
} }

SimpleChatClient类:

 package com.coder.client;

 import java.io.BufferedReader;
import java.io.InputStreamReader; 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;
/**
* 客户端
* 开启客户端,接收控制台输入并发送给服务端
*/
public class SimpleChatClient {
private final String host; // IP
private final int port; // 端口 public SimpleChatClient(String host, int port) {
this.host = host;
this.port = port;
} // 配置并运行客户端
public void run() throws Exception {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap(); // 客户端辅助启动类
b.group(group) // 客户端只需要一个用来接收并处理连接
.channel(NioSocketChannel.class) // 设置如何接受连接
.handler(new SimpleChatClientInitializer());// 配置 channel
// 连接服务器
Channel channel = b.connect(host, port).sync().channel();
// 读取控制台输入字符
BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
while(true) {
// 每行成一帧输出,以"\r\n"结尾
channel.writeAndFlush(in.readLine() + "\r\n");
}
} catch (Exception e) {
e.printStackTrace(); // 输出异常
} finally {
group.shutdownGracefully(); // 关闭
}
} public static void main(String[] args) throws Exception {
new SimpleChatClient("localhost", 8080).run(); // 启动客户端
} }

   运行结果:

沉淀再出发:关于netty的一些理解和使用沉淀再出发:关于netty的一些理解和使用

沉淀再出发:关于netty的一些理解和使用沉淀再出发:关于netty的一些理解和使用

沉淀再出发:关于netty的一些理解和使用沉淀再出发:关于netty的一些理解和使用

沉淀再出发:关于netty的一些理解和使用沉淀再出发:关于netty的一些理解和使用

沉淀再出发:关于netty的一些理解和使用

六、总结

通过代码的形式,我们对netty有了直观的了解和实际上的掌握。

程序源码

参考文献:https://www.cnblogs.com/coderJiebao/tag/netty/

上一篇:C语言结构体里的成员数组和指针


下一篇:linux group