上一篇 netty(1)
一、TCP/IP 流式传输
在上文演示了2进制流式传输引起的TCP拆包问题，这里继续演示文本型的传输问题，文本型的可以有以下几种策略
1.1 以特殊字符表示结尾
HTTP协议中以\r\n\r\n表示请求首部结束，这里也以\r\n\r\n表示特殊字符，非常容易理解，没有碰到\r\n\r\n就继续写入缓冲，碰到了表明是一个完整的逻辑数据，可以处理了。
Server端代码
public class Server {
public static final String SERVER_DELIMITER = "\r\n\r\n";
public static void main(String[] args) throws Exception {
//1 创建2个线程，一个是负责接收客户端的连接。一个是负责进行数据传输的
EventLoopGroup pGroup = new NioEventLoopGroup();
EventLoopGroup cGroup = new NioEventLoopGroup();
//2 创建服务器辅助类
ServerBootstrap b = new ServerBootstrap();
b.group(pGroup, cGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 1024)
.option(ChannelOption.SO_SNDBUF, 32 * 1024)
.option(ChannelOption.SO_RCVBUF, 32 * 1024)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel sc) throws Exception {
//设置特殊分隔符
ByteBuf buf = Unpooled.copiedBuffer(SERVER_DELIMITER.getBytes());
sc.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, buf));
//设置字符串形式的解码
sc.pipeline().addLast(new StringDecoder());
sc.pipeline().addLast(new ServerHandler());
}
});
//4 绑定连接
ChannelFuture cf = b.bind(8765).sync();
//等待服务器监听端口关闭
cf.channel().closeFuture().sync();
pGroup.shutdownGracefully();
cGroup.shutdownGracefully();
}
}
public class ServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println(" server channel active... ");
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
String request = (String)msg;
System.out.println("Server :" + msg);
String response = "服务器响应：" + msg + Server.SERVER_DELIMITER;
ctx.writeAndFlush(Unpooled.copiedBuffer(response.getBytes()));
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable t) throws Exception {
ctx.close();
}
}
客户端代码：
public class Client {
public static final String CLIENT_DELIMITER = "\r\n\r\n";
public static void main(String[] args) throws Exception {
EventLoopGroup group = new NioEventLoopGroup();
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel sc) throws Exception {
//
ByteBuf buf = Unpooled.copiedBuffer(CLIENT_DELIMITER.getBytes());
sc.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, buf));
sc.pipeline().addLast(new StringDecoder());
sc.pipeline().addLast(new ClientHandler());
}
});
ChannelFuture cf = b.connect("127.0.0.1", 8765).sync();
StringBuffer sb = new StringBuffer("GET " + "/index.jsp" + " HTTP/1.1\r\n");
sb.append("Host: www.javathinker.org\r\n");
sb.append("Accept: */*\r\n");
sb.append("Accept-Language: zh-cn\r\n");
sb.append("Accept-Encoding: gzip, deflate\r\n");
sb.append("User-Agent: Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.0)\r\n");
sb.append("Connection: Keep-Alive\r\n\r\n");
cf.channel().writeAndFlush(Unpooled.wrappedBuffer(sb.toString().getBytes()));
// Thread.sleep(1000);
// cf.addListener(ChannelFutureListener.CLOSE);
//等待客户端端口关闭
cf.channel().closeFuture().sync();
group.shutdownGracefully();
}
}
public class ClientHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("client channel active... ");
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
try {
String response = (String)msg;
System.out.println("Client: " + response);
} finally {
ReferenceCountUtil.release(msg);
}
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
}
1.2 以FixedLength的方式
跟上述代码类似，只需要修改下Decoder即可
b.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel sc) throws Exception {
sc.pipeline().addLast(new FixedLengthFrameDecoder(5));
sc.pipeline().addLast(new StringDecoder());
sc.pipeline().addLast(new ClientHandler());
}
});
二、使用Java对象传输
官网Object demo：http://netty.io/4.1/xref/io/netty/example/objectecho/package-summary.html
使用Java对象传输，只需要配置对应的encoder: Object->byte[] 和对应的decoder: byte[]->Object即可
2.1 使用java searilizable
先写个工具类，提供GZIP压缩功能，以及发送请求和处理请求的通用方法
package netty; import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import netty.marshalling.Req;
import netty.marshalling.Resp; import java.io.*;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream; /**
* Created by carl.yu on 2016/11/7.
*/
public class Utils {
public static byte[] gzip(byte[] data) throws Exception{
ByteArrayOutputStream bos = new ByteArrayOutputStream();
GZIPOutputStream gzip = new GZIPOutputStream(bos);
gzip.write(data);
gzip.finish();
gzip.close();
byte[] ret = bos.toByteArray();
bos.close();
return ret;
} public static byte[] ungzip(byte[] data) throws Exception{
ByteArrayInputStream bis = new ByteArrayInputStream(data);
GZIPInputStream gzip = new GZIPInputStream(bis);
byte[] buf = new byte[1024];
int num = -1;
ByteArrayOutputStream bos = new ByteArrayOutputStream();
while((num = gzip.read(buf, 0 , buf.length)) != -1 ){
bos.write(buf, 0, num);
}
gzip.close();
bis.close();
byte[] ret = bos.toByteArray();
bos.flush();
bos.close();
return ret;
} public static void send(ChannelFuture cf) throws Exception{
for (int i = 1; i <= 2; i++) {
Req req = new Req();
req.setId("" + i);
req.setName("pro" + i);
req.setRequestMessage("数据信息" + i);
String path = "G:\\projects-helloworld\\lucene\\src\\main\\resources\\in\\" + i + ".jpg";
String fileName = i + ".jpg";
req.setFileName(fileName);
File file = new File(path);
FileInputStream in = new FileInputStream(file);
byte[] data = new byte[in.available()]; in.read(data);
in.close();
req.setAttachment(Utils.gzip(data));
cf.channel().writeAndFlush(req);
}
} public static void recv(ChannelHandlerContext ctx, Object msg) throws Exception{
Req req = (Req) msg;
System.out.println("req:"+req);
System.out.println("Server : " + req.getId() + ", " + req.getName() + ", " + req.getRequestMessage());
byte[] attachment = Utils.ungzip(req.getAttachment());
String fileName = req.getFileName();
String path = "G:\\projects-helloworld\\lucene\\src\\main\\resources\\out\\" + fileName;
FileOutputStream fos = new FileOutputStream(path);
fos.write(attachment);
fos.close(); Resp resp = new Resp();
resp.setId(req.getId());
resp.setName("resp" + req.getId());
resp.setResponseMessage("响应内容" + req.getId());
ctx.writeAndFlush(resp);//.addListener(ChannelFutureListener.CLOSE);
} public static void main(String[] args) throws Exception{ //读取文件
String readPath = System.getProperty("user.dir") + File.separatorChar + "sources" + File.separatorChar + "006.jpg";
File file = new File(readPath);
FileInputStream in = new FileInputStream(file);
byte[] data = new byte[in.available()];
in.read(data);
in.close(); System.out.println("文件原始大小:" + data.length);
//测试压缩 byte[] ret1 = gzip(data);
System.out.println("压缩之后大小:" + ret1.length); byte[] ret2 = ungzip(ret1);
System.out.println("还原之后大小:" + ret2.length); //写出文件
String writePath = System.getProperty("user.dir") + File.separatorChar + "receive" + File.separatorChar + "006.jpg";
FileOutputStream fos = new FileOutputStream(writePath);
fos.write(ret2);
fos.close(); }
}
Req:
import java.io.Serializable;
public class Req implements Serializable {
private static final long SerialVersionUID = 1L;
private String id;
private String name;
private String requestMessage;
private byte[] attachment;
private String fileName;
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getRequestMessage() {
return requestMessage;
}
public void setRequestMessage(String requestMessage) {
this.requestMessage = requestMessage;
}
public byte[] getAttachment() {
return attachment;
}
public void setAttachment(byte[] attachment) {
this.attachment = attachment;
}
public String getFileName() {
return fileName;
}
public void setFileName(String fileName) {
this.fileName = fileName;
}
@Override
public String toString() {
return "Req{" +
"id='" + id + '\'' +
", name='" + name + '\'' +
", requestMessage='" + requestMessage + '\'' +
", fileName='" + fileName + '\'' +
'}';
}
}
Resp:
import java.io.Serializable;
public class Resp implements Serializable{
private static final long serialVersionUID = 1L;
private String id = "1";
private String name ="aaa";
private String responseMessage ="this is demo";
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getResponseMessage() {
return responseMessage;
}
public void setResponseMessage(String responseMessage) {
this.responseMessage = responseMessage;
}
@Override
public String toString() {
return "Resp{" +
"id='" + id + '\'' +
", name='" + name + '\'' +
", responseMessage='" + responseMessage + '\'' +
'}';
}
}
Server端代码：
public class ObjectEchoServer {
static final boolean SSL = System.getProperty("ssl") != null;
static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));
public static void main(String[] args) throws Exception {
// Configure SSL.
final SslContext sslCtx;
if (SSL) {
SelfSignedCertificate ssc = new SelfSignedCertificate();
sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();
} else {
sslCtx = null;
}
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
if (sslCtx != null) {
p.addLast(sslCtx.newHandler(ch.alloc()));
}
p.addLast(
new ObjectEncoder(),
new ObjectDecoder(ClassResolvers.cacheDisabled(null)),
new ObjectEchoServerHandler());
}
});
// Bind and start to accept incoming connections.
b.bind(PORT).sync().channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
public class ObjectEchoServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// Echo back the received object to the client.
Utils.recv(ctx, msg);
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) {
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}
client端代码：
public final class ObjectEchoClient {
static final boolean SSL = System.getProperty("ssl") != null;
static final String HOST = System.getProperty("host", "127.0.0.1");
static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));
static final int SIZE = Integer.parseInt(System.getProperty("size", "256"));
public static void main(String[] args) throws Exception {
// Configure SSL.
final SslContext sslCtx;
if (SSL) {
sslCtx = SslContextBuilder.forClient()
.trustManager(InsecureTrustManagerFactory.INSTANCE).build();
} else {
sslCtx = null;
}
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
if (sslCtx != null) {
p.addLast(sslCtx.newHandler(ch.alloc(), HOST, PORT));
}
p.addLast(
new ObjectEncoder(),
new ObjectDecoder(ClassResolvers.cacheDisabled(null)),
new ObjectEchoClientHandler());
}
});
// Start the connection attempt.
ChannelFuture cf = b.connect(HOST, PORT).sync();
Utils.send(cf);
cf.channel().closeFuture().sync();
} finally {
group.shutdownGracefully();
}
}
}
public class ObjectEchoClientHandler extends ChannelInboundHandlerAdapter {
private final List<Integer> firstMessage;
/**
* Creates a client-side handler.
*/
public ObjectEchoClientHandler() {
firstMessage = new ArrayList<Integer>(ObjectEchoClient.SIZE);
for (int i = 0; i < ObjectEchoClient.SIZE; i++) {
firstMessage.add(Integer.valueOf(i));
}
}
@Override
public void channelActive(ChannelHandlerContext ctx) {
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
try {
Resp resp = (Resp)msg;
System.out.println("Client : " + resp.getId() + ", " + resp.getName() + ", " + resp.getResponseMessage());
} finally {
ReferenceCountUtil.release(msg);
}
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) {
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}
2.2 使用jboss marshalling序列化
<dependency>
<groupId>org.jboss.marshalling</groupId>
<artifactId>jboss-marshalling-serial</artifactId>
<version>2.0.0.Beta2</version>
</dependency>
jboss marshalling自动支持netty的codec
public final class MarshallingCodeCFactory {
/**
* 创建Jboss Marshalling解码器MarshallingDecoder
* @return MarshallingDecoder
*/
public static MarshallingDecoder buildMarshallingDecoder() {
//首先通过Marshalling工具类的精通方法获取Marshalling实例对象 参数serial标识创建的是java序列化工厂对象。
final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
//创建了MarshallingConfiguration对象，配置了版本号为5
final MarshallingConfiguration configuration = new MarshallingConfiguration();
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;
}
}
直接在server中配置codec即可，其他都类似
b.group(pGroup, cGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 1024)
//设置日志
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<SocketChannel>() {
protected void initChannel(SocketChannel sc) throws Exception {
sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
sc.pipeline().addLast(new ServerHandler());
}
});
b.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel sc) throws Exception {
sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
sc.pipeline().addLast(new ClientHandler());
}
});
2.3 其他序列化框架
其他序列化框架都类似，只要有对应的encoder和decoder，比如Kryo和Google protocol Buf
三、其他
3.1 read_time_out和write_time_out
在用netty进行socket通信时，通常也会遇到read time out和write time out的设置问题，netty也是通过handler来实现的，netty默认提供了2个类
// The connection is closed when there is no inbound traffic
// for 30 seconds. public class MyChannelInitializer extends ChannelInitializer<Channel> {
public void initChannel(Channel channel) {
channel.pipeline().addLast("readTimeoutHandler", new ReadTimeoutHandler(30);
channel.pipeline().addLast("myHandler", new MyHandler());
}
} // Handler should handle the ReadTimeoutException.
public class MyHandler extends ChannelDuplexHandler {
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
throws Exception {
if (cause instanceof ReadTimeoutException) {
// do something
} else {
super.exceptionCaught(ctx, cause);
}
}
} ServerBootstrap bootstrap = ...;
...
bootstrap.childHandler(new MyChannelInitializer());
...
当读超时，可以捕获该异常，也可以丢弃此连接，防止占用服务器资源。
同样的
// The connection is closed when a write operation cannot finish in 30 seconds.
public class MyChannelInitializer extends ChannelInitializer<Channel> {
public void initChannel(Channel channel) {
channel.pipeline().addLast("writeTimeoutHandler", new WriteTimeoutHandler(30);
channel.pipeline().addLast("myHandler", new MyHandler());
}
}
// Handler should handle the WriteTimeoutException.
public class MyHandler extends ChannelDuplexHandler {
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
throws Exception {
if (cause instanceof WriteTimeoutException) {
// do something
} else {
super.exceptionCaught(ctx, cause);
}
}
}
ServerBootstrap bootstrap = ...;
...
bootstrap.childHandler(new MyChannelInitializer());
...