netty 学习 (3)发送对象
netty 学习 (3)发送对象
yaokangjun 发表于4年前
netty 学习 (3)发送对象
  • 发表于 4年前
  • 阅读 6927
  • 收藏 6
  • 点赞 1
  • 评论 1

新睿云服务器60天免费使用,快来体验!>>>   

摘要: Netty中,通讯的双方建立连接后,会把数据按照ByteBuf的方式进行传输,例如http协议中,就是通过HttpRequestDecoder对ByteBuf数据流进行处理,转换成http的对象。基于这个思路,我自定义一种通讯协议:Server和客户端直接传输java对象。 实现的原理是通过Encoder把java对象转换成ByteBuf流进行传输,通过Decoder把ByteBuf转换成java对象进行处理。 参看:http://blog.csdn.net/u013252773/article/details/21608951

Netty中,通讯的双方建立连接后,会把数据按照ByteBuf的方式进行传输,例如http协议中,就是通过HttpRequestDecoder对ByteBuf数据流进行处理,转换成http的对象。基于这个思路,我自定义一种通讯协议:Server和客户端直接传输java对象。

实现的原理是通过Encoder把java对象转换成ByteBuf流进行传输,通过Decoder把ByteBuf转换成java对象进行处理,处理逻辑如下图所示:

使用的jar包:

使用的log4j.xml文件:

<?xml version="1.0"?>
<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">

<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/">
    <appender name="CONSOLE" class="org.apache.log4j.ConsoleAppender">
        <layout class="org.apache.log4j.PatternLayout">
            <param name="ConversionPattern" value="[%-5p] [%d] [%t] [%c] %m%n"/>
        </layout>
    </appender>
    
    <appender name="FILE" class="org.apache.log4j.DailyRollingFileAppender">
        <param name="File" value="./log/netty.log"/>
        <layout class="org.apache.log4j.PatternLayout">
            <param name="ConversionPattern" value="[%-5p] [%d] [%t] [%c] %m%n"/>
        </layout>
    </appender>
    
    <appender name="FILE_ERR" class="org.apache.log4j.DailyRollingFileAppender">
        <param name="File" value="./log/netty_err.log"/>
        <param name="Threshold" value="ERROR" />
        <layout class="org.apache.log4j.PatternLayout">
            <param name="ConversionPattern" value="[%-5p] [%d] [%t] [%c] %m%n"/>
        </layout>
    </appender>	    
    
	<logger name="io.netty" additivity="false">
		<level value="INFO,DEBUG" />
		<appender-ref ref="FILE" />
		<appender-ref ref="FILE_ERR" />
		<appender-ref ref="CONSOLE" />
	</logger>
	<logger name="com.yao" additivity="false">
		<level value="INFO,DEBUG" />
		<appender-ref ref="FILE" />
		<appender-ref ref="FILE_ERR" />
		<appender-ref ref="CONSOLE" />
	</logger>
    
    <root>
    	
      <level value="debug"/>
        <appender-ref ref="FILE"/>
        <appender-ref ref="CONSOLE"/>
        <appender-ref ref="FILE_ERR" />
    </root>

</log4j:configuration>



传输的java bean为Person:

package com.yao.nettyobject;

import java.io.Serializable;

// 必须实现Serializable接口
public class Person implements Serializable{
	private static final long	serialVersionUID	= 1L;
	private String	name;
	private String	sex;
	private int		age;

	public String toString() {
		return "name:" + name + " sex:" + sex + " age:" + age;
	}

	public String getName() {
		return name;
	}

	public void setName(String name) {
		this.name = name;
	}

	public String getSex() {
		return sex;
	}

	public void setSex(String sex) {
		this.sex = sex;
	}

	public int getAge() {
		return age;
	}

	public void setAge(int age) {
		this.age = age;
	}
}


Server端类:Server PersonDecoder BusinessHandler

1、Server:启动netty服务

package com.yao.nettyobject;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class Server {
	public void start(int port) throws Exception {
		EventLoopGroup bossGroup = new NioEventLoopGroup(); 
		EventLoopGroup workerGroup = new NioEventLoopGroup();
		try {
			ServerBootstrap b = new ServerBootstrap(); 
			b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class) 
					.childHandler(new ChannelInitializer<SocketChannel>() { 
								@Override
								public void initChannel(SocketChannel ch) throws Exception {
									//解码
									ch.pipeline().addLast(new PersonDecoder());
									//业务处理
									ch.pipeline().addLast(new BusinessHandler());
								}
							}).option(ChannelOption.SO_BACKLOG, 128) 
					.childOption(ChannelOption.SO_KEEPALIVE, true); 

			ChannelFuture f = b.bind(port).sync(); 

			f.channel().closeFuture().sync();
		} finally {
			workerGroup.shutdownGracefully();
			bossGroup.shutdownGracefully();
		}
	}

	public static void main(String[] args) throws Exception {
		Server server = new Server();
		server.start(8000);
	}
}



