logo

Netty4.0学习笔记系列之六：多种通讯协议支持博客分类：应用服务Netty多种通讯协议

上文介绍了如何应用Netty开发自定义通讯协议，本文在此基础上进一步深化，研究如何同时支持不同的通讯协议。

此处所谓的通讯协议，指的是把Netty通讯管道中的二进制流转换为对象、把对象转换成二进制流的过程。转换过程追根究底还是ChannelInboundHandler、ChannelOutboundHandler的实现类在进行处理。ChannelInboundHandler负责把二进制流转换为对象，ChannelOutboundHandler负责把对象转换为二进制流。

接下来要构建一个Server，同时支持Person通讯协议和String通讯协议。

Person通讯协议：二进制流与Person对象间的互相转换。

String通讯协议：二进制流与有固定格式要求的String的相互转换。String格式表示的也是一个Person对象，格式规定为：name:xx;age:xx;sex:xx;

这时候，来自客户端的请求，会依次传递给两个通讯解析接口进行解析，每个通讯接口判断是否是匹配的协议，如果是则进行解析，如果不是则传递给其它通讯接口进行解析。

实体类Person.java

package com.bijian.netty.dto;

import java.io.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.java、PersonDecoder.java、StringDecoder.java、BusinessHandler.java

1.Server.java开启Netty服务

package com.bijian.netty.server;

import io.netty.bootstrap.ServerBootstrap;

import io.netty.channel.ChannelFuture;

import io.netty.channel.ChannelInitializer;

import io.netty.channel.ChannelOption;

import io.netty.channel.EventLoopGroup;

import io.netty.channel.nio.NioEventLoopGroup;

import io.netty.channel.socket.SocketChannel;

import io.netty.channel.socket.nio.NioServerSocketChannel;

// 测试coder 和 handler 的混合使用

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 StringDecoder());

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.java把二进制流转换成Person对象

package com.bijian.netty.server;

import io.netty.buffer.ByteBuf;

import io.netty.channel.ChannelHandlerContext;

import io.netty.handler.codec.ByteToMessageDecoder;

import java.util.List;

import com.bijian.netty.util.ByteBufToBytes;

import com.bijian.netty.util.ByteObjConverter;

public class PersonDecoder extends ByteToMessageDecoder {

@Override

protected void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws Exception {

byte n = "n".getBytes()[0];

byte p = in.readByte();

in.resetReaderIndex();

if (n != p) {

// 把读取的起始位置重置

ByteBufToBytes reader = new ByteBufToBytes();

out.add(ByteObjConverter.ByteToObject(reader.read(in)));

} else {

// 执行其它的decode

ctx.fireChannelRead(in);

}

}

}

3.StringDecoder.java把满足条件的字符串转换成Person对象

package com.bijian.netty.server;

import io.netty.buffer.ByteBuf;

import io.netty.channel.ChannelHandlerContext;

import io.netty.handler.codec.ByteToMessageDecoder;

import java.util.List;

import com.bijian.netty.dto.Person;

import com.bijian.netty.util.ByteBufToBytes;

public class StringDecoder extends ByteToMessageDecoder {

@Override

protected void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws Exception {

// 判断是否是String协议

byte n = "n".getBytes()[0];

byte p = in.readByte();

// 把读取的起始位置重置

in.resetReaderIndex();

if (n == p) {

ByteBufToBytes reader = new ByteBufToBytes();

String msg = new String(reader.read(in));

Person person = buildPerson(msg);

out.add(person);

//in.release();

} else {

ctx.fireChannelRead(in);

}

}

private Person buildPerson(String msg) {

Person person = new Person();

String[] msgArray = msg.split(";|:");

person.setName(msgArray[1]);

person.setAge(Integer.parseInt(msgArray[3]));

person.setSex(msgArray[5]);

return person;

}

}

4.BusinessHandler.java展现客户端请求的内容

package com.bijian.netty.server;

import io.netty.channel.ChannelHandlerContext;

import io.netty.channel.ChannelInboundHandlerAdapter;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import com.bijian.netty.dto.Person;

public class BusinessHandler extends ChannelInboundHandlerAdapter {

private Logger logger = LoggerFactory.getLogger(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 {

ctx.close();

}

}

客户端1发送Person格式的协议：Client.java、ClientInitHandler.java、PersonEncoder.java

1.Client.java

package com.bijian.netty.client;

import io.netty.bootstrap.Bootstrap;

import io.netty.channel.ChannelFuture;

import io.netty.channel.ChannelInitializer;

import io.netty.channel.ChannelOption;

import io.netty.channel.EventLoopGroup;

import io.netty.channel.nio.NioEventLoopGroup;

import io.netty.channel.socket.SocketChannel;

import io.netty.channel.socket.nio.NioSocketChannel;

import com.bijian.netty.dto.Person;

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());

