一个简单 netty client pojo通讯实现

------------------------------------

1

------------------------------------

package admin.netty;

import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;

public class TimeClientHandler extends SimpleChannelHandler {?
?
??? private String returnFlag = null;
???
??? @Override
??? public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
??? ??? CtrlProtocol m = (CtrlProtocol) e.getMessage();
??? ??? this.returnFlag = m.getFlag();
??? ??? //System.out.println(returnFlag);
??? ??? e.getChannel().close();
??? }
?
??? @Override?
??? public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {??
??????? e.getCause().printStackTrace();??
??????? e.getChannel().close();??
??? }??????
?????
??? public ChannelFuture process(Channel channel,CtrlProtocol requestParameter) {
??????? return channel.write(requestParameter);
??? }
???
??? public String getReturnFlag(){
??? ??? return this.returnFlag;
??? }
}

-------------------------------?

2

-------------------------------

package admin.netty;

import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;

public class TimeClientPipelineFactory implements ChannelPipelineFactory {

??? public ChannelPipeline getPipeline() {
??? ??? ChannelPipeline pipeline = Channels.pipeline();
??? ??? pipeline.addLast("decoder", new CtrlProtocolDecoder());
??? ??? pipeline.addLast("encoder", new CtrlProtocolEncoder());??? ???
??? ??? pipeline.addLast("handler", new TimeClientHandler());
??? ??? return pipeline;
??? }
}

-------------------------------?

3

-------------------------------

package admin.netty;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.handler.codec.frame.FrameDecoder;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class CtrlProtocolDecoder extends FrameDecoder {

??? private static final Log??? ilog??? = LogFactory.getLog(CtrlProtocolDecoder.class);

??? /*
??? ?* (non-Javadoc)
??? ?* @see org.jboss.netty.handler.codec.frame.FrameDecoder#decode(org.jboss.netty.channel.ChannelHandlerContext,
??? ?* org.jboss.netty.channel.Channel, org.jboss.netty.buffer.ChannelBuffer)
??? ?*/
??? @Override
??? protected CtrlProtocol decode(ChannelHandlerContext arg0, Channel arg1, ChannelBuffer arg2) throws Exception {

??? ??? ChannelBuffer buf = arg2;

??? ??? if (buf.readableBytes() < 8) {
??? ??? ??? ilog.error("收到的协议数据不完整，无法解析协议头.");??? ??? ???
??? ??? ??? return null;
??? ??? }

??? ??? short head_flag = buf.readShort();
??? ??? short head_code = buf.readShort();
??? ??? int head_length = buf.readInt();

??? ??? CtrlProtocol protocol = new CtrlProtocol(head_flag, head_code, head_length);
??? ???
??? ??? ilog.info("收到的协议:" + head_flag + "??? " + head_code + "??? " + head_length + "\n");

??? ??? return protocol;
??? }
}

-------------------------------?

4

-------------------------------

package admin.netty;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.handler.codec.oneone.OneToOneEncoder;

public class CtrlProtocolEncoder extends OneToOneEncoder {

??? /*
??? ?* (non-Javadoc)
??? ?* @see org.jboss.netty.handler.codec.oneone.OneToOneEncoder#encode(org.jboss.netty.channel.ChannelHandlerContext,
??? ?* org.jboss.netty.channel.Channel, java.lang.Object)
??? ?*/
??? @Override
??? protected Object encode(ChannelHandlerContext arg0, Channel arg1, Object arg2) throws Exception {
??? ??? CtrlProtocol protocol = (CtrlProtocol) arg2;

??? ??? ChannelBuffer buf = ChannelBuffers.dynamicBuffer();

??? ??? buf.writeShort((short) protocol.getFlag());
??? ??? buf.writeShort((short) protocol.getCode());
??? ??? buf.writeInt(protocol.getLength());
/*略去*/
??? ??? return buf;
??? }
}

-------------------------------?

5

-------------------------------

package admin.netty;

public class CtrlProtocol {

??? private short??? ??? ??? flag??? ??? = 0;
??? private short??? ??? ??? code??? ??? = 0;
??? private int??? ??? ??? length??? ??? = 0;

??? public CtrlProtocol(short f, short c, int l) {
??? ??? flag = f;
??? ??? code = c;
??? ??? length = l;
??? }
???
??? public static CtrlProtocol getCtrlInstance() {
??? ??? return new CtrlProtocol((short) 1, (short) 1, (short) 1);
??? }

??? public short getFlag() {

??? ??? return flag;
??? }

??? public void setFlag(short flag) {

??? ??? this.flag = flag;
??? }

??? public short getCode() {

??? ??? return code;
??? }

??? public void setCode(short code) {

??? ??? this.code = code;
??? }

??? public int getLength() {

??? ??? return length;
??? }

??? public void setLength(int length) {

??? ??? this.length = length;
??? }
? /* 略去*/
}

-------------------------------?

6 客户端执行异步发送接收（服务端返回数据与客户端发送数据结构一致）

-------------------------------

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;

....

??? ??? ??? CtrlProtocol cp = CtrlProtocol.getCtrlInstance();
....

??? ??? ??? String flag = this.toDo(cp);

///////////////////////////////////////////////

??? //Socket异步发送接收通讯协议
??? //参数：CtrlProtocol 协议

private String toDo(CtrlProtocol cp) throws Exception{
??? ??? String flag = "0";//结果
??? ??? String host ="127.0.0.1";

??????? int port = 7788;

??? ??? ChannelFactory factory = new NioClientSocketChannelFactory(
??? ??? ??? ??? Executors.newCachedThreadPool(),
??? ??? ??? ??? Executors.newCachedThreadPool());

??? ??? ClientBootstrap bootstrap = new ClientBootstrap(factory);

??? ??? bootstrap.setPipelineFactory(new TimeClientPipelineFactory());

??? ??? bootstrap.setOption("tcpNoDelay", true);
??? ??? bootstrap.setOption("keepAlive", true);
??? ???
??? ??? ChannelFuture future = bootstrap.connect(new InetSocketAddress(
??? ??? ??? ??? host, port));
??? ??? Channel channel = future.awaitUninterruptibly().getChannel();
??? ??? TimeClientHandler handler = channel.getPipeline().get(TimeClientHandler.class);
??? ??? ??? ???
??? ??? handler.process(channel, cp);??? ???
??? ???
??? ??? future.awaitUninterruptibly();
??? ??? if (!future.isSuccess()) {
??? ??? ??? future.getCause().printStackTrace();
??? ??? }??? ???
??? ???
??? ??? future.getChannel().getCloseFuture().awaitUninterruptibly();
??? ??? factory.releaseExternalResources();
??? ??? flag = handler.getReturnFlag();
??? ??? ??? ???
??? ??? return flag;
??? }

?

?

?