Netty筆記之八：自定義通信協(xié)議

Netty中雙方建立通信之后，對象數(shù)據(jù)會按照ByteBuf字節(jié)碼的方式進(jìn)行傳輸。

自定義一種通信協(xié)議，協(xié)議將傳輸數(shù)據(jù)定義了消息頭和消息正文。

管道中傳遞LuckMessage對象，LuckMessage中定義了消息頭LuckHeader和消息正文content。消息頭header包括version，contentLength，sessionId。

消息定義：

// 消息的頭部
public class LuckHeader {

    // 協(xié)議版本
    private int version;
    // 消息內(nèi)容長度
    private int contentLength;
    // 服務(wù)名稱
    private String sessionId;

    public LuckHeader(int version, int contentLength, String sessionId) {
        this.version = version;
        this.contentLength = contentLength;
        this.sessionId = sessionId;
    }

    public int getVersion() {
        return version;
    }

    public void setVersion(int version) {
        this.version = version;
    }

    public int getContentLength() {
        return contentLength;
    }

    public void setContentLength(int contentLength) {
        this.contentLength = contentLength;
    }

    public String getSessionId() {
        return sessionId;
    }

    public void setSessionId(String sessionId) {
        this.sessionId = sessionId;
    }
}


// 消息的主體
public class LuckMessage {

    private LuckHeader luckHeader;
    private String content;

    public LuckMessage(LuckHeader luckHeader, String content) {
        this.luckHeader = luckHeader;
        this.content = content;
    }

    public LuckHeader getLuckHeader() {
        return luckHeader;
    }

    public void setLuckHeader(LuckHeader luckHeader) {
        this.luckHeader = luckHeader;
    }

    public String getContent() {
        return content;
    }

    public void setContent(String content) {
        this.content = content;
    }

    @Override
    public String toString() {
        return String.format("[version=%d,contentLength=%d,sessionId=%s,content=%s]",
                luckHeader.getVersion(),
                luckHeader.getContentLength(),
                luckHeader.getSessionId(),
                content);
    }
}

服務(wù)端代碼：

public class LuckServer {

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

        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {

            ServerBootstrap serverBootstrap = new ServerBootstrap();
            // 指定socket的一些屬性
            serverBootstrap.option(ChannelOption.SO_BACKLOG, 1024);
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)  // 指定是一個NIO連接通道
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new LuckServerInitializer());

            // 綁定對應(yīng)的端口號,并啟動開始監(jiān)聽端口上的連接
            Channel ch = serverBootstrap.bind(8899).sync().channel();

            System.out.printf("luck協(xié)議啟動地址：127.0.0.1:%d/\n", 8899);

            // 等待關(guān)閉,同步端口
            ch.closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

服務(wù)端初始化連接：

package com.zhihao.miao.test.day08;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;

public class LuckServerInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel channel) throws Exception {

        ChannelPipeline pipeline = channel.pipeline();

        pipeline.addLast(new LuckEncoder());
        pipeline.addLast(new LuckDecoder());
        // 添加邏輯控制層
        pipeline.addLast(new LuckServerHandler());

    }
}

編碼Handler：

package com.zhihao.miao.test.day08;

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

public class LuckEncoder extends MessageToByteEncoder<LuckMessage> {

    @Override
    protected void encode(ChannelHandlerContext ctx, LuckMessage message, ByteBuf out) throws Exception {

        // 將Message轉(zhuǎn)換成二進(jìn)制數(shù)據(jù)
        LuckHeader header = message.getLuckHeader();

        // 這里寫入的順序就是協(xié)議的順序.

        // 寫入Header信息
        out.writeInt(header.getVersion());
        out.writeInt(message.getContent().length());
        out.writeBytes(header.getSessionId().getBytes());

        // 寫入消息主體信息
        out.writeBytes(message.getContent().getBytes());
    }
}

解碼器Handler：

package com.zhihao.miao.test.day08;

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

import java.util.List;

public class LuckDecoder extends ByteToMessageDecoder {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {

        // 獲取協(xié)議的版本
        int version = in.readInt();
        // 獲取消息長度
        int contentLength = in.readInt();
        // 獲取SessionId
        byte[] sessionByte = new byte[36];
        in.readBytes(sessionByte);
        String sessionId = new String(sessionByte);

        // 組裝協(xié)議頭
        LuckHeader header = new LuckHeader(version, contentLength, sessionId);

        // 讀取消息內(nèi)容，這邊demo中不對
        byte[] contentbys = new byte[in.readableBytes()];
        in.readBytes(contentbys);

        String content = new String(contentbys);

        LuckMessage message = new LuckMessage(header, content);

        out.add(message);
    }
}

自定義服務(wù)端handler處理器：

package com.zhihao.miao.test.day08;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

public class LuckServerHandler extends SimpleChannelInboundHandler<LuckMessage> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, LuckMessage msg) throws Exception {
        // 簡單地打印出server接收到的消息
        System.out.println(msg.toString());
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("service exception:"+cause.getMessage());
    }
}

客戶端：

package com.zhihao.miao.test.day08;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.util.UUID;

public class LuckClient {

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

        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group).channel(NioSocketChannel.class)
                    .handler(new LuckServerInitializer());

            // Start the connection attempt.
            Channel ch = b.connect("127.0.0.1", 8899).sync().channel();

            int version = 1;
            String sessionId = UUID.randomUUID().toString();
            String content = "I'm the luck protocol!";

            LuckHeader header = new LuckHeader(version, content.length(), sessionId);
            LuckMessage message = new LuckMessage(header, content);
            ch.writeAndFlush(message);

            ch.close();

        } finally {
            group.shutdownGracefully();
        }
    }
}

客戶端初始化連接：

package com.zhihao.miao.test.day08;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;

public class LuckClientInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel channel) throws Exception {

        ChannelPipeline pipeline = channel.pipeline();

        // 添加編解碼器, 由于ByteToMessageDecoder的子類無法使用@Sharable注解,
        // 這里必須給每個Handler都添加一個獨立的Decoder.
        pipeline.addLast(new LuckEncoder());
        pipeline.addLast(new LuckDecoder());

        pipeline.addLast(new LuckClientHandler());

    }
}

客戶端自定義handler：

package com.zhihao.miao.test.day08;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

public class LuckClientHandler extends SimpleChannelInboundHandler<LuckMessage> {

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, LuckMessage message) throws Exception {
        System.out.println(message);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("client exception:"+cause.getMessage());
    }
}

啟動服務(wù)器和客戶端，服務(wù)器端控制臺打?。?/p> 

七月 04, 2017 5:22:27 下午 io.netty.handler.logging.LoggingHandler channelRead
信息: [id: 0x9df966cc, L:/0:0:0:0:0:0:0:0:8899] READ: [id: 0x99c6480a, L:/127.0.0.1:8899 - R:/127.0.0.1:55722]
七月 04, 2017 5:22:27 下午 io.netty.handler.logging.LoggingHandler channelReadComplete
信息: [id: 0x9df966cc, L:/0:0:0:0:0:0:0:0:8899] READ COMPLETE
[version=1,contentLength=22,sessionId=c9345f67-99b6-46d2-97ff-eef853c9d569,content=I'm the luck protocol!]

完整demo

參考資料

利用Netty構(gòu)建自定義協(xié)議的通信

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容