2. 程式人生 > >Netty學習之路（八）-Google Protobuf編碼

Netty學習之路（八）-Google Protobuf編碼

阿新 • • 發佈：2018-12-05

Protobuf是一個靈活，高效，結構化的資料序列化框架，相比於XML等傳統的序列化工具，它更小，更快，更簡單。Protobuf支援資料結構化一次可以到處使用，甚至可以跨語言使用，通過程式碼生成工具可以自動生成不同語言版本的原始碼，甚至可以在使用不同版本的資料結構程序間進行資料傳遞，實現資料結構的前向相容。

Protobuf入門

首先下載Protobuf的最新Windows版本：地址，下載底部的protoc-3.6.1-win32.zip。
解壓後得到protoc.exe工具，此工具根據.proto檔案生成程式碼。詳細protobuf java使用請看這裡，根據教程編寫如下.proto檔案：

syntax = "proto3";

message book_data {
    int64 id = 1;
    string name =2;
    string dataTime = 3;
    string type = 4;
}

再通過protoc -I=src/com/ph/proto/resource --java_out=src/com/ph/proto/java src/com/ph/proto/resource/book_data.proto生成java程式碼。匯入jar包

在Netty中使用Protobuf

首先需要用protoc.exe生成對應的BookData java程式碼。

服務端

package com.ph.Netty;

import com.ph.proto.java.BookData;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;

/**
 * Create by PH on 2018/11/9
 */
public class NettyServer {

    public static void main(String[] args) throws Exception {
        int port = 8080;
        if(args !=null && args.length>0) {
            try {
                port = Integer.valueOf(args[0]);
            }catch (NumberFormatException e) {
                //採用預設值
            }
        }
        new NettyServer().bind(port);
    }

    public void bind(int port) throws Exception{
        //NioEventLoopGroup是一個執行緒組，包含了一組NIO執行緒，專門用於網路事件的處理，實際上他們就是Reactor執行緒組
        //bossGroup僅接收客戶端連線，不做複雜的邏輯處理，為了儘可能減少資源的佔用，取值越小越好
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        //用於進行SocketChannel的網路讀寫
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            //是Netty用於啟動NIO服務端的輔助啟動類，目的是降低服務端的開發複雜度
            ServerBootstrap b = new ServerBootstrap();
            //配置NIO服務端
            b.group(bossGroup, workerGroup)
                    //指定使用NioServerSocketChannel產生一個Channel用來接收連線，他的功能對應於JDK
                    // NIO類庫中的ServerSocketChannel類。
                    .channel(NioServerSocketChannel.class)
                    //BACKLOG用於構造服務端套接字ServerSocket物件，標識當伺服器請求處理執行緒全滿時，
                    // 用於臨時存放已完成三次握手的請求的佇列的最大長度。如果未設定或所設定的值小於1，
                    // Java將使用預設值50。
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    //繫結I/O事件處理類，作用類似於Reactor模式中的Handler類，主要用於處理網路I/O事件
                    .childHandler(new ChannelInitializer<SocketChannel>() {

                        protected void initChannel(SocketChannel arg0) throws Exception {
                            //處理半包訊息
                            arg0.pipeline().addLast(new ProtobufVarint32FrameDecoder());
                            //解碼器，引數是com.google.protobuf.MessageLite，告訴ProtobufDecoder需要解碼的目標型別
                            arg0.pipeline().addLast(new ProtobufDecoder(BookData.book_data.getDefaultInstance()));
                            arg0.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
                            arg0.pipeline().addLast(new ProtobufEncoder());
                            arg0.pipeline().addLast(new ServerHandler());
                        }
                    });
            //繫結埠，同步等待繫結操作完成，完成後返回一個ChannelFuture,用於非同步操作的通知回撥
            ChannelFuture f = b.bind(port).sync();
            //等待服務端監聽埠關閉之後才退出main函式
            f.channel().closeFuture().sync();
        } finally {
            //退出，釋放執行緒池資源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

}

/**
 * ChannelInboundHandlerAdapter實現自ChannelInboundHandler
 * ChannelInboundHandler提供了不同的事件處理方法你可以重寫
 */
class ServerHandler extends ChannelInboundHandlerAdapter {

