博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
netty 学习 (3)发送对象
阅读量:6353 次
发布时间:2019-06-22

本文共 8216 字,大约阅读时间需要 27 分钟。

  hot3.png

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

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

使用的jar包:

使用的log4j.xml文件:

传输的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
() { @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 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
() { @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
{ @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数据的时候,并没有对多次流式数据进行处理,而是简单的一次性接收,如果数据量大的情况下,可能会出现数据不完整,这个问题会在后续的学习中解决。

转载于:https://my.oschina.net/jamaly/blog/272451

你可能感兴趣的文章
Percona Server安装
查看>>
函数为左边表达式
查看>>
2015.06.04 工作任务与心得
查看>>
icinga2使用587端口发邮件
查看>>
hpasmcli查看HP服务器内存状态
查看>>
【14】Python100例基础练习(1)
查看>>
boost bind使用指南
查看>>
使用ntpdate更新系统时间
查看>>
Android M 特性 Doze and App Standby模式详解
查看>>
IE FF(火狐) line-height兼容详解
查看>>
谷歌Pixel 3吸引三星用户, 但未动摇iPhone地位
查看>>
VUE中使用vuex,cookie,全局变量(少代码示例)
查看>>
grep -w 的解析_学习笔记
查看>>
TX Text Control文字处理教程(3)打印操作
查看>>
CENTOS 7 如何修改IP地址为静态!
查看>>
MyCat分片算法学习(纯转)
查看>>
IO Foundation 3 -文件解析器 FileParser
查看>>
linux学习经验之谈
查看>>
mysqld_multi实现多主一从复制
查看>>
中介模式
查看>>