Person person = new Person();

person.setName("test");

person.setSex("man");

person.setAge(30);

ch.pipeline().addLast(new ClientInitHandler(person));

}

});

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.java向服务端发送Person对象

package com.bijian.netty.client;

import io.netty.channel.ChannelHandlerContext;

import io.netty.channel.ChannelInboundHandlerAdapter;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import com.bijian.netty.dto.Person;

public class ClientInitHandler extends ChannelInboundHandlerAdapter {

private static Logger logger = LoggerFactory.getLogger(ClientInitHandler.class);

private Person person;

public ClientInitHandler(Person person) {

this.person = person;

}

@Override

public void channelActive(ChannelHandlerContext ctx) throws Exception {

logger.info("ClientInitHandler.channelActive");

ctx.write(person);

ctx.flush();

}

}

3.PersonEncoder.java把Person对象转换成二进制进行传送

package com.bijian.netty.client;

import io.netty.buffer.ByteBuf;

import io.netty.channel.ChannelHandlerContext;

import io.netty.handler.codec.MessageToByteEncoder;

import com.bijian.netty.dto.Person;

import com.bijian.netty.util.ByteObjConverter;

public class PersonEncoder extends MessageToByteEncoder {

@Override

protected void encode(ChannelHandlerContext ctx, Person msg, ByteBuf out) throws Exception {

out.writeBytes(ByteObjConverter.ObjectToByte(msg));

}

}

客户端2发送String格式的协议：Client2.java、StringEncoder.java同样使用了客户端1中定义的ClientInitHandler 进行数据发送操作。

1.Client2.java

package com.bijian.netty.client;

import io.netty.bootstrap.Bootstrap;

import io.netty.channel.ChannelFuture;

import io.netty.channel.ChannelInitializer;

import io.netty.channel.ChannelOption;

import io.netty.channel.EventLoopGroup;

import io.netty.channel.nio.NioEventLoopGroup;

import io.netty.channel.socket.SocketChannel;

import io.netty.channel.socket.nio.NioSocketChannel;

import com.bijian.netty.dto.Person;

public class Client2 {

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 StringEncoder());

Person person = new Person();

person.setName("test2");

person.setSex("girl");

person.setAge(4);

ch.pipeline().addLast(new ClientInitHandler(person));

}

});

ChannelFuture f = b.connect(host, port).sync();

f.channel().closeFuture().sync();

} finally {

workerGroup.shutdownGracefully();

}

}

public static void main(String[] args) throws Exception {

Client2 client = new Client2();

client.connect("127.0.0.1", 8000);

}

}

2.StringEncoder.java把Person对象转换成固定格式的String的二进制流进行传送

package com.bijian.netty.client;

import io.netty.buffer.ByteBuf;

import io.netty.channel.ChannelHandlerContext;

import io.netty.handler.codec.MessageToByteEncoder;

import com.bijian.netty.dto.Person;

public class StringEncoder extends MessageToByteEncoder {

@Override

protected void encode(ChannelHandlerContext ctx, Person msg, ByteBuf out) throws Exception {

// 转成字符串：name:xx;age:xx;sex:xx;

StringBuffer sb = new StringBuffer();

sb.append("name:").append(msg.getName()).append(";");

sb.append("age:").append(msg.getAge()).append(";");

sb.append("sex:").append(msg.getSex()).append(";");

out.writeBytes(sb.toString().getBytes());

}

}

其它：工具类ByteBufToBytes（读取ByteBuf数据的工具类）、ByteObjConverter（Object与byte互转的工具类）在以前的文章中已经存在，在此省略。

注意事项：

1.该段代码能运行出结果，但是运行的时候会报 io.netty.util.IllegalReferenceCountException: refCnt: 0, decrement: 1 异常，已经解决。日志中的提示信息为：

An exceptionCaught() event was fired, and it reached at the tail of the pipeline. It usually means the last handler in the pipeline did not handle the exception

说明缺少exceptionCaught方法，在server端最后一个Handler中增加这个方法即可。

2.PersonDecoder和StringDecoder中有一个if判断，是为了判断消息究竟是什么协议。如果是String协议的话，格式是【name:xx;age:xx;sex:xx;】，第一个字母是英文字母n，所以判断协议类型时候是读取二进制流的第一个字符进行判断，当然这种判断方式非常幼稚，以后有机会可以进行改善。

文章来源：http://blog.csdn.net/u013252773/article/details/22108385

X
Copyright © 2088 u19男篮世界杯_世界杯在哪看 - xmzohe.com All Rights Reserved.
友情链接