一个简单 netty client pojo通讯实现

guoc

浏览: 29998 次
性别:
来自: 北京

最近访客更多访客>>

zhouyong9891

jones868

smart6x

sx54605628

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

JBoss Socket Apache 数据结构 F#

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

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;
    }
}

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

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;
    }
}

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

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;
    }
}

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

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;
    }
}

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

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;
    }