Netty是由JBOSS提供的一个java开源框架。Netty提供异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。也就是说，Netty 是一个基于NIO的客户、服务器端编程框架，使用Netty 可以确保你快速和简单的开发出一个网络应用，例如实现了某种协议的客户，服务端应用。Netty相当简化和流线化了网络应用的编程开发过程，例如，TCP和UDP的socket服务开发。
“快速”和“简单”并不用产生维护性或性能上的问题。Netty 是一个吸收了多种协议的实现经验，这些协议包括FTP,SMTP,HTTP，各种二进制，文本协议，并经过相当精心设计的项目，最终，Netty 成功的找到了一种方式，在保证易于开发的同时还保证了其应用的性能，稳定性和伸缩性。
Netty从4.x版本开始，需要使用JDK1.6及以上版本提供基础支撑。
在设计上：针对多种传输类型的统一接口 - 阻塞和非阻塞；简单但更强大的线程模型；真正的无连接的数据报套接字支持；链接逻辑支持复用。
在性能上：比核心 Java API 更好的吞吐量，较低的延时；资源消耗更少，这个得益于共享池和重用；减少内存拷贝。
在健壮性上：消除由于慢，快，或重载连接产生的 OutOfMemoryError；消除经常发现在 NIO 在高速网络中的应用中的不公平的读/写比。
在安全上：完整的 SSL / TLS 和 StartTLS 的支持，且已得到大量商业应用的真实验证,如：Hadoop项目的Avro（RPC框架）、Dubbo、Dubbox等RPC框架。
Netty的官网是：
http://netty.io有三方提供的中文翻译Netty用户手册（官网提供源信息）：
http://ifeve.com/netty5-user-guide/
在ServerBootstrap调用方法group的时候，传递的参数是同一个线程组，且在构造线程组的时候，构造参数为1，这种开发方式，就是一个单线程模型。个人机开发测试使用。不推荐。
在ServerBootstrap调用方法group的时候，传递的参数是两个不同的线程组。负责监听的acceptor线程组，线程数为1，也就是构造参数为1。负责处理客户端任务的线程组，线程数大于1，也就是构造参数大于1。这种开发方式，就是多线程模型。长连接，且客户端数量较少，连接持续时间较长情况下使用。如：企业内部交流应用。
在ServerBootstrap调用方法group的时候，传递的参数是两个不同的线程组。负责监听的acceptor线程组，线程数大于1，也就是构造参数大于1。负责处理客户端任务的线程组，线程数大于1，也就是构造参数大于1。这种开发方式，就是主从多线程模型。长连接，客户端数量相对较多，连接持续时间比较长的情况下使用。如：对外提供服务的相册服务器。
/**
* 1. 双线程组
* 2. Bootstrap配置启动信息
* 3. 注册业务处理Handler
* 4. 绑定服务监听端口并启动服务
*/
package com.hhxy.netty.first;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
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 Server4HelloWorld {
// 监听线程组，监听客户端请求
private EventLoopGroup acceptorGroup = null;
// 处理客户端相关操作线程组，负责处理与客户端的数据通讯
private EventLoopGroup clientGroup = null;
// 服务启动相关配置信息
private ServerBootstrap bootstrap = null;
public Server4HelloWorld(){
init();
}
private void init(){
// 初始化线程组,构建线程组的时候，如果不传递参数，则默认构建的线程组线程数是CPU核心数量。
acceptorGroup = new NioEventLoopGroup();
clientGroup = new NioEventLoopGroup();
// 初始化服务的配置
bootstrap = new ServerBootstrap();
// 绑定线程组
bootstrap.group(acceptorGroup, clientGroup);
// 设定通讯模式为NIO， 同步非阻塞
bootstrap.channel(NioServerSocketChannel.class);
// 设定缓冲区大小， 缓存区的单位是字节。
bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
// SO_SNDBUF发送缓冲区，SO_RCVBUF接收缓冲区，SO_KEEPALIVE开启心跳监测（保证连接有效）
bootstrap.option(ChannelOption.SO_SNDBUF, 16*1024)
.option(ChannelOption.SO_RCVBUF, 16*1024)
.option(ChannelOption.SO_KEEPALIVE, true);
}
/**
* 监听处理逻辑。
* @param port 监听端口。
* @param acceptorHandlers 处理器， 如何处理客户端请求。
* @return
* @throws InterruptedException
*/
public ChannelFuture doAccept(int port, final ChannelHandler... acceptorHandlers) throws InterruptedException{
/*
* childHandler是服务的Bootstrap独有的方法。是用于提供处理对象的。
* 可以一次性增加若干个处理逻辑。是类似责任链模式的处理方式。
* 增加A，B两个处理逻辑，在处理客户端请求数据的时候，根据A-》B顺序依次处理。
*
* ChannelInitializer - 用于提供处理器的一个模型对象。
* 其中定义了一个方法，initChannel方法。
* 方法是用于初始化处理逻辑责任链条的。
* 可以保证服务端的Bootstrap只初始化一次处理器，尽量提供处理逻辑的重用。
* 避免反复的创建处理器对象。节约资源开销。
*/
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(acceptorHandlers);
}
});
// bind方法 - 绑定监听端口的。ServerBootstrap可以绑定多个监听端口。 多次调用bind方法即可
// sync - 开始监听逻辑。 返回一个ChannelFuture。 返回结果代表的是监听成功后的一个对应的未来结果
// 可以使用ChannelFuture实现后续的服务器和客户端的交互。
ChannelFuture future = bootstrap.bind(port).sync();
return future;
}
/**
* shutdownGracefully - 方法是一个安全关闭的方法。可以保证不放弃任何一个已接收的客户端请求。
*/
public void release(){
this.acceptorGroup.shutdownGracefully();
this.clientGroup.shutdownGracefully();
}
public static void main(String[] args){
ChannelFuture future = null;
Server4HelloWorld server = null;
try{
server = new Server4HelloWorld();
future = server.doAccept(9999,new Server4HelloWorldHandler());
System.out.println("server started.");
// 关闭连接的。
future.channel().closeFuture().sync();
}catch(InterruptedException e){
e.printStackTrace();
}finally{
if(null != future){
try {
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
if(null != server){
server.release();
}
}
}
}
/**
* @Sharable注解 -
* 代表当前Handler是一个可以分享的处理器。也就意味着，服务器注册此Handler后，可以分享给多个客户端同时使用。
* 如果不使用注解描述类型，则每次客户端请求时，必须为客户端重新创建一个新的Handler对象。
* 如果handler是一个Sharable的，一定避免定义可写的实例变量。
* bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new XxxHandler());
}
});
*/
package com.hhxy.netty.first;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
@Sharable
public class Server4HelloWorldHandler extends ChannelHandlerAdapter {
/**
* 业务处理逻辑
* 用于处理读取数据请求的逻辑。
* ctx - 上下文对象。其中包含于客户端建立连接的所有资源。 如： 对应的Channel
* msg - 读取到的数据。 默认类型是ByteBuf，是Netty自定义的。是对ByteBuffer的封装。 不需要考虑复位问题。
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// 获取读取的数据， 是一个缓冲。
ByteBuf readBuffer = (ByteBuf) msg;
// 创建一个字节数组，用于保存缓存中的数据。
byte[] tempDatas = new byte[readBuffer.readableBytes()];
// 将缓存中的数据读取到字节数组中。
readBuffer.readBytes(tempDatas);
String message = new String(tempDatas, "UTF-8");
System.out.println("from client : " + message);
if("exit".equals(message)){
ctx.close();
return;
}
String line = "server message to client!";
// 写操作自动释放缓存，避免内存溢出问题。
ctx.writeAndFlush(Unpooled.copiedBuffer(line.getBytes("UTF-8")));
// 注意，如果调用的是write方法。不会刷新缓存，缓存中的数据不会发送到客户端，必须再次调用flush方法才行。
// ctx.write(Unpooled.copiedBuffer(line.getBytes("UTF-8")));
// ctx.flush();
}
/**
* 异常处理逻辑， 当客户端异常退出的时候，也会运行。
* ChannelHandlerContext关闭，也代表当前与客户端连接的资源关闭。
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("server exceptionCaught method run...");
// cause.printStackTrace();
ctx.close();
}
}
/**
* 1. 单线程组
* 2. Bootstrap配置启动信息
* 3. 注册业务处理Handler
* 4. connect连接服务，并发起请求
*/
package com.hhxy.netty.first;
import java.util.Scanner;
import java.util.concurrent.TimeUnit;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
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 Client4HelloWorld {
// 处理请求和处理服务端响应的线程组
private EventLoopGroup group = null;
// 客户端启动相关配置信息
private Bootstrap bootstrap = null;
public Client4HelloWorld(){
init();
}
private void init(){
group = new NioEventLoopGroup();
bootstrap = new Bootstrap();
// 绑定线程组
bootstrap.group(group);
// 设定通讯模式为NIO
bootstrap.channel(NioSocketChannel.class);
}
public ChannelFuture doRequest(String host, int port, final ChannelHandler... handlers) throws InterruptedException{
/*
* 客户端的Bootstrap没有childHandler方法。只有handler方法。
* 方法含义等同ServerBootstrap中的childHandler
* 在客户端必须绑定处理器，也就是必须调用handler方法。
* 服务器必须绑定处理器，必须调用childHandler方法。
*/
this.bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(handlers);
}
});
// 建立连接。
ChannelFuture future = this.bootstrap.connect(host, port).sync();
return future;
}
public void release(){
this.group.shutdownGracefully();
}
public static void main(String[] args) {
Client4HelloWorld client = null;
ChannelFuture future = null;
try{
client = new Client4HelloWorld();
future = client.doRequest("localhost", 9999, new Client4HelloWorldHandler());
Scanner s = null;
while(true){
s = new Scanner(System.in);
System.out.print("enter message send to server (enter 'exit' for close client) > ");
String line = s.nextLine();
if("exit".equals(line)){
// addListener - 增加监听，当某条件满足的时候，触发监听器。
// ChannelFutureListener.CLOSE - 关闭监听器，代表ChannelFuture执行返回后，关闭连接。
future.channel().writeAndFlush(Unpooled.copiedBuffer(line.getBytes("UTF-8")))
.addListener(ChannelFutureListener.CLOSE);
break;
}
future.channel().writeAndFlush(Unpooled.copiedBuffer(line.getBytes("UTF-8")));
TimeUnit.SECONDS.sleep(1);
}
}catch(Exception e){
e.printStackTrace();
}finally{
if(null != future){
try {
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
if(null != client){
client.release();
}
}
}
}
package com.hhxy.netty.first;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.ReferenceCountUtil;
public class Client4HelloWorldHandler extends ChannelHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
try{
ByteBuf readBuffer = (ByteBuf) msg;
byte[] tempDatas = new byte[readBuffer.readableBytes()];
readBuffer.readBytes(tempDatas);
System.out.println("from server : " + new String(tempDatas, "UTF-8"));
}finally{
// 用于释放缓存。避免内存溢出
ReferenceCountUtil.release(msg);
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("client exceptionCaught method run...");
// cause.printStackTrace();
ctx.close();
}
/*@Override // 断开连接时执行
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
System.out.println("channelInactive method run...");
}
@Override // 连接通道建立成功时执行
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("channelActive method run...");
}
@Override // 每次读取完成时执行
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
System.out.println("channelReadComplete method run...");
}*/
}
netty使用tcp/ip协议传输数据。而tcp/ip协议是类似水流一样的数据传输方式。多次访问的时候有可能出现数据粘包的问题，解决这种问题的方式如下：
客户端和服务器，提前协调好，每个消息长度固定。（如：长度10）。如果客户端或服务器写出的数据不足10，则使用空白字符补足（如：使用空格）。
/**
* 1. 双线程组
* 2. Bootstrap配置启动信息
* 3. 注册业务处理Handler
* 4. 绑定服务监听端口并启动服务
*/
package com.hhxy.netty.fixedlength;
import java.nio.charset.Charset;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
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;
import io.netty.handler.codec.FixedLengthFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
public class Server4FixedLength {
// 监听线程组，监听客户端请求
private EventLoopGroup acceptorGroup = null;
// 处理客户端相关操作线程组，负责处理与客户端的数据通讯
private EventLoopGroup clientGroup = null;
// 服务启动相关配置信息
private ServerBootstrap bootstrap = null;
public Server4FixedLength(){
init();
}
private void init(){
acceptorGroup = new NioEventLoopGroup();
clientGroup = new NioEventLoopGroup();
bootstrap = new ServerBootstrap();
// 绑定线程组
bootstrap.group(acceptorGroup, clientGroup);
// 设定通讯模式为NIO
bootstrap.channel(NioServerSocketChannel.class);
// 设定缓冲区大小
bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
// SO_SNDBUF发送缓冲区，SO_RCVBUF接收缓冲区，SO_KEEPALIVE开启心跳监测（保证连接有效）
bootstrap.option(ChannelOption.SO_SNDBUF, 16*1024)
.option(ChannelOption.SO_RCVBUF, 16*1024)
.option(ChannelOption.SO_KEEPALIVE, true);
}
public ChannelFuture doAccept(int port) throws InterruptedException{
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelHandler[] acceptorHandlers = new ChannelHandler[3];
// 定长Handler。通过构造参数设置消息长度（单位是字节）。发送的消息长度不足可以使用空格补全。
acceptorHandlers[0] = new FixedLengthFrameDecoder(5);
// 字符串解码器Handler，会自动处理channelRead方法的msg参数，将ByteBuf类型的数据转换为字符串对象
acceptorHandlers[1] = new StringDecoder(Charset.forName("UTF-8"));
acceptorHandlers[2] = new Server4FixedLengthHandler();
ch.pipeline().addLast(acceptorHandlers);
}
});
ChannelFuture future = bootstrap.bind(port).sync();
return future;
}
public void release(){
this.acceptorGroup.shutdownGracefully();
this.clientGroup.shutdownGracefully();
}
public static void main(String[] args){
ChannelFuture future = null;
Server4FixedLength server = null;
try{
server = new Server4FixedLength();
future = server.doAccept(9999);
System.out.println("server started.");
future.channel().closeFuture().sync();
}catch(InterruptedException e){
e.printStackTrace();
}finally{
if(null != future){
try {
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
if(null != server){
server.release();
}
}
}
}
package com.hhxy.netty.fixedlength;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
public class Server4FixedLengthHandler extends ChannelHandlerAdapter {
// 业务处理逻辑
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
String message = msg.toString();
System.out.println("from client : " + message.trim());
String line = "ok ";
ctx.writeAndFlush(Unpooled.copiedBuffer(line.getBytes("UTF-8")));
}
// 异常处理逻辑
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("server exceptionCaught method run...");
// cause.printStackTrace();
ctx.close();
}
}
/**
* 1. 单线程组
* 2. Bootstrap配置启动信息
* 3. 注册业务处理Handler
* 4. connect连接服务，并发起请求
*/
package com.hhxy.netty.fixedlength;
import java.nio.charset.Charset;
import java.util.Scanner;
import java.util.concurrent.TimeUnit;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.FixedLengthFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
public class Client4FixedLength {
// 处理请求和处理服务端响应的线程组
private EventLoopGroup group = null;
// 服务启动相关配置信息
private Bootstrap bootstrap = null;
public Client4FixedLength(){
init();
}
private void init(){
group = new NioEventLoopGroup();
bootstrap = new Bootstrap();
// 绑定线程组
bootstrap.group(group);
// 设定通讯模式为NIO
bootstrap.channel(NioSocketChannel.class);
}
public ChannelFuture doRequest(String host, int port) throws InterruptedException{
this.bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelHandler[] handlers = new ChannelHandler[3];
handlers[0] = new FixedLengthFrameDecoder(3);
// 字符串解码器Handler，会自动处理channelRead方法的msg参数，将ByteBuf类型的数据转换为字符串对象
handlers[1] = new StringDecoder(Charset.forName("UTF-8"));
handlers[2] = new Client4FixedLengthHandler();
ch.pipeline().addLast(handlers);
}
});
ChannelFuture future = this.bootstrap.connect(host, port).sync();
return future;
}
public void release(){
this.group.shutdownGracefully();
}
public static void main(String[] args) {
Client4FixedLength client = null;
ChannelFuture future = null;
try{
client = new Client4FixedLength();
future = client.doRequest("localhost", 9999);
Scanner s = null;
while(true){
s = new Scanner(System.in);
System.out.print("enter message send to server > ");
String line = s.nextLine();
byte[] bs = new byte[5];
byte[] temp = line.getBytes("UTF-8");
if(temp.length <= 5){
for(int i = 0; i < temp.length; i++){
bs[i] = temp[i];
}
}
future.channel().writeAndFlush(Unpooled.copiedBuffer(bs));
TimeUnit.SECONDS.sleep(1);
}
}catch(Exception e){
e.printStackTrace();
}finally{
if(null != future){
try {
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
if(null != client){
client.release();
}
}
}
}
package com.hhxy.netty.fixedlength;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.ReferenceCountUtil;
public class Client4FixedLengthHandler extends ChannelHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
try{
String message = msg.toString();
System.out.println("from server : " + message);
}finally{
// 用于释放缓存。避免内存溢出
ReferenceCountUtil.release(msg);
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("client exceptionCaught method run...");
// cause.printStackTrace();
ctx.close();
}
}
客户端和服务器，协商定义一个特殊的分隔符号，分隔符号长度自定义。如：‘#’、‘$_$’、‘AA@’。在通讯的时候，只要没有发送分隔符号，则代表一条数据没有结束。
/**
* 1. 双线程组
* 2. Bootstrap配置启动信息
* 3. 注册业务处理Handler
* 4. 绑定服务监听端口并启动服务
*/
package com.hhxy.netty.delimiter;
import java.nio.charset.Charset;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
public class Server4Delimiter {
// 监听线程组，监听客户端请求
private EventLoopGroup acceptorGroup = null;
// 处理客户端相关操作线程组，负责处理与客户端的数据通讯
private EventLoopGroup clientGroup = null;
// 服务启动相关配置信息
private ServerBootstrap bootstrap = null;
public Server4Delimiter(){
init();
}
private void init(){
acceptorGroup = new NioEventLoopGroup();
clientGroup = new NioEventLoopGroup();
bootstrap = new ServerBootstrap();
// 绑定线程组
bootstrap.group(acceptorGroup, clientGroup);
// 设定通讯模式为NIO
bootstrap.channel(NioServerSocketChannel.class);
// 设定缓冲区大小
bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
// SO_SNDBUF发送缓冲区，SO_RCVBUF接收缓冲区，SO_KEEPALIVE开启心跳监测（保证连接有效）
bootstrap.option(ChannelOption.SO_SNDBUF, 16*1024)
.option(ChannelOption.SO_RCVBUF, 16*1024)
.option(ChannelOption.SO_KEEPALIVE, true);
}
public ChannelFuture doAccept(int port) throws InterruptedException{
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// 数据分隔符, 定义的数据分隔符一定是一个ByteBuf类型的数据对象。
ByteBuf delimiter = Unpooled.copiedBuffer("$E$".getBytes());
ChannelHandler[] acceptorHandlers = new ChannelHandler[3];
// 处理固定结束标记符号的Handler。这个Handler没有@Sharable注解修饰，
// 必须每次初始化通道时创建一个新对象
// 使用特殊符号分隔处理数据粘包问题，也要定义每个数据包最大长度。netty建议数据有最大长度。
acceptorHandlers[0] = new DelimiterBasedFrameDecoder(1024, delimiter);
// 字符串解码器Handler，会自动处理channelRead方法的msg参数，将ByteBuf类型的数据转换为字符串对象
acceptorHandlers[1] = new StringDecoder(Charset.forName("UTF-8"));
acceptorHandlers[2] = new Server4DelimiterHandler();
ch.pipeline().addLast(acceptorHandlers);
}
});
ChannelFuture future = bootstrap.bind(port).sync();
return future;
}
public void release(){
this.acceptorGroup.shutdownGracefully();
this.clientGroup.shutdownGracefully();
}
public static void main(String[] args){
ChannelFuture future = null;
Server4Delimiter server = null;
try{
server = new Server4Delimiter();
future = server.doAccept(9999);
System.out.println("server started.");
future.channel().closeFuture().sync();
}catch(InterruptedException e){
e.printStackTrace();
}finally{
if(null != future){
try {
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
if(null != server){
server.release();
}
}
}
}
package com.hhxy.netty.delimiter;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
public class Server4DelimiterHandler extends ChannelHandlerAdapter {
// 业务处理逻辑
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
String message = msg.toString();
System.out.println("from client : " + message);
String line = "server message $E$ test delimiter handler!! $E$ second message $E$";
ctx.writeAndFlush(Unpooled.copiedBuffer(line.getBytes("UTF-8")));
}
// 异常处理逻辑
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("server exceptionCaught method run...");
// cause.printStackTrace();
ctx.close();
}
}
/**
* 1. 单线程组
* 2. Bootstrap配置启动信息
* 3. 注册业务处理Handler
* 4. connect连接服务，并发起请求
*/
package com.hhxy.netty.delimiter;
import java.nio.charset.Charset;
import java.util.Scanner;
import java.util.concurrent.TimeUnit;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
public class Client4Delimiter {
// 处理请求和处理服务端响应的线程组
private EventLoopGroup group = null;
// 服务启动相关配置信息
private Bootstrap bootstrap = null;
public Client4Delimiter(){
init();
}
private void init(){
group = new NioEventLoopGroup();
bootstrap = new Bootstrap();
// 绑定线程组
bootstrap.group(group);
// 设定通讯模式为NIO
bootstrap.channel(NioSocketChannel.class);
}
public ChannelFuture doRequest(String host, int port) throws InterruptedException{
this.bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// 数据分隔符
ByteBuf delimiter = Unpooled.copiedBuffer("$E$".getBytes());
ChannelHandler[] handlers = new ChannelHandler[3];
handlers[0] = new DelimiterBasedFrameDecoder(1024, delimiter);
// 字符串解码器Handler，会自动处理channelRead方法的msg参数，将ByteBuf类型的数据转换为字符串对象
handlers[1] = new StringDecoder(Charset.forName("UTF-8"));
handlers[2] = new Client4DelimiterHandler();
ch.pipeline().addLast(handlers);
}
});
ChannelFuture future = this.bootstrap.connect(host, port).sync();
return future;
}
public void release(){
this.group.shutdownGracefully();
}
public static void main(String[] args) {
Client4Delimiter client = null;
ChannelFuture future = null;
try{
client = new Client4Delimiter();
future = client.doRequest("localhost", 9999);
Scanner s = null;
while(true){
s = new Scanner(System.in);
System.out.print("enter message send to server > ");
String line = s.nextLine();
future.channel().writeAndFlush(Unpooled.copiedBuffer(line.getBytes("UTF-8")));
TimeUnit.SECONDS.sleep(1);
}
}catch(Exception e){
e.printStackTrace();
}finally{
if(null != future){
try {
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
if(null != client){
client.release();
}
}
}
}
package com.hhxy.netty.delimiter;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.ReferenceCountUtil;
public class Client4DelimiterHandler extends ChannelHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
try{
String message = msg.toString();
System.out.println("from server : " + message);
}finally{
// 用于释放缓存。避免内存溢出
ReferenceCountUtil.release(msg);
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("client exceptionCaught method run...");
// cause.printStackTrace();
ctx.close();
}
}
相对最成熟的数据传递方式。有服务器的开发者提供一个固定格式的协议标准。客户端和服务器发送数据和接受数据的时候，都依据协议制定和解析消息。
协议格式：
HEADcontent-length:xxxxHEADBODYxxxxxxBODY/**
* 1. 双线程组
* 2. Bootstrap配置启动信息
* 3. 注册业务处理Handler
* 4. 绑定服务监听端口并启动服务
*/
package com.hhxy.netty.protocol;
import java.nio.charset.Charset;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
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;
import io.netty.handler.codec.string.StringDecoder;
public class Server4Protocol {
// 监听线程组，监听客户端请求
private EventLoopGroup acceptorGroup = null;
// 处理客户端相关操作线程组，负责处理与客户端的数据通讯
private EventLoopGroup clientGroup = null;
// 服务启动相关配置信息
private ServerBootstrap bootstrap = null;
public Server4Protocol(){
init();
}
private void init(){
acceptorGroup = new NioEventLoopGroup();
clientGroup = new NioEventLoopGroup();
bootstrap = new ServerBootstrap();
// 绑定线程组
bootstrap.group(acceptorGroup, clientGroup);
// 设定通讯模式为NIO
bootstrap.channel(NioServerSocketChannel.class);
// 设定缓冲区大小
bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
// SO_SNDBUF发送缓冲区，SO_RCVBUF接收缓冲区，SO_KEEPALIVE开启心跳监测（保证连接有效）
bootstrap.option(ChannelOption.SO_SNDBUF, 16*1024)
.option(ChannelOption.SO_RCVBUF, 16*1024)
.option(ChannelOption.SO_KEEPALIVE, true);
}
public ChannelFuture doAccept(int port, final ChannelHandler... acceptorHandlers) throws InterruptedException{
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new StringDecoder(Charset.forName("UTF-8")));
ch.pipeline().addLast(acceptorHandlers);
}
});
ChannelFuture future = bootstrap.bind(port).sync();
return future;
}
public void release(){
this.acceptorGroup.shutdownGracefully();
this.clientGroup.shutdownGracefully();
}
public static void main(String[] args){
ChannelFuture future = null;
Server4Protocol server = null;
try{
server = new Server4Protocol();
future = server.doAccept(9999,new Server4ProtocolHandler());
System.out.println("server started.");
future.channel().closeFuture().sync();
}catch(InterruptedException e){
e.printStackTrace();
}finally{
if(null != future){
try {
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
if(null != server){
server.release();
}
}
}
}
/**
* @Sharable注解 -
* 代表当前Handler是一个可以分享的处理器。也就意味着，服务器注册此Handler后，可以分享给多个客户端同时使用。
* 如果不使用注解描述类型，则每次客户端请求时，必须为客户端重新创建一个新的Handler对象。
*
*/
package com.hhxy.netty.protocol;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
@Sharable
public class Server4ProtocolHandler extends ChannelHandlerAdapter {
// 业务处理逻辑
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
String message = msg.toString();
System.out.println("server receive protocol content : " + message);
message = ProtocolParser.parse(message);
if(null == message){
System.out.println("error request from client");
return ;
}
System.out.println("from client : " + message);
String line = "server message";
line = ProtocolParser.transferTo(line);
System.out.println("server send protocol content : " + line);
ctx.writeAndFlush(Unpooled.copiedBuffer(line.getBytes("UTF-8")));
}
// 异常处理逻辑
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("server exceptionCaught method run...");
cause.printStackTrace();
ctx.close();
}
static class ProtocolParser{
public static String parse(String message){
String[] temp = message.split("HEADBODY");
temp[0] = temp[0].substring(4);
temp[1] = temp[1].substring(0, (temp[1].length()-4));
int length = Integer.parseInt(temp[0].substring(temp[0].indexOf(":")+1));
if(length != temp[1].length()){
return null;
}
return temp[1];
}
public static String transferTo(String message){
message = "HEADcontent-length:" + message.length() + "HEADBODY" + message + "BODY";
return message;
}
}
}
/**
* 1. 单线程组
* 2. Bootstrap配置启动信息
* 3. 注册业务处理Handler
* 4. connect连接服务，并发起请求
*/
package com.hhxy.netty.protocol;
import java.nio.charset.Charset;
import java.util.Scanner;
import java.util.concurrent.TimeUnit;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
public class Client4Protocol {
// 处理请求和处理服务端响应的线程组
private EventLoopGroup group = null;
// 服务启动相关配置信息
private Bootstrap bootstrap = null;
public Client4Protocol(){
init();
}
private void init(){
group = new NioEventLoopGroup();
bootstrap = new Bootstrap();
// 绑定线程组
bootstrap.group(group);
// 设定通讯模式为NIO
bootstrap.channel(NioSocketChannel.class);
}
public ChannelFuture doRequest(String host, int port, final ChannelHandler... handlers) throws InterruptedException{
this.bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new StringDecoder(Charset.forName("UTF-8")));
ch.pipeline().addLast(handlers);
}
});
ChannelFuture future = this.bootstrap.connect(host, port).sync();
return future;
}
public void release(){
this.group.shutdownGracefully();
}
public static void main(String[] args) {
Client4Protocol client = null;
ChannelFuture future = null;
try{
client = new Client4Protocol();
future = client.doRequest("localhost", 9999, new Client4ProtocolHandler());
Scanner s = null;
while(true){
s = new Scanner(System.in);
System.out.print("enter message send to server > ");
String line = s.nextLine();
line = Client4ProtocolHandler.ProtocolParser.transferTo(line);
System.out.println("client send protocol content : " + line);
future.channel().writeAndFlush(Unpooled.copiedBuffer(line.getBytes("UTF-8")));
TimeUnit.SECONDS.sleep(1);
}
}catch(Exception e){
e.printStackTrace();
}finally{
if(null != future){
try {
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
if(null != client){
client.release();
}
}
}
}
package com.hhxy.netty.protocol;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.ReferenceCountUtil;
public class Client4ProtocolHandler extends ChannelHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
try{
String message = msg.toString();
System.out.println("client receive protocol content : " + message);
message = ProtocolParser.parse(message);
if(null == message){
System.out.println("error response from server");
return ;
}
System.out.println("from server : " + message);
}finally{
// 用于释放缓存。避免内存溢出
ReferenceCountUtil.release(msg);
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("client exceptionCaught method run...");
// cause.printStackTrace();
ctx.close();
}
static class ProtocolParser{
public static String parse(String message){
String[] temp = message.split("HEADBODY");
temp[0] = temp[0].substring(4);
temp[1] = temp[1].substring(0, (temp[1].length()-4));
int length = Integer.parseInt(temp[0].substring(temp[0].indexOf(":")+1));
if(length != temp[1].length()){
return null;
}
return temp[1];
}
public static String transferTo(String message){
message = "HEADcontent-length:" + message.length() + "HEADBODY" + message + "BODY";
return message;
}
}
}
JBoss Marshalling序列化：Java是面向对象的开发语言。传递的数据如果是Java对象，应该是最方便且可靠。
/**
* 1. 双线程组
* 2. Bootstrap配置启动信息
* 3. 注册业务处理Handler
* 4. 绑定服务监听端口并启动服务
*/
package com.hhxy.netty.serialized;
import com.hhxy.utils.SerializableFactory4Marshalling;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
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 Server4Serializable {
// 监听线程组，监听客户端请求
private EventLoopGroup acceptorGroup = null;
// 处理客户端相关操作线程组，负责处理与客户端的数据通讯
private EventLoopGroup clientGroup = null;
// 服务启动相关配置信息
private ServerBootstrap bootstrap = null;
public Server4Serializable(){
init();
}
private void init(){
acceptorGroup = new NioEventLoopGroup();
clientGroup = new NioEventLoopGroup();
bootstrap = new ServerBootstrap();
// 绑定线程组
bootstrap.group(acceptorGroup, clientGroup);
// 设定通讯模式为NIO
bootstrap.channel(NioServerSocketChannel.class);
// 设定缓冲区大小
bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
// SO_SNDBUF发送缓冲区，SO_RCVBUF接收缓冲区，SO_KEEPALIVE开启心跳监测（保证连接有效）
bootstrap.option(ChannelOption.SO_SNDBUF, 16*1024)
.option(ChannelOption.SO_RCVBUF, 16*1024)
.option(ChannelOption.SO_KEEPALIVE, true);
}
public ChannelFuture doAccept(int port, final ChannelHandler... acceptorHandlers) throws InterruptedException{
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(SerializableFactory4Marshalling.buildMarshallingDecoder());
ch.pipeline().addLast(SerializableFactory4Marshalling.buildMarshallingEncoder());
ch.pipeline().addLast(acceptorHandlers);
}
});
ChannelFuture future = bootstrap.bind(port).sync();
return future;
}
public void release(){
this.acceptorGroup.shutdownGracefully();
this.clientGroup.shutdownGracefully();
}
public static void main(String[] args){
ChannelFuture future = null;
Server4Serializable server = null;
try{
server = new Server4Serializable();
future = server.doAccept(9999,new Server4SerializableHandler());
System.out.println("server started.");
future.channel().closeFuture().sync();
}catch(InterruptedException e){
e.printStackTrace();
}finally{
if(null != future){
try {
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
if(null != server){
server.release();
}
}
}
}
/**
* @Sharable注解 -
* 代表当前Handler是一个可以分享的处理器。也就意味着，服务器注册此Handler后，可以分享给多个客户端同时使用。
* 如果不使用注解描述类型，则每次客户端请求时，必须为客户端重新创建一个新的Handler对象。
*
*/
package com.hhxy.netty.serialized;
import io.netty.channel.ChannelHandler.Sharable;
import com.hhxy.utils.GzipUtils;
import com.hhxy.utils.RequestMessage;
import com.hhxy.utils.ResponseMessage;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
@Sharable
public class Server4SerializableHandler extends ChannelHandlerAdapter {
// 业务处理逻辑
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("from client : ClassName - " + msg.getClass().getName()
+ " ; message : " + msg.toString());
if(msg instanceof RequestMessage){
RequestMessage request = (RequestMessage)msg;
byte[] attachment = GzipUtils.unzip(request.getAttachment());
System.out.println(new String(attachment));
}
ResponseMessage response = new ResponseMessage(0L, "test response");
ctx.writeAndFlush(response);
}
// 异常处理逻辑
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("server exceptionCaught method run...");
cause.printStackTrace();
ctx.close();
}
}
/**
* 1. 单线程组
* 2. Bootstrap配置启动信息
* 3. 注册业务处理Handler
* 4. connect连接服务，并发起请求
*/
package com.hhxy.netty.serialized;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import com.hhxy.utils.GzipUtils;
import com.hhxy.utils.RequestMessage;
import com.hhxy.utils.SerializableFactory4Marshalling;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
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 Client4Serializable {
// 处理请求和处理服务端响应的线程组
private EventLoopGroup group = null;
// 服务启动相关配置信息
private Bootstrap bootstrap = null;
public Client4Serializable(){
init();
}
private void init(){
group = new NioEventLoopGroup();
bootstrap = new Bootstrap();
// 绑定线程组
bootstrap.group(group);
// 设定通讯模式为NIO
bootstrap.channel(NioSocketChannel.class);
}
public ChannelFuture doRequest(String host, int port, final ChannelHandler... handlers) throws InterruptedException{
this.bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(SerializableFactory4Marshalling.buildMarshallingDecoder());
ch.pipeline().addLast(SerializableFactory4Marshalling.buildMarshallingEncoder());
ch.pipeline().addLast(handlers);
}
});
ChannelFuture future = this.bootstrap.connect(host, port).sync();
return future;
}
public void release(){
this.group.shutdownGracefully();
}
public static void main(String[] args) {
Client4Serializable client = null;
ChannelFuture future = null;
try{
client = new Client4Serializable();
future = client.doRequest("localhost", 9999, new Client4SerializableHandler());
String attachment = "test attachment";
byte[] attBuf = attachment.getBytes();
attBuf = GzipUtils.zip(attBuf);
RequestMessage msg = new RequestMessage(new Random().nextLong(),
"test", attBuf);
future.channel().writeAndFlush(msg);
TimeUnit.SECONDS.sleep(1);
future.addListener(ChannelFutureListener.CLOSE);
}catch(Exception e){
e.printStackTrace();
}finally{
if(null != future){
try {
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
if(null != client){
client.release();
}
}
}
}
package com.hhxy.netty.serialized;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
public class Client4SerializableHandler extends ChannelHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("from server : ClassName - " + msg.getClass().getName()
+ " ; message : " + msg.toString());
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("client exceptionCaught method run...");
cause.printStackTrace();
ctx.close();
}
}
客户端断线重连机制。
客户端数量多，且需要传递的数据量级较大。可以周期性的发送数据的时候，使用。要求对数据的即时性不高的时候，才可使用。
优点： 可以使用数据缓存。不是每条数据进行一次数据交互。可以定时回收资源，对资源利用率高。相对来说，即时性可以通过其他方式保证。如： 120秒自动断线。数据变化1000次请求服务器一次。300秒中自动发送不足1000次的变化数据。
/**
* 1. 双线程组
* 2. Bootstrap配置启动信息
* 3. 注册业务处理Handler
* 4. 绑定服务监听端口并启动服务
*/
package com.hhxy.netty.timer;
import com.hhxy.utils.SerializableFactory4Marshalling;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
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;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.ReadTimeoutHandler;
public class Server4Timer {
// 监听线程组，监听客户端请求
private EventLoopGroup acceptorGroup = null;
// 处理客户端相关操作线程组，负责处理与客户端的数据通讯
private EventLoopGroup clientGroup = null;
// 服务启动相关配置信息
private ServerBootstrap bootstrap = null;
public Server4Timer(){
init();
}
private void init(){
acceptorGroup = new NioEventLoopGroup();
clientGroup = new NioEventLoopGroup();
bootstrap = new ServerBootstrap();
// 绑定线程组
bootstrap.group(acceptorGroup, clientGroup);
// 设定通讯模式为NIO
bootstrap.channel(NioServerSocketChannel.class);
// 设定缓冲区大小
bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
// SO_SNDBUF发送缓冲区，SO_RCVBUF接收缓冲区，SO_KEEPALIVE开启心跳监测（保证连接有效）
bootstrap.option(ChannelOption.SO_SNDBUF, 16*1024)
.option(ChannelOption.SO_RCVBUF, 16*1024)
.option(ChannelOption.SO_KEEPALIVE, true);
// 增加日志Handler，日志级别为info
// bootstrap.handler(new LoggingHandler(LogLevel.INFO));
}
public ChannelFuture doAccept(int port) throws InterruptedException{
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(SerializableFactory4Marshalling.buildMarshallingDecoder());
ch.pipeline().addLast(SerializableFactory4Marshalling.buildMarshallingEncoder());
// 定义一个定时断线处理器，当多长时间内，没有任何的可读取数据，自动断开连接。
// 构造参数，就是间隔时长。 默认的单位是秒。
// 自定义间隔时长单位。 new ReadTimeoutHandler(long times, TimeUnit unit);
ch.pipeline().addLast(new ReadTimeoutHandler(3));
ch.pipeline().addLast(new Server4TimerHandler());
}
});
ChannelFuture future = bootstrap.bind(port).sync();
return future;
}
public void release(){
this.acceptorGroup.shutdownGracefully();
this.clientGroup.shutdownGracefully();
}
public static void main(String[] args){
ChannelFuture future = null;
Server4Timer server = null;
try{
server = new Server4Timer();
future = server.doAccept(9999);
System.out.println("server started.");
future.channel().closeFuture().sync();
}catch(InterruptedException e){
e.printStackTrace();
}finally{
if(null != future){
try {
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
if(null != server){
server.release();
}
}
}
}
/**
* @Sharable注解 -
* 代表当前Handler是一个可以分享的处理器。也就意味着，服务器注册此Handler后，可以分享给多个客户端同时使用。
* 如果不使用注解描述类型，则每次客户端请求时，必须为客户端重新创建一个新的Handler对象。
*
*/
package com.hhxy.netty.timer;
import io.netty.channel.ChannelHandler.Sharable;
import com.hhxy.utils.GzipUtils;
import com.hhxy.utils.RequestMessage;
import com.hhxy.utils.ResponseMessage;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
@Sharable
public class Server4TimerHandler extends ChannelHandlerAdapter {
// 业务处理逻辑
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("from client : ClassName - " + msg.getClass().getName()
+ " ; message : " + msg.toString());
ResponseMessage response = new ResponseMessage(0L, "test response");
ctx.writeAndFlush(response);
}
// 异常处理逻辑
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("server exceptionCaught method run...");
// cause.printStackTrace();
ctx.close();
}
}
/**
* 1. 单线程组
* 2. Bootstrap配置启动信息
* 3. 注册业务处理Handler
* 4. connect连接服务，并发起请求
*/
package com.hhxy.netty.timer;
import java.hhxy.Random;
import java.hhxy.concurrent.TimeUnit;
import com.hhxy.utils.RequestMessage;
import com.hhxy.utils.SerializableFactory4Marshalling;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.WriteTimeoutHandler;
public class Client4Timer {
// 处理请求和处理服务端响应的线程组
private EventLoopGroup group = null;
// 服务启动相关配置信息
private Bootstrap bootstrap = null;
private ChannelFuture future = null;
public Client4Timer(){
init();
}
private void init(){
group = new NioEventLoopGroup();
bootstrap = new Bootstrap();
// 绑定线程组
bootstrap.group(group);
// 设定通讯模式为NIO
bootstrap.channel(NioSocketChannel.class);
// bootstrap.handler(new LoggingHandler(LogLevel.INFO));
}
public void setHandlers() throws InterruptedException{
this.bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(SerializableFactory4Marshalling.buildMarshallingDecoder());
ch.pipeline().addLast(SerializableFactory4Marshalling.buildMarshallingEncoder());
// 写操作自定断线。 在指定时间内，没有写操作，自动断线。
ch.pipeline().addLast(new WriteTimeoutHandler(3));
ch.pipeline().addLast(new Client4TimerHandler());
}
});
}
public ChannelFuture getChannelFuture(String host, int port) throws InterruptedException{
if(future == null){
future = this.bootstrap.connect(host, port).sync();
}
if(!future.channel().isActive()){
future = this.bootstrap.connect(host, port).sync();
}
return future;
}
public void release(){
this.group.shutdownGracefully();
}
public static void main(String[] args) {
Client4Timer client = null;
ChannelFuture future = null;
try{
client = new Client4Timer();
client.setHandlers();
future = client.getChannelFuture("localhost", 9999);
for(int i = 0; i < 3; i++){
RequestMessage msg = new RequestMessage(new Random().nextLong(),
"test"+i, new byte[0]);
future.channel().writeAndFlush(msg);
TimeUnit.SECONDS.sleep(2);
}
TimeUnit.SECONDS.sleep(5);
future = client.getChannelFuture("localhost", 9999);
RequestMessage msg = new RequestMessage(new Random().nextLong(),
"test", new byte[0]);
future.channel().writeAndFlush(msg);
}catch(Exception e){
e.printStackTrace();
}finally{
if(null != future){
try {
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
if(null != client){
client.release();
}
}
}
}
package com.hhxy.netty.timer;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
public class Client4TimerHandler extends ChannelHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("from server : ClassName - " + msg.getClass().getName()
+ " ; message : " + msg.toString());
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("client exceptionCaught method run...");
cause.printStackTrace();
ctx.close();
}
/**
* 当连接建立成功后，出发的代码逻辑。
* 在一次连接中只运行唯一一次。
* 通常用于实现连接确认和资源初始化的。
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("client channel active");
}
}
使用定时发送消息的方式，实现硬件检测，达到心态检测的目的。
心跳监测是用于检测电脑硬件和软件信息的一种技术。如：CPU使用率，磁盘使用率，内存使用率，进程情况，线程情况等。
需要下载一个zip压缩包。内部包含若干sigar需要的操作系统文件。sigar插件是通过JVM访问操作系统，读取计算机硬件的一个插件库。读取计算机硬件过程中，必须由操作系统提供硬件信息。硬件信息是通过操作系统提供的。zip压缩包中是sigar编写的操作系统文件，如：windows中的动态链接库文件。
解压需要的操作系统文件，将操作系统文件赋值到${Java_home}/bin目录中。
/**
* 1. 双线程组
* 2. Bootstrap配置启动信息
* 3. 注册业务处理Handler
* 4. 绑定服务监听端口并启动服务
*/
package com.hhxy.netty.heatbeat;
import com.hhxy.utils.SerializableFactory4Marshalling;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
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 Server4Heatbeat {
// 监听线程组，监听客户端请求
private EventLoopGroup acceptorGroup = null;
// 处理客户端相关操作线程组，负责处理与客户端的数据通讯
private EventLoopGroup clientGroup = null;
// 服务启动相关配置信息
private ServerBootstrap bootstrap = null;
public Server4Heatbeat(){
init();
}
private void init(){
acceptorGroup = new NioEventLoopGroup();
clientGroup = new NioEventLoopGroup();
bootstrap = new ServerBootstrap();
// 绑定线程组
bootstrap.group(acceptorGroup, clientGroup);
// 设定通讯模式为NIO
bootstrap.channel(NioServerSocketChannel.class);
// 设定缓冲区大小
bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
// SO_SNDBUF发送缓冲区，SO_RCVBUF接收缓冲区，SO_KEEPALIVE开启心跳监测（保证连接有效）
bootstrap.option(ChannelOption.SO_SNDBUF, 16*1024)
.option(ChannelOption.SO_RCVBUF, 16*1024)
.option(ChannelOption.SO_KEEPALIVE, true);
}
public ChannelFuture doAccept(int port) throws InterruptedException{
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(SerializableFactory4Marshalling.buildMarshallingDecoder());
ch.pipeline().addLast(SerializableFactory4Marshalling.buildMarshallingEncoder());
ch.pipeline().addLast(new Server4HeatbeatHandler());
}
});
ChannelFuture future = bootstrap.bind(port).sync();
return future;
}
public void release(){
this.acceptorGroup.shutdownGracefully();
this.clientGroup.shutdownGracefully();
}
public static void main(String[] args){
ChannelFuture future = null;
Server4Heatbeat server = null;
try{
server = new Server4Heatbeat();
future = server.doAccept(9999);
System.out.println("server started.");
future.channel().closeFuture().sync();
}catch(InterruptedException e){
e.printStackTrace();
}finally{
if(null != future){
try {
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
if(null != server){
server.release();
}
}
}
}
/**
* @Sharable注解 -
* 代表当前Handler是一个可以分享的处理器。也就意味着，服务器注册此Handler后，可以分享给多个客户端同时使用。
* 如果不使用注解描述类型，则每次客户端请求时，必须为客户端重新创建一个新的Handler对象。
*
*/
package com.hhxy.netty.heatbeat;
import java.util.ArrayList;
import java.util.List;
import com.hhxy.utils.HeatbeatMessage;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
@Sharable
public class Server4HeatbeatHandler extends ChannelHandlerAdapter {
private static List<String> credentials = new ArrayList<>();
private static final String HEATBEAT_SUCCESS = "SERVER_RETURN_HEATBEAT_SUCCESS";
public Server4HeatbeatHandler(){
// 初始化客户端列表信息。一般通过配置文件读取或数据库读取。
credentials.add("192.168.199.222_WIN-QIUB2JF5TDP");
}
// 业务处理逻辑
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if(msg instanceof String){
this.checkCredential(ctx, msg.toString());
} else if (msg instanceof HeatbeatMessage){
this.readHeatbeatMessage(ctx, msg);
} else {
ctx.writeAndFlush("wrong message").addListener(ChannelFutureListener.CLOSE);
}
}
private void readHeatbeatMessage(ChannelHandlerContext ctx, Object msg){
HeatbeatMessage message = (HeatbeatMessage) msg;
System.out.println(message);
System.out.println("=======================================");
ctx.writeAndFlush("receive heatbeat message");
}
/**
* 身份检查。检查客户端身份是否有效。
* 客户端身份信息应该是通过数据库或数据文件定制的。
* 身份通过 - 返回确认消息。
* 身份无效 - 断开连接
* @param ctx
* @param credential
*/
private void checkCredential(ChannelHandlerContext ctx, String credential){
System.out.println(credential);
System.out.println(credentials);
if(credentials.contains(credential)){
ctx.writeAndFlush(HEATBEAT_SUCCESS);
}else{
ctx.writeAndFlush("no credential contains").addListener(ChannelFutureListener.CLOSE);
}
}
// 异常处理逻辑
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("server exceptionCaught method run...");
// cause.printStackTrace();
ctx.close();
}
}
/**
* 1. 单线程组
* 2. Bootstrap配置启动信息
* 3. 注册业务处理Handler
* 4. connect连接服务，并发起请求
*/
package com.hhxy.netty.heatbeat;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import com.hhxy.utils.GzipUtils;
import com.hhxy.utils.RequestMessage;
import com.hhxy.utils.SerializableFactory4Marshalling;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
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 Client4Heatbeat {
// 处理请求和处理服务端响应的线程组
private EventLoopGroup group = null;
// 服务启动相关配置信息
private Bootstrap bootstrap = null;
public Client4Heatbeat(){
init();
}
private void init(){
group = new NioEventLoopGroup();
bootstrap = new Bootstrap();
// 绑定线程组
bootstrap.group(group);
// 设定通讯模式为NIO
bootstrap.channel(NioSocketChannel.class);
}
public ChannelFuture doRequest(String host, int port) throws InterruptedException{
this.bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(SerializableFactory4Marshalling.buildMarshallingDecoder());
ch.pipeline().addLast(SerializableFactory4Marshalling.buildMarshallingEncoder());
ch.pipeline().addLast(new Client4HeatbeatHandler());
}
});
ChannelFuture future = this.bootstrap.connect(host, port).sync();
return future;
}
public void release(){
this.group.shutdownGracefully();
}
public static void main(String[] args) {
Client4Heatbeat client = null;
ChannelFuture future = null;
try{
client = new Client4Heatbeat();
future = client.doRequest("localhost", 9999);
}catch(Exception e){
e.printStackTrace();
}finally{
if(null != future){
try {
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
if(null != client){
client.release();
}
}
}
}
package com.hhxy.netty.heatbeat;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import org.hyperic.sigar.CpuPerc;
import org.hyperic.sigar.FileSystem;
import org.hyperic.sigar.Mem;
import org.hyperic.sigar.Sigar;
import com.hhxy.utils.HeatbeatMessage;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.ReferenceCountUtil;
public class Client4HeatbeatHandler extends ChannelHandlerAdapter {
private ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
private ScheduledFuture heatbeat;
private InetAddress remoteAddr;
private static final String HEATBEAT_SUCCESS = "SERVER_RETURN_HEATBEAT_SUCCESS";
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
// 获取本地INET信息
this.remoteAddr = InetAddress.getLocalHost();
// 获取本地计算机名
String computerName = System.getenv().get("COMPUTERNAME");
String credentials = this.remoteAddr.getHostAddress() + "_" + computerName;
System.out.println(credentials);
// 发送到服务器，作为信息比对证书
ctx.writeAndFlush(credentials);
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
try{
if(msg instanceof String){
if(HEATBEAT_SUCCESS.equals(msg)){
this.heatbeat = this.executorService.scheduleWithFixedDelay(new HeatbeatTask(ctx), 0L, 2L, TimeUnit.SECONDS);
System.out.println("client receive - " + msg);
}else{
System.out.println("client receive - " + msg);
}
}
}finally{
ReferenceCountUtil.release(msg);
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("client exceptionCaught method run...");
// cause.printStackTrace();
// 回收资源
if(this.heatbeat != null){
this.heatbeat.cancel(true);
this.heatbeat = null;
}
ctx.close();
}
class HeatbeatTask implements Runnable{
private ChannelHandlerContext ctx;
public HeatbeatTask(){
}
public HeatbeatTask(ChannelHandlerContext ctx){
this.ctx = ctx;
}
public void run(){
try {
HeatbeatMessage msg = new HeatbeatMessage();
msg.setIp(remoteAddr.getHostAddress());
Sigar sigar = new Sigar();
// CPU信息
CpuPerc cpuPerc = sigar.getCpuPerc();
Map<String, Object> cpuMsgMap = new HashMap<>();
cpuMsgMap.put("Combined", cpuPerc.getCombined());
cpuMsgMap.put("User", cpuPerc.getUser());
cpuMsgMap.put("Sys", cpuPerc.getSys());
cpuMsgMap.put("Wait", cpuPerc.getWait());
cpuMsgMap.put("Idle", cpuPerc.getIdle());
// 内存信息
Map<String, Object> memMsgMap = new HashMap<>();
Mem mem = sigar.getMem();
memMsgMap.put("Total", mem.getTotal());
memMsgMap.put("Used", mem.getUsed());
memMsgMap.put("Free", mem.getFree());
// 文件系统
Map<String, Object> fileSysMsgMap = new HashMap<>();
FileSystem[] list = sigar.getFileSystemList();
fileSysMsgMap.put("FileSysCount", list.length);
List<String> msgList = null;
for(FileSystem fs : list){
msgList = new ArrayList<>();
msgList.add(fs.getDevName() + "总大小: " + sigar.getFileSystemUsage(fs.getDirName()).getTotal() + "KB");
msgList.add(fs.getDevName() + "剩余大小: " + sigar.getFileSystemUsage(fs.getDirName()).getFree() + "KB");
fileSysMsgMap.put(fs.getDevName(), msgList);
}
msg.setCpuMsgMap(cpuMsgMap);
msg.setMemMsgMap(memMsgMap);
msg.setFileSysMsgMap(fileSysMsgMap);
ctx.writeAndFlush(msg);
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
使用Netty服务开发。实现HTTP协议处理逻辑。
package com.hhxy.netty.http;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
/**
* http协议文件传输
* @author Qixuan.Chen
* 创建时间：2015年5月4日
*/
public class HttpStaticFileServer {
private final int port;//端口
public HttpStaticFileServer(int port) {
this.port = port;
}
public void run() throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup();//线程一 //这个是用于serversocketchannel的event
EventLoopGroup workerGroup = new NioEventLoopGroup();//线程二//这个是用于处理accept到的channel
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new HttpStaticFileServerInitializer());
b.bind(port).sync().channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws Exception {
int port = 8089;
if (args.length > 0) {
port = Integer.parseInt(args[0]);
} else {
port = 8089;
}
new HttpStaticFileServer(port).run();//启动服务
}
} package com.hhxy.netty.http;
import static io.netty.handler.codec.http.HttpHeaderNames.CACHE_CONTROL;
import static io.netty.handler.codec.http.HttpHeaderNames.CONNECTION;
import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_TYPE;
import static io.netty.handler.codec.http.HttpHeaderNames.DATE;
import static io.netty.handler.codec.http.HttpHeaderNames.EXPIRES;
import static io.netty.handler.codec.http.HttpHeaderNames.IF_MODIFIED_SINCE;
import static io.netty.handler.codec.http.HttpHeaderNames.LAST_MODIFIED;
import static io.netty.handler.codec.http.HttpHeaderNames.LOCATION;
import static io.netty.handler.codec.http.HttpMethod.GET;
import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
import static io.netty.handler.codec.http.HttpResponseStatus.FORBIDDEN;
import static io.netty.handler.codec.http.HttpResponseStatus.FOUND;
import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;
import static io.netty.handler.codec.http.HttpResponseStatus.METHOD_NOT_ALLOWED;
import static io.netty.handler.codec.http.HttpResponseStatus.NOT_FOUND;
import static io.netty.handler.codec.http.HttpResponseStatus.NOT_MODIFIED;
import static io.netty.handler.codec.http.HttpResponseStatus.OK;
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.RandomAccessFile;
import java.io.UnsupportedEncodingException;
import java.net.URLDecoder;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.GregorianCalendar;
import java.util.Locale;
import java.util.TimeZone;
import java.util.regex.Pattern;
import javax.activation.MimetypesFileTypeMap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelProgressiveFuture;
import io.netty.channel.ChannelProgressiveFutureListener;
import io.netty.channel.DefaultFileRegion;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.DefaultHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaderUtil;
import io.netty.handler.codec.http.HttpHeaderValues;
import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.LastHttpContent;
import io.netty.handler.stream.ChunkedFile;
import io.netty.util.CharsetUtil;
/**
* A simple handler that serves incoming HTTP requests to send their respective
* HTTP responses. It also implements {@code 'If-Modified-Since'} header to
* take advantage of browser cache, as described in
* <a href="http://tools.ietf.org/html/rfc2616#section-14.25">RFC 2616</a>.
*
* <h3>How Browser Caching Works</h3>
*
* Web browser caching works with HTTP headers as illustrated by the following
* sample:
* <ol>
* <li>Request #1 returns the content of {@code /file1.txt}.</li>
* <li>Contents of {@code /file1.txt} is cached by the browser.</li>
* <li>Request #2 for {@code /file1.txt} does return the contents of the
* file again. Rather, a 304 Not Modified is returned. This tells the
* browser to use the contents stored in its cache.</li>
* <li>The server knows the file has not been modified because the
* {@code If-Modified-Since} date is the same as the file's last
* modified date.</li>
* </ol>
*
* <pre>
* Request #1 Headers
* ===================
* GET /file1.txt HTTP/1.1
*
* Response #1 Headers
* ===================
* HTTP/1.1 200 OK
* Date: Tue, 01 Mar 2011 22:44:26 GMT
* Last-Modified: Wed, 30 Jun 2010 21:36:48 GMT
* Expires: Tue, 01 Mar 2012 22:44:26 GMT
* Cache-Control: private, max-age=31536000
*
* Request #2 Headers
* ===================
* GET /file1.txt HTTP/1.1
* If-Modified-Since: Wed, 30 Jun 2010 21:36:48 GMT
*
* Response #2 Headers
* ===================
* HTTP/1.1 304 Not Modified
* Date: Tue, 01 Mar 2011 22:44:28 GMT
*
* </pre>
*/
public class HttpStaticFileServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
public static final String HTTP_DATE_FORMAT = "EEE, dd MMM yyyy HH:mm:ss zzz";
public static final String HTTP_DATE_GMT_TIMEZONE = "GMT";
public static final int HTTP_CACHE_SECONDS = 60;
private final boolean useSendFile;
public HttpStaticFileServerHandler(boolean useSendFile) {
this.useSendFile = useSendFile;
}
/**
* 类似channelRead方法。
*/
@Override
public void messageReceived(
ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {
if (!request.decoderResult().isSuccess()) {
sendError(ctx, BAD_REQUEST);
return;
}
if (request.method() != GET) {
sendError(ctx, METHOD_NOT_ALLOWED);
return;
}
final String uri = request.uri();
System.out.println("-----uri----"+uri);
final String path = sanitizeUri(uri);
System.out.println("-----path----"+path);
if (path == null) {
sendError(ctx, FORBIDDEN);
return;
}
File file = new File(path);
if (file.isHidden() || !file.exists()) {
sendError(ctx, NOT_FOUND);
return;
}
if (file.isDirectory()) {
if (uri.endsWith("/")) {
sendListing(ctx, file);
} else {
sendRedirect(ctx, uri + '/');
}
return;
}
if (!file.isFile()) {
sendError(ctx, FORBIDDEN);
return;
}
// Cache Validation
String ifModifiedSince = (String) request.headers().get(IF_MODIFIED_SINCE);
if (ifModifiedSince != null && !ifModifiedSince.isEmpty()) {
SimpleDateFormat dateFormatter = new SimpleDateFormat(HTTP_DATE_FORMAT, Locale.US);
Date ifModifiedSinceDate = dateFormatter.parse(ifModifiedSince);
// Only compare up to the second because the datetime format we send to the client
// does not have milliseconds
long ifModifiedSinceDateSeconds = ifModifiedSinceDate.getTime() / 1000;
long fileLastModifiedSeconds = file.lastModified() / 1000;
if (ifModifiedSinceDateSeconds == fileLastModifiedSeconds) {
sendNotModified(ctx);
return;
}
}
RandomAccessFile raf;
try {
raf = new RandomAccessFile(file, "r");
} catch (FileNotFoundException fnfe) {
sendError(ctx, NOT_FOUND);
return;
}
long fileLength = raf.length();
HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK);
//setContentLength(response, fileLength);
HttpHeaderUtil.setContentLength(response, fileLength);
setContentTypeHeader(response, file);
setDateAndCacheHeaders(response, file);
if (HttpHeaderUtil.isKeepAlive(request)) {
response.headers().set(CONNECTION, HttpHeaderValues.KEEP_ALIVE);
}
// Write the initial line and the header.
ctx.write(response);
// Write the content.
ChannelFuture sendFileFuture;
if (useSendFile) {
sendFileFuture =
ctx.write(new DefaultFileRegion(raf.getChannel(), 0, fileLength), ctx.newProgressivePromise());
} else {
sendFileFuture =
ctx.write(new ChunkedFile(raf, 0, fileLength, 8192), ctx.newProgressivePromise());
}
sendFileFuture.addListener(new ChannelProgressiveFutureListener() {
@Override
public void operationProgressed(ChannelProgressiveFuture future, long progress, long total) {
if (total < 0) { // total unknown
System.err.println("Transfer progress: " + progress);
} else {
System.err.println("Transfer progress: " + progress + " / " + total);
}
}
@Override
public void operationComplete(ChannelProgressiveFuture future) throws Exception {
System.err.println("Transfer complete.");
}
});
// Write the end marker
ChannelFuture lastContentFuture = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
// Decide whether to close the connection or not.
if (!HttpHeaderUtil.isKeepAlive(request)) {
// Close the connection when the whole content is written out.
lastContentFuture.addListener(ChannelFutureListener.CLOSE);
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
if (ctx.channel().isActive()) {
sendError(ctx, INTERNAL_SERVER_ERROR);
}
}
private static final Pattern INSECURE_URI = Pattern.compile(".*[<>&\"].*");
/**
* 路径解码
* @param uri
* @return
*/
private static String sanitizeUri(String uri) {
// Decode the path.
try {
uri = URLDecoder.decode(uri, "UTF-8");
} catch (UnsupportedEncodingException e) {
try {
uri = URLDecoder.decode(uri, "ISO-8859-1");
} catch (UnsupportedEncodingException e1) {
throw new Error();
}
}
if (!uri.startsWith("/")) {
return null;
}
// Convert file separators.
uri = uri.replace('/', File.separatorChar);
// Simplistic dumb security check.
// You will have to do something serious in the production environment.
if (uri.contains(File.separator + '.') ||
uri.contains('.' + File.separator) ||
uri.startsWith(".") || uri.endsWith(".") ||
INSECURE_URI.matcher(uri).matches()) {
return null;
}
// Convert to absolute path.
return System.getProperty("user.dir") + File.separator + uri;
}
private static final Pattern ALLOWED_FILE_NAME = Pattern.compile("[A-Za-z0-9][-_A-Za-z0-9\\.]*");
private static void sendListing(ChannelHandlerContext ctx, File dir) {
FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, OK);
response.headers().set(CONTENT_TYPE, "text/html; charset=UTF-8");
StringBuilder buf = new StringBuilder();
String dirPath = dir.getPath();
buf.append("<!DOCTYPE html>\r\n");
buf.append("<html><head><title>");
buf.append("Listing of: ");
buf.append(dirPath);
buf.append("</title></head><body>\r\n");
buf.append("<h3>Listing of: ");
buf.append(dirPath);
buf.append("</h3>\r\n");
buf.append("<ul>");
buf.append("<li><a href=\"../\">..</a></li>\r\n");
for (File f: dir.listFiles()) {
if (f.isHidden() || !f.canRead()) {
continue;
}
String name = f.getName();
if (!ALLOWED_FILE_NAME.matcher(name).matches()) {
continue;
}
buf.append("<li><a href=\"");
buf.append(name);
buf.append("\">");
buf.append(name);
buf.append("</a></li>\r\n");
}
buf.append("</ul></body></html>\r\n");
ByteBuf buffer = Unpooled.copiedBuffer(buf, CharsetUtil.UTF_8);
response.content().writeBytes(buffer);
buffer.release();
// Close the connection as soon as the error message is sent.
ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}
private static void sendRedirect(ChannelHandlerContext ctx, String newUri) {
FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, FOUND);
response.headers().set(LOCATION, newUri);
// Close the connection as soon as the error message is sent.
ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}
private static void sendError(ChannelHandlerContext ctx, HttpResponseStatus status) {
FullHttpResponse response = new DefaultFullHttpResponse(
HTTP_1_1, status, Unpooled.copiedBuffer("Failure: " + status.toString() + "\r\n", CharsetUtil.UTF_8));
response.headers().set(CONTENT_TYPE, "text/plain; charset=UTF-8");
// Close the connection as soon as the error message is sent.
ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}
/**
* When file timestamp is the same as what the browser is sending up, send a "304 Not Modified"
*
* @param ctx
* Context
*/
private static void sendNotModified(ChannelHandlerContext ctx) {
FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, NOT_MODIFIED);
setDateHeader(response);
// Close the connection as soon as the error message is sent.
ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}
/**
* Sets the Date header for the HTTP response
*
* @param response
* HTTP response
*/
private static void setDateHeader(FullHttpResponse response) {
SimpleDateFormat dateFormatter = new SimpleDateFormat(HTTP_DATE_FORMAT, Locale.US);
dateFormatter.setTimeZone(TimeZone.getTimeZone(HTTP_DATE_GMT_TIMEZONE));
Calendar time = new GregorianCalendar();
response.headers().set(DATE, dateFormatter.format(time.getTime()));
}
/**
* Sets the Date and Cache headers for the HTTP Response
*
* @param response
* HTTP response
* @param fileToCache
* file to extract content type
*/
private static void setDateAndCacheHeaders(HttpResponse response, File fileToCache) {
SimpleDateFormat dateFormatter = new SimpleDateFormat(HTTP_DATE_FORMAT, Locale.US);
dateFormatter.setTimeZone(TimeZone.getTimeZone(HTTP_DATE_GMT_TIMEZONE));
// Date header
Calendar time = new GregorianCalendar();
response.headers().set(DATE, dateFormatter.format(time.getTime()));
// Add cache headers
time.add(Calendar.SECOND, HTTP_CACHE_SECONDS);
response.headers().set(EXPIRES, dateFormatter.format(time.getTime()));
response.headers().set(CACHE_CONTROL, "private, max-age=" + HTTP_CACHE_SECONDS);
response.headers().set(
LAST_MODIFIED, dateFormatter.format(new Date(fileToCache.lastModified())));
}
/**
* Sets the content type header for the HTTP Response
*
* @param response
* HTTP response
* @param file
* file to extract content type
*/
private static void setContentTypeHeader(HttpResponse response, File file) {
MimetypesFileTypeMap mimeTypesMap = new MimetypesFileTypeMap();
response.headers().set(CONTENT_TYPE, mimeTypesMap.getContentType(file.getPath()));
}
} package com.hhxy.netty.http;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpRequestDecoder;
import io.netty.handler.codec.http.HttpResponseEncoder;
import io.netty.handler.stream.ChunkedWriteHandler;
public class HttpStaticFileServerInitializer extends ChannelInitializer<SocketChannel> {
@Override
public void initChannel(SocketChannel ch) throws Exception {
// Create a default pipeline implementation.
ChannelPipeline pipeline = ch.pipeline();
// Uncomment the following line if you want HTTPS
//SSLEngine engine = SecureChatSslContextFactory.getServerContext().createSSLEngine();
//engine.setUseClientMode(false);
//pipeline.addLast("ssl", new SslHandler(engine));
/**
* （1）ReadTimeoutHandler，用于控制读取数据的时候的超时，10表示如果10秒钟都没有数据读取了，那么就引发超时，然后关闭当前的channel
（2）WriteTimeoutHandler，用于控制数据输出的时候的超时，构造参数1表示如果持续1秒钟都没有数据写了，那么就超时。
（3）HttpRequestrianDecoder，这个handler用于从读取的数据中将http报文信息解析出来，无非就是什么requestline，header，body什么的。。。
（4）然后HttpObjectAggregator则是用于将上卖解析出来的http报文的数据组装成为封装好的httprequest对象。。
（5）HttpresponseEncoder，用于将用户返回的httpresponse编码成为http报文格式的数据
（6）HttpHandler，自定义的handler，用于处理接收到的http请求。
*/
pipeline.addLast("decoder", new HttpRequestDecoder());// http-request解码器,http服务器端对request解码
pipeline.addLast("aggregator", new HttpObjectAggregator(65536));//对传输文件大少进行限制
pipeline.addLast("encoder", new HttpResponseEncoder());//http-response解码器,http服务器端对response编码
// 向客户端发送数据的一个Handler
pipeline.addLast("chunkedWriter", new ChunkedWriteHandler());
pipeline.addLast("handler", new HttpStaticFileServerHandler(true)); // Specify false if SSL.(如果是ssl,就指定为false)
}
} package com.hhxy.utils;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;
public class GzipUtils {
public static void main(String[] args) throws Exception {
FileInputStream fis = new FileInputStream("D:\\3\\1.jpg");
byte[] temp = new byte[fis.available()];
int length = fis.read(temp);
System.out.println("长度 : " + length);
byte[] zipArray = GzipUtils.zip(temp);
System.out.println("压缩后的长度 : " + zipArray.length);
byte[] unzipArray = GzipUtils.unzip(zipArray);
System.out.println("解压缩后的长度 : " + unzipArray.length);
FileOutputStream fos = new FileOutputStream("D:\\3\\101.jpg");
fos.write(unzipArray);
fos.flush();
fos.close();
fis.close();
}
/**
* 解压缩
* @param source 源数据。需要解压的数据。
* @return 解压后的数据。 恢复的数据。
* @throws Exception
*/
public static byte[] unzip(byte[] source) throws Exception{
ByteArrayOutputStream out = new ByteArrayOutputStream();
ByteArrayInputStream in = new ByteArrayInputStream(source);
// JDK提供的。 专门用于压缩使用的流对象。可以处理字节数组数据。
GZIPInputStream zipIn = new GZIPInputStream(in);
byte[] temp = new byte[256];
int length = 0;
while((length = zipIn.read(temp, 0, temp.length)) != -1){
out.write(temp, 0, length);
}
// 将字节数组输出流中的数据，转换为一个字节数组。
byte[] target = out.toByteArray();
zipIn.close();
out.close();
return target;
}
/**
* 压缩
* @param source 源数据，需要压缩的数据
* @return 压缩后的数据。
* @throws Exception
*/
public static byte[] zip(byte[] source) throws Exception{
ByteArrayOutputStream out = new ByteArrayOutputStream();
// 输出流，JDK提供的，提供解压缩功能。
GZIPOutputStream zipOut = new GZIPOutputStream(out);
// 将压缩信息写入到内存。 写入的过程会实现解压。
zipOut.write(source);
// 结束。
zipOut.finish();
byte[] target = out.toByteArray();
zipOut.close();
return target;
}
}
package com.hhxy.utils;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Map;
import java.util.Properties;
import org.hyperic.sigar.CpuInfo;
import org.hyperic.sigar.CpuPerc;
import org.hyperic.sigar.FileSystem;
import org.hyperic.sigar.FileSystemUsage;
import org.hyperic.sigar.Mem;
import org.hyperic.sigar.NetFlags;
import org.hyperic.sigar.NetInterfaceConfig;
import org.hyperic.sigar.NetInterfaceStat;
import org.hyperic.sigar.OperatingSystem;
import org.hyperic.sigar.Sigar;
import org.hyperic.sigar.SigarException;
import org.hyperic.sigar.Swap;
import org.hyperic.sigar.Who;
public class OSUtils {
public static void main(String[] args) {
try {
// System信息，从jvm获取
property();
System.out.println("----------------------------------");
// cpu信息
cpu();
System.out.println("----------------------------------");
// 内存信息
memory();
System.out.println("----------------------------------");
// 操作系统信息
os();
System.out.println("----------------------------------");
// 用户信息
who();
System.out.println("----------------------------------");
// 文件系统信息
file();
System.out.println("----------------------------------");
// 网络信息
net();
System.out.println("----------------------------------");
// 以太网信息
ethernet();
System.out.println("----------------------------------");
} catch (Exception e1) {
e1.printStackTrace();
}
}
private static void property() throws UnknownHostException {
Runtime r = Runtime.getRuntime();
Properties props = System.getProperties();
InetAddress addr;
addr = InetAddress.getLocalHost();
String ip = addr.getHostAddress();
Map<String, String> map = System.getenv();
String userName = map.get("USERNAME");// 获取用户名
String computerName = map.get("COMPUTERNAME");// 获取计算机名
String userDomain = map.get("USERDOMAIN");// 获取计算机域名
System.out.println("用户名: " + userName);
System.out.println("计算机名: " + computerName);
System.out.println("计算机域名: " + userDomain);
System.out.println("本地ip地址: " + ip);
System.out.println("本地主机名: " + addr.getHostName());
System.out.println("JVM可以使用的总内存: " + r.totalMemory());
System.out.println("JVM可以使用的剩余内存: " + r.freeMemory());
System.out.println("JVM可以使用的处理器个数: " + r.availableProcessors());
System.out.println("Java的运行环境版本： " + props.getProperty("java.version"));
System.out.println("Java的运行环境供应商： " + props.getProperty("java.vendor"));
System.out.println("Java供应商的URL： " + props.getProperty("java.vendor.url"));
System.out.println("Java的安装路径： " + props.getProperty("java.home"));
System.out.println("Java的虚拟机规范版本： " + props.getProperty("java.vm.specification.version"));
System.out.println("Java的虚拟机规范供应商： " + props.getProperty("java.vm.specification.vendor"));
System.out.println("Java的虚拟机规范名称： " + props.getProperty("java.vm.specification.name"));
System.out.println("Java的虚拟机实现版本： " + props.getProperty("java.vm.version"));
System.out.println("Java的虚拟机实现供应商： " + props.getProperty("java.vm.vendor"));
System.out.println("Java的虚拟机实现名称： " + props.getProperty("java.vm.name"));
System.out.println("Java运行时环境规范版本： " + props.getProperty("java.specification.version"));
System.out.println("Java运行时环境规范供应商： " + props.getProperty("java.specification.vender"));
System.out.println("Java运行时环境规范名称： " + props.getProperty("java.specification.name"));
System.out.println("Java的类格式版本号： " + props.getProperty("java.class.version"));
System.out.println("Java的类路径： " + props.getProperty("java.class.path"));
System.out.println("加载库时搜索的路径列表： " + props.getProperty("java.library.path"));
System.out.println("默认的临时文件路径： " + props.getProperty("java.io.tmpdir"));
System.out.println("一个或多个扩展目录的路径： " + props.getProperty("java.ext.dirs"));
System.out.println("操作系统的名称： " + props.getProperty("os.name"));
System.out.println("操作系统的构架： " + props.getProperty("os.arch"));
System.out.println("操作系统的版本： " + props.getProperty("os.version"));
System.out.println("文件分隔符： " + props.getProperty("file.separator"));
System.out.println("路径分隔符： " + props.getProperty("path.separator"));
System.out.println("行分隔符： " + props.getProperty("line.separator"));
System.out.println("用户的账户名称： " + props.getProperty("user.name"));
System.out.println("用户的主目录： " + props.getProperty("user.home"));
System.out.println("用户的当前工作目录： " + props.getProperty("user.dir"));
}
private static void memory() throws SigarException {
Sigar sigar = new Sigar();
Mem mem = sigar.getMem();
// 内存总量
System.out.println("内存总量: " + mem.getTotal() / 1024L + "K av");
// 当前内存使用量
System.out.println("当前内存使用量: " + mem.getUsed() / 1024L + "K used");
// 当前内存剩余量
System.out.println("当前内存剩余量: " + mem.getFree() / 1024L + "K free");
Swap swap = sigar.getSwap();
// 交换区总量
System.out.println("交换区总量: " + swap.getTotal() / 1024L + "K av");
// 当前交换区使用量
System.out.println("当前交换区使用量: " + swap.getUsed() / 1024L + "K used");
// 当前交换区剩余量
System.out.println("当前交换区剩余量: " + swap.getFree() / 1024L + "K free");
}
private static void cpu() throws SigarException {
Sigar sigar = new Sigar();
CpuInfo infos[] = sigar.getCpuInfoList();
CpuPerc cpuList[] = null;
cpuList = sigar.getCpuPercList();
for (int i = 0; i < infos.length; i++) {// 不管是单块CPU还是多CPU都适用
CpuInfo info = infos[i];
System.out.println("第" + (i + 1) + "块CPU信息");
System.out.println("CPU的总量MHz: " + info.getMhz());// CPU的总量MHz
System.out.println("CPU生产商: " + info.getVendor());// 获得CPU的卖主，如：Intel
System.out.println("CPU类别: " + info.getModel());// 获得CPU的类别，如：Celeron
System.out.println("CPU缓存数量: " + info.getCacheSize());// 缓冲存储器数量
printCpuPerc(cpuList[i]);
}
}
private static void printCpuPerc(CpuPerc cpu) {
System.out.println("CPU用户使用率: " + CpuPerc.format(cpu.getUser()));// 用户使用率
System.out.println("CPU系统使用率: " + CpuPerc.format(cpu.getSys()));// 系统使用率
System.out.println("CPU当前等待率: " + CpuPerc.format(cpu.getWait()));// 当前等待率
System.out.println("CPU当前错误率: " + CpuPerc.format(cpu.getNice()));//
System.out.println("CPU当前空闲率: " + CpuPerc.format(cpu.getIdle()));// 当前空闲率
System.out.println("CPU总的使用率: " + CpuPerc.format(cpu.getCombined()));// 总的使用率
}
private static void os() {
OperatingSystem OS = OperatingSystem.getInstance();
// 操作系统内核类型如： 386、486、586等x86
System.out.println("操作系统: " + OS.getArch());
System.out.println("操作系统CpuEndian(): " + OS.getCpuEndian());//
System.out.println("操作系统DataModel(): " + OS.getDataModel());//
// 系统描述
System.out.println("操作系统的描述: " + OS.getDescription());
// 操作系统类型
// System.out.println("OS.getName(): " + OS.getName());
// System.out.println("OS.getPatchLevel(): " + OS.getPatchLevel());//
// 操作系统的卖主
System.out.println("操作系统的卖主: " + OS.getVendor());
// 卖主名称
System.out.println("操作系统的卖主名: " + OS.getVendorCodeName());
// 操作系统名称
System.out.println("操作系统名称: " + OS.getVendorName());
// 操作系统卖主类型
System.out.println("操作系统卖主类型: " + OS.getVendorVersion());
// 操作系统的版本号
System.out.println("操作系统的版本号: " + OS.getVersion());
}
private static void who() throws SigarException {
Sigar sigar = new Sigar();
Who who[] = sigar.getWhoList();
if (who != null && who.length > 0) {
for (int i = 0; i < who.length; i++) {
// System.out.println("当前系统进程表中的用户名" + String.valueOf(i));
Who _who = who[i];
System.out.println("用户控制台: " + _who.getDevice());
System.out.println("用户host: " + _who.getHost());
// System.out.println("getTime(): " + _who.getTime());
// 当前系统进程表中的用户名
System.out.println("当前系统进程表中的用户名: " + _who.getUser());
}
}
}
private static void file() throws Exception {
Sigar sigar = new Sigar();
FileSystem fslist[] = sigar.getFileSystemList();
try {
for (int i = 0; i < fslist.length; i++) {
System.out.println("分区的盘符名称" + i);
FileSystem fs = fslist[i];
// 分区的盘符名称
System.out.println("盘符名称: " + fs.getDevName());
// 分区的盘符名称
System.out.println("盘符路径: " + fs.getDirName());
System.out.println("盘符标志: " + fs.getFlags());//
// 文件系统类型，比如 FAT32、NTFS
System.out.println("盘符类型: " + fs.getSysTypeName());
// 文件系统类型名，比如本地硬盘、光驱、网络文件系统等
System.out.println("盘符类型名: " + fs.getTypeName());
// 文件系统类型
System.out.println("盘符文件系统类型: " + fs.getType());
FileSystemUsage usage = null;
usage = sigar.getFileSystemUsage(fs.getDirName());
switch (fs.getType()) {
case 0: // TYPE_UNKNOWN ：未知
break;
case 1: // TYPE_NONE
break;
case 2: // TYPE_LOCAL_DISK : 本地硬盘
// 文件系统总大小
System.out.println(fs.getDevName() + "总大小: " + usage.getTotal() + "KB");
// 文件系统剩余大小
System.out.println(fs.getDevName() + "剩余大小: " + usage.getFree() + "KB");
// 文件系统可用大小
System.out.println(fs.getDevName() + "可用大小: " + usage.getAvail() + "KB");
// 文件系统已经使用量
System.out.println(fs.getDevName() + "已经使用量: " + usage.getUsed() + "KB");
double usePercent = usage.getUsePercent() * 100D;
// 文件系统资源的利用率
System.out.println(fs.getDevName() + "资源的利用率: " + usePercent + "%");
break;
case 3:// TYPE_NETWORK ：网络
break;
case 4:// TYPE_RAM_DISK ：闪存
break;
case 5:// TYPE_CDROM ：光驱
break;
case 6:// TYPE_SWAP ：页面交换
break;
}
System.out.println(fs.getDevName() + "读出： " + usage.getDiskReads());
System.out.println(fs.getDevName() + "写入： " + usage.getDiskWrites());
}
} catch (Exception e) {
// TODO: handle exception
e.printStackTrace();
}
return;
}
private static void net() throws Exception {
Sigar sigar = new Sigar();
String ifNames[] = sigar.getNetInterfaceList();
for (int i = 0; i < ifNames.length; i++) {
String name = ifNames[i];
NetInterfaceConfig ifconfig = sigar.getNetInterfaceConfig(name);
System.out.println("网络设备名: " + name);// 网络设备名
System.out.println("IP地址: " + ifconfig.getAddress());// IP地址
System.out.println("子网掩码: " + ifconfig.getNetmask());// 子网掩码
if ((ifconfig.getFlags() & 1L) <= 0L) {
System.out.println("!IFF_UP...skipping getNetInterfaceStat");
continue;
}
NetInterfaceStat ifstat = sigar.getNetInterfaceStat(name);
System.out.println(name + "接收的总包裹数:" + ifstat.getRxPackets());// 接收的总包裹数
System.out.println(name + "发送的总包裹数:" + ifstat.getTxPackets());// 发送的总包裹数
System.out.println(name + "接收到的总字节数:" + ifstat.getRxBytes());// 接收到的总字节数
System.out.println(name + "发送的总字节数:" + ifstat.getTxBytes());// 发送的总字节数
System.out.println(name + "接收到的错误包数:" + ifstat.getRxErrors());// 接收到的错误包数
System.out.println(name + "发送数据包时的错误数:" + ifstat.getTxErrors());// 发送数据包时的错误数
System.out.println(name + "接收时丢弃的包数:" + ifstat.getRxDropped());// 接收时丢弃的包数
System.out.println(name + "发送时丢弃的包数:" + ifstat.getTxDropped());// 发送时丢弃的包数
}
}
private static void ethernet() throws SigarException {
Sigar sigar = null;
sigar = new Sigar();
String[] ifaces = sigar.getNetInterfaceList();
for (int i = 0; i < ifaces.length; i++) {
NetInterfaceConfig cfg = sigar.getNetInterfaceConfig(ifaces[i]);
if (NetFlags.LOOPBACK_ADDRESS.equals(cfg.getAddress()) || (cfg.getFlags() & NetFlags.IFF_LOOPBACK) != 0
|| NetFlags.NULL_HWADDR.equals(cfg.getHwaddr())) {
continue;
}
System.out.println(cfg.getName() + "IP地址:" + cfg.getAddress());// IP地址
System.out.println(cfg.getName() + "网关广播地址:" + cfg.getBroadcast());// 网关广播地址
System.out.println(cfg.getName() + "网卡MAC地址:" + cfg.getHwaddr());// 网卡MAC地址
System.out.println(cfg.getName() + "子网掩码:" + cfg.getNetmask());// 子网掩码
System.out.println(cfg.getName() + "网卡描述信息:" + cfg.getDescription());// 网卡描述信息
System.out.println(cfg.getName() + "网卡类型" + cfg.getType());//
}
}
}package com.hhxy.utils;
import org.jboss.marshalling.MarshallerFactory;
import org.jboss.marshalling.Marshalling;
import org.jboss.marshalling.MarshallingConfiguration;
import io.netty.handler.codec.marshalling.DefaultMarshallerProvider;
import io.netty.handler.codec.marshalling.DefaultUnmarshallerProvider;
import io.netty.handler.codec.marshalling.MarshallerProvider;
import io.netty.handler.codec.marshalling.MarshallingDecoder;
import io.netty.handler.codec.marshalling.MarshallingEncoder;
import io.netty.handler.codec.marshalling.UnmarshallerProvider;
public class SerializableFactory4Marshalling {
/**
* 创建Jboss Marshalling解码器MarshallingDecoder
* @return MarshallingDecoder
*/
public static MarshallingDecoder buildMarshallingDecoder() {
//首先通过Marshalling工具类的精通方法获取Marshalling实例对象 参数serial标识创建的是java序列化工厂对象。
//jboss-marshalling-serial 包提供
final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
//创建了MarshallingConfiguration对象，配置了版本号为5
final MarshallingConfiguration configuration = new MarshallingConfiguration();
// 序列化版本。只要使用JDK5以上版本，version只能定义为5。
configuration.setVersion(5);
//根据marshallerFactory和configuration创建provider
UnmarshallerProvider provider = new DefaultUnmarshallerProvider(marshallerFactory, configuration);
//构建Netty的MarshallingDecoder对象，俩个参数分别为provider和单个消息序列化后的最大长度
MarshallingDecoder decoder = new MarshallingDecoder(provider, 1024 * 1024 * 1);
return decoder;
}
/**
* 创建Jboss Marshalling编码器MarshallingEncoder
* @return MarshallingEncoder
*/
public static MarshallingEncoder buildMarshallingEncoder() {
final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
final MarshallingConfiguration configuration = new MarshallingConfiguration();
configuration.setVersion(5);
MarshallerProvider provider = new DefaultMarshallerProvider(marshallerFactory, configuration);
//构建Netty的MarshallingEncoder对象，MarshallingEncoder用于实现序列化接口的POJO对象序列化为二进制数组
MarshallingEncoder encoder = new MarshallingEncoder(provider);
return encoder;
}
}
package com.hhxy.utils;
import java.io.Serializable;
public class RequestMessage implements Serializable {
private static final long serialVersionUID = 7084843947860990140L;
private Long id;
private String message;
private byte[] attachment;
@Override
public String toString() {
return "RequestMessage [id=" + id + ", message=" + message + "]";
}
public RequestMessage() {
super();
}
public RequestMessage(Long id, String message, byte[] attachment) {
super();
this.id = id;
this.message = message;
this.attachment = attachment;
}
public Long getId() {
return id;
}
public void setId(Long id) {
this.id = id;
}
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
public byte[] getAttachment() {
return attachment;
}
public void setAttachment(byte[] attachment) {
this.attachment = attachment;
}
}
package com.hhxy.utils;
import java.io.Serializable;
public class ResponseMessage implements Serializable {
private static final long serialVersionUID = -8134313953478922076L;
private Long id;
private String message;
@Override
public String toString() {
return "ResponseMessage [id=" + id + ", message=" + message + "]";
}
public ResponseMessage() {
super();
}
public ResponseMessage(Long id, String message) {
super();
this.id = id;
this.message = message;
}
public Long getId() {
return id;
}
public void setId(Long id) {
this.id = id;
}
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
}
package com.hhxy.utils;
import java.io.Serializable;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Map;
public class HeatbeatMessage implements Serializable {
private static final long serialVersionUID = 2827219147304706826L;
private String ip;
private Map<String, Object> cpuMsgMap;
private Map<String, Object> memMsgMap;
private Map<String, Object> fileSysMsgMap;
@Override
public String toString() {
return "HeatbeatMessage [\nip=" + ip
+ ", \ncpuMsgMap=" + cpuMsgMap
+ ", \nmemMsgMap=" + memMsgMap
+ ", \nfileSysMsgMap=" + fileSysMsgMap + "]";
}
public String getIp() {
return ip;
}
public void setIp(String ip) {
this.ip = ip;
}
public Map<String, Object> getCpuMsgMap() {
return cpuMsgMap;
}
public void setCpuMsgMap(Map<String, Object> cpuMsgMap) {
this.cpuMsgMap = cpuMsgMap;
}
public Map<String, Object> getMemMsgMap() {
return memMsgMap;
}
public void setMemMsgMap(Map<String, Object> memMsgMap) {
this.memMsgMap = memMsgMap;
}
public Map<String, Object> getFileSysMsgMap() {
return fileSysMsgMap;
}
public void setFileSysMsgMap(Map<String, Object> fileSysMsgMap) {
this.fileSysMsgMap = fileSysMsgMap;
}
}
在基于流的传输里比如TCP/IP，接收到的数据会先被存储到一个socket接收缓冲里。不幸的是，基于流的传输并不是一个数据包队列，而是一个字节队列。即使你发送了2个独立的数据包，操作系统也不会作为2个消息处理而仅仅是作为一连串的字节而言。因此这是不能保证你远程写入的数据就会准确地读取。所以一个接收方不管他是客户端还是服务端，都应该把接收到的数据整理成一个或者多个更有意思并且能够让程序的业务逻辑更好理解的数据。
在处理流数据粘包拆包时，可以使用下述处理方式：
使用定长数据处理，如：每个完整请求数据长度为8字节等。（FixedLengthFrameDecoder）
使用特殊分隔符的方式处理，如：每个完整请求数据末尾使用’\0’作为数据结束标记。（DelimiterBasedFrameDecoder）
使用自定义协议方式处理，如：http协议格式等。
使用POJO来替代传递的流数据，如：每个完整的请求数据都是一个RequestMessage对象，在Java语言中，使用POJO更符合语种特性，推荐使用。