    /**
     * 接受客戶端傳送的訊息
     * @param ctx
     * @param msg
     * @throws Exception
     */
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        BookData.book_data req = (BookData.book_data)msg;
        System.out.println("Server receive : " + req.toString());
        BookData.book_data.Builder builder = BookData.book_data.newBuilder();
        builder.setId(req.getId());
        builder.setName("Netty");
        builder.setDataTime("2018-11-9 14:25:44");
        builder.setType("buy succeed");
        ctx.writeAndFlush(builder.build());
    }

    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        //將訊息傳送佇列中的訊息寫入到SocketChannel中傳送給對方
        ctx.flush();
    }

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        //當發生異常時釋放資源
        ctx.close();
    }
}

客戶端

package com.ph.Netty;

import com.ph.proto.java.BookData;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;

/**
 * Create by PH on 2018/11/9
 */
public class NettyClient {
    public static void main(String[] args) throws Exception {
        int port = 8080;
        if (args != null && args.length > 0) {
            try {
                port = Integer.valueOf(args[0]);
            } catch (NumberFormatException e) {
                //採用預設值
            }
        }
        new NettyClient().connect(port, "127.0.0.1", 100);
    }

    public void connect(int port, String host, int sendNumber) throws  Exception{
        //配置客戶端NIO執行緒組
        EventLoopGroup group = new NioEventLoopGroup();
        try{
            Bootstrap b = new Bootstrap();
            b.group(group).channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        public void initChannel(SocketChannel ch) throws Exception{
                            //處理半包訊息
                            ch.pipeline().addLast(new ProtobufVarint32FrameDecoder());
                            //解碼器，引數是com.google.protobuf.MessageLite，告訴ProtobufDecoder需要解碼的目標型別
                            ch.pipeline().addLast(new ProtobufDecoder(BookData.book_data.getDefaultInstance()));
                            ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
                            ch.pipeline().addLast(new ProtobufEncoder());
                            ch.pipeline().addLast(new ClientHandler(sendNumber));
                        }
                    });
            //發起非同步連線操作
            ChannelFuture f = b.connect(host, port).sync();
            //等待客戶端鏈路關閉
            f.channel().closeFuture().sync();
        }finally {
            group.shutdownGracefully();
        }
    }
}

class ClientHandler extends ChannelInboundHandlerAdapter {

    private int sendNumber;

    public ClientHandler(int sendNumber) {
        this.sendNumber = sendNumber;
    }

    /**
     * 當客戶端和服務端TCP鏈路建立成功之後，Netty的NIO執行緒會呼叫此方法
     * @param ctx
     */
    public void channelActive(ChannelHandlerContext ctx) {
        for (int i=0;i<sendNumber;i++) {
            BookData.book_data.Builder builder = BookData.book_data.newBuilder();
            builder.setId(i);
            builder.setName("Netty");
            builder.setDataTime("2018-11-9 14:25:44");
            builder.setType("buy book");
            ctx.write(builder.build());
        }
        ctx.flush();
    }

    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception{
        System.out.println("Client receive :" + msg);
    }

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        ctx.close();
    }
}

Protobuf的使用注意事項

ProtobufDecoder僅僅負責解碼，它不支援讀半包。因此，在ProtobufDecoder前面，一定要有能夠處理半包的解碼器，有以下三種方式可以選擇：

  1. 使用Netty提供的ProtobufVarint32FrameDecoder,它可以處理半包訊息
  2. 繼承Netty提供的通用半包解碼器LengthFieldBasedFrameDecoder
  3. 繼承ByteToMessageDecoder類，自己處理半包訊息

相關推薦

Netty學習之路（八）-Google Protobuf編碼

Protobuf是一個靈活，高效，結構化的資料序列化框架，相比於XML等傳統的序列化工具，它更小，更快，更簡單。Protobuf支援資料結構化一次可以到處使用，甚至可以跨語言使用，通過程式碼生成工具可以自動生成不同語言版本的原始碼，甚至可以在使用不同版本的資料結構程序間進行資料傳遞，實現資料

Git+Jenkins學習之路（八）之發布maven項目及按版本發布

maven 創建 tar.gz pri .cn class a settings sed family 一、什麽是Maven maven是一個項目管理和綜合工具。Maven提供給開發人員構建一個完整的生命周期框架。 開發團隊可以自動完成該項目的基礎設施建設，

Vue學習之路（八） --- vue-router使用

1.首先安裝vue-router ，在專案的資料夾中使用npm命令即npm install vue-router –save（提示：如果你在vue-cli腳手架工具中已經選擇安裝路由可以不用執行這一步） 2.在實戰中，一般路由放在src/router/index.js中具體操作

Netty學習之路（六）-分隔符和定長解碼器的應用

之前已經使用了LineBasedFrameDecoder解決TCP粘包問題，現在再學兩種解決TCP粘包的方法。 DelimiterBasedFrameDecoder：可以自動完成以分隔符做結束標誌的訊息的解碼，分隔符自定義。 FixedLengthFrameDecoder:

Netty學習之路（五）-TCP粘包/拆包問題

TCP是個“流協議”，所謂流，就是沒有界限的一串資料。TCP底層並不瞭解上層業務資料的具體含義，它會根據TCP緩衝區的實際情況進行包的劃分，所以一個完整的包可能會被TCP拆分成多個包進行傳送，也有可能吧多個小的包封裝成一個大的資料包傳送，這就是TCP粘包和拆包問題。 TCP粘包/拆包產生

Netty學習之路（四）-Netty入門實戰

前面學習了用Java原生NIO的程式設計實踐，過程還是挺複雜的，需要熟練掌握Selector，ServerSocketChannel握，SocketChannel，ByteBuffer等。所以在絕大多數業務場景中我們可以使用Netty來進行NIO程式設計。先總結一下Netty的優點：

Netty學習之路（三）-AIO程式設計

NIO 2.0引入了新的非同步通道的概念，與之前非阻塞IO(NIO)不同的是，NIO 2.0非同步套接字通道是真正的非同步非阻塞I/O，對應於UNIX網路程式設計中的事件驅動I/O(AIO)。它不需要通過多路複用器對註冊的通道進行輪詢操作即可實現非同步讀寫，從而簡化了NIO的程式設計模型。話

Netty學習之路（九）-JBoss Marshalling編解碼

JBoss Marshalling 是一個Java物件序列化包，對JDK預設的序列化框架進行了優化，但又保持跟java.io.Serializable介面的相容，同時增加了一些可調的引數和附加的特性。 Marshalling開發環境準備 下載相關的Marshalling類庫:地址,將

Netty學習之路（二）-非同步IO（NIO）程式設計

NIO到底是什麼簡稱？有人稱之為New I/O，原因為他相對於之前的I/O類庫來說是新增的。這是官方叫法。但是，由於之前老的I/O類庫是阻塞I/O,New I/O類庫的目標就是讓java支援非阻塞I/O,所以更多的人稱之為非阻塞I/O(Non-block I/O)。在開始進行NIO程式設計之

Netty學習之路（一）- 同步與非同步IO

本篇部落格主要是講一些基礎，記錄我的學習過程，同時嘗試養成寫部落格的習慣。內容基本來自Netty權威指南加上一丟丟的個人理解。。。。 I/O基礎入門 在jdk1.4以前，java對i/o的支援並不完善，開發人員在開發高效能i/o時會遇到巨大的挑戰與困難，主要問題如下： 沒

Netty學習之路（七）-編解碼技術

當進行遠端跨程序服務呼叫時，需要把被傳輸的Java物件編碼為位元組陣列或者ByteBuffer物件。而當遠端服務讀取到ByteBuffer物件或者位元組陣列時，需要將其解碼為傳送時的Java物件。這被稱為Java物件編解碼技術。而我們常見得Java序列化僅僅是Java編解碼技術的一種，由於j

Python小白學習之路（八）—【變數】【基本資料型別分類】【集合】【集合的功能】

一、變數 變數的作用：記錄狀態的變化變數的值：描述不同的狀態 二、五大基本資料型別的分類 五大基本資料型別（數字 字串 列表 元祖 字典） 按照可變不可變來進行分類 可變：列表、字典 不可變：字串、數字、元祖 我個人理解，所謂可變，是指當修改內容是，內容改變，地址不變。（即新修改的內容可以覆蓋之

SQL資料庫學習之路（八）

資料庫介紹： 在網頁上的主程式中進行註冊操作，然後把資料傳送給人，人傳遞這些資料到資料庫當中。 為什麼學ADO.NET：之前我們所學的只能在查詢分析器中檢視資料，操作資料。我們不能讓一個普通使用者使用SQL去操作，所以我們搭建一個介面（Web Winform）讓使

Netty學習之路（五）

TCP粘包/拆包問題 TCP是個“流協議”，所謂流，就是沒有界限的一串資料。TCP底層並不瞭解上層業務資料的具體含義，它會根據TCP緩衝區的實際情況進行包的劃分，所以一個完整的包可能會被TCP拆分成多個包進行傳送，也有可能吧多個小的包封裝成一個大的資料包傳送，這

Nginx學習之路（八）Nginx中的事件驅動過程詳解-----以listenfd註冊過程為例

Nginx的高效得益於它的事件驅動機制，整個事件驅動機制基本框架就是linux下的select,poll,epoll這幾個IO多路複用模式，但是nginx絕不單單只是使用它們這麼簡單，今天以epoll模式為例，從nginx最開始的listenfd的監聽的過程來說明nginx是

Django學習之路（八）Admin的學習

Admin 是django自帶的自動化資料管理介面，被授權的使用者可以在Admin中管理資料庫。 配置Admin admin是django自帶的一個app。 1,執行程式： python manage.py createsuperuer #建立超級使用者 之後就會創

JDBC學習之路（八）可滾動結果集和資料分頁

如果在資料庫中有大量資料，然後要求從第150個數據開始讀取，讀10個數據，這種要求在論壇回覆和交流中常見 ，當然了，主流資料庫都提供了指定查詢集的方法，下面我們先來演示一下如果不使用Mysql的查詢集方法怎麼做，當 然，這個在實際開發中應該用處不大，但是作為一種想法還是需要

Netty框架學習之路（一）—— Java網路IO模型

前言 《Unix網路程式設計：卷1》中介紹了5中IO模型，JAVA作為執行在宿主機上的程式，底層也遵循這5中I/O模型規則。這5中I/O模型分別是： 阻塞式IO 非阻塞式IO I/O複用 訊號驅動式IO 非同步IO 按POSIX標準來分，IO分為同步

Netty框架學習之路（五）—— EventLoop及事件迴圈機制

在前面的博文中，我們大致分析瞭解了Channel及其相關概念。在Netty的執行緒模型中，每個channel都有唯一的一個eventLoop與之相繫結，那麼在這篇博文中我們來看一下EvenLoop及其相關概念。 在傳統的Java NIO程式設計中，我們經常使用

Netty框架學習之路（六）—— 引導

前言 之前的博文介紹了Channel、EventLoop及由其衍生出來的相關概念。但這些都是靜態的概念，為了能使Netty程式執行起來，必須要有一個全域性的元件將上述內容串聯起來，這便就是本文所講內容——引導。 引導類 引導類的層次結構如下： 主