2、PersonDecoder:把ByteBuf流转换成Person对象,其中ByteBufToBytes是读取ButeBuf的工具类,上一篇文章中提到过,在此不在详述。ByteObjConverter是byte和obj的互相转换的工具。

package com.yao.nettyobject;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

import java.util.List;


public class PersonDecoder extends ByteToMessageDecoder {
	@Override
	protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
		ByteBufToBytes read = new ByteBufToBytes();
		Object obj = ByteObjConverter.byteToObject(read.read(in));
		out.add(obj);
	}

}



3、BusinessHandler 读取Person信息,并打印
package com.yao.nettyobject;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class BusinessHandler extends ChannelInboundHandlerAdapter {
	private Log	logger	= LogFactory.getLog(BusinessHandler.class);

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		Person person = (Person) msg;
		logger.info("BusinessHandler read msg from client :" + person);
	}

	@Override
	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
		ctx.flush();
	}
	
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		
	}
}



Client端的类:Client ClientInitHandler PersonEncoder

1、Client 建立与Server的连接

package com.yao.nettyobject;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

public class Client {
	public void connect(String host, int port) throws Exception {
		EventLoopGroup workerGroup = new NioEventLoopGroup();

		try {
			Bootstrap b = new Bootstrap(); 
			b.group(workerGroup); 
			b.channel(NioSocketChannel.class); 
			b.option(ChannelOption.SO_KEEPALIVE, true); 
			b.handler(new ChannelInitializer<SocketChannel>() {
				@Override
				public void initChannel(SocketChannel ch) throws Exception {
					//编码
					ch.pipeline().addLast(new PersonEncoder());
					//
					ch.pipeline().addLast(new ClientInitHandler());
				}
			});

			ChannelFuture f = b.connect(host, port).sync();
			f.channel().closeFuture().sync();
		} finally {
			workerGroup.shutdownGracefully();
		}

	}

	public static void main(String[] args) throws Exception {
		Client client = new Client();
		client.connect("127.0.0.1", 8000);
	}
}



2、ClientInitHandler 向Server发送Person对象

package com.yao.nettyobject;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class ClientInitHandler extends ChannelInboundHandlerAdapter {
	private static Log	logger	= LogFactory.getLog(ClientInitHandler.class);
	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		logger.info("HelloClientIntHandler.channelActive");
		Person person = new Person();
		person.setName("yaokj");
		person.setSex("man");
		person.setAge(30);
		ctx.write(person);
		ctx.flush();
	}
}



3、PersonEncoder 把Person对象转换成ByteBuf进行传送

package com.yao.nettyobject;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;

public class PersonEncoder extends MessageToByteEncoder<Person> {

	@Override
	protected void encode(ChannelHandlerContext ctx, Person msg, ByteBuf out) throws Exception {
		byte[] datas = ByteObjConverter.objectToByte(msg);
		out.writeBytes(datas);
		ctx.flush();
	}
}



工具类:ByteObjConverter

package com.yao.nettyobject;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;

public class ByteObjConverter {

	public static Object byteToObject(byte[] bytes) {
		Object obj = null;
		ByteArrayInputStream bi = new ByteArrayInputStream(bytes);
		ObjectInputStream oi = null;
		try {
			oi = new ObjectInputStream(bi);
			obj = oi.readObject();
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				bi.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
			try {
				oi.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return obj;
	}

	public static byte[] objectToByte(Object obj) {
		byte[] bytes = null;
		ByteArrayOutputStream bo = new ByteArrayOutputStream();
		ObjectOutputStream oo = null;
		try {
			oo = new ObjectOutputStream(bo);
			oo.writeObject(obj);
			bytes = bo.toByteArray();
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				bo.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
			try {
				oo.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return bytes;
	}
}



工具类:ByteBufToBytes

package com.yao.nettyobject;

import io.netty.buffer.ByteBuf;

public class ByteBufToBytes {

	public byte[] read(ByteBuf datas) {
		byte[] bytes = new byte[datas.readableBytes()];
		datas.readBytes(bytes);
		return bytes;
	}
}



通过上述代码,实现了Server端与Client端直接使用person对象进行通信的目的。基于此,可以构建更为复杂的场景:Server端同时支撑多种协议,不同的协议采用不同的Decoder进行解析,解析结果保持统一,这样业务处理类可以保持接口一致。下一节将编写这样一个案例。

本例中需要注意的事项是:

1、Person对象必须实现Serializable接口,否则不能进行序列化。

2、PersonDecoder读取ByteBuf数据的时候,并没有对多次流式数据进行处理,而是简单的一次性接收,如果数据量大的情况下,可能会出现数据不完整,这个问题会在后续的学习中解决。


  • 打赏
  • 点赞
  • 收藏
  • 分享
共有 人打赏支持
粉丝 25
博文 62
码字总数 29408
评论 (1)
java_sheng
netty 提供了ObjectEncoder ObjectDecoder 方便传输 不过基于java自带的序列化 在大并发下 效率 还是低了一点
×
yaokangjun
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: