Netty结合Protobuf进行编解码的方法

/ / 区分不同的protobuf版本，必须有

syntax = "proto2" ;

package netty;

/ / 生成的目标类的包路径

option java_package = "cn.ddlover.nettystudy.protobuf" ;

/ / 生成的目标类的名字

option java_outer_classname = "subscribereqproto" ;

message subscribereq{

     required int32 subreqid = 1 ;

     required string username = 2 ;

     required string productname = 3 ;

     repeated string address = 4 ;

}

syntax = "proto2" ;

package netty;

option java_package = "cn.ddlover.nettystudy.protobuf" ;

option java_outer_classname = "subscriberespproto" ;

message subscriberesp{

     required int32 subreqid = 1 ;

     required int32 respcode = 2 ;

     required string desc = 3 ;

}

<?xml version = "1.0" encoding = "utf-8" ?>

<project xmlns = "http://maven.apache.org/pom/4.0.0"

          xmlns:xsi = "http://www.w3.org/2001/xmlschema-instance"

          xsi:schemalocation = "http://maven.apache.org/pom/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" >

     <modelversion> 4.0 . 0 < / modelversion>

     <groupid>cn.ddlover< / groupid>

     <artifactid>nettystudy< / artifactid>

     <version> 1.0 - snapshot< / version>

     <properties>

         <java.version> 1.8 < / java.version>

     < / properties>

     <dependencies>

         <! - - https: / / mvnrepository.com / artifact / io.netty / netty - all - - >

         <dependency>

             <groupid>io.netty< / groupid>

             <artifactid>netty - all < / artifactid>

             <version> 4.1 . 33.final < / version>

         < / dependency>

         <! - - https: / / mvnrepository.com / artifact / com.google.protobuf / protobuf - java - - >

         <dependency>

             <groupid>com.google.protobuf< / groupid>

             <artifactid>protobuf - java< / artifactid>

             <version> 3.6 . 1 < / version>

         < / dependency>

         <dependency>

             <groupid>org.projectlombok< / groupid>

             <artifactid>lombok< / artifactid>

             <version> 1.18 . 4 < / version>

         < / dependency>

     < / dependencies>

< / project>

import cn.ddlover.nettystudy.handler.subreqserverhandler;

import cn.ddlover.nettystudy.protobuf.subscribereqproto;

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;

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;

import io.netty.handler.logging.loglevel;

import io.netty.handler.logging.logginghandler;

/ * *

  * protobuf版本图书订购代码

  * /

public class subreqserver {

     private static final int port = 8080 ;

     public static void main(string[] args) {

         bind(port);

     }

     private static void bind( int port) {

         eventloopgroup bossgroup = new nioeventloopgroup();

         eventloopgroup workergroup = new nioeventloopgroup();

         serverbootstrap b = new serverbootstrap();

         try {

             b.group(bossgroup, workergroup)

                     .channel(nioserversocketchannel. class )

                     .option(channeloption.so_backlog, 100 )

                     .handler(new logginghandler(loglevel.info))

                     .childhandler(new channelinitializer<socketchannel>() {

                         @override

                         protected void initchannel(socketchannel socketchannel) throws exception {

                             / / 半包处理

                             socketchannel.pipeline().addlast(new protobufvarint32framedecoder());

                             socketchannel.pipeline().addlast(new protobufdecoder(subscribereqproto.subscribereq.getdefaultinstance()));

                             socketchannel.pipeline().addlast(new protobufvarint32lengthfieldprepender());

                             socketchannel.pipeline().addlast(new protobufencoder());

                             socketchannel.pipeline().addlast(new subreqserverhandler());

                         }

                     });

             channelfuture future = b.bind(port).sync();

             future.channel().closefuture().sync();

         } catch (interruptedexception e) {

             e.printstacktrace();

         } finally {

             bossgroup.shutdowngracefully();

             workergroup.shutdowngracefully();

         }

     }

}

import cn.ddlover.nettystudy.protobuf.subscribereqproto;

import cn.ddlover.nettystudy.protobuf.subscriberespproto;

import io.netty.channel.channelhandlercontext;

import io.netty.channel.channelinboundhandleradapter;

public class subreqserverhandler extends channelinboundhandleradapter {

     @override

     public void channelread(channelhandlercontext ctx, object msg) throws exception {

         subscribereqproto.subscribereq req = (subscribereqproto.subscribereq)msg;

         if ( "张三" .equals(req.getusername())) {

             system.out.println( "server accept clietn subscribe req : [" + req.tostring() + "]" );

             ctx.writeandflush(resp(req.getsubreqid()));

         }

     }

     private subscriberespproto.subscriberesp resp( int subreqid) {

         subscriberespproto.subscriberesp.builder builder = subscriberespproto.subscriberesp.newbuilder();

         builder.setsubreqid(subreqid);

         builder.setrespcode( 0 );

         builder.setdesc( "netty书籍下单成功，3天后将会送到你的住处" );

         return builder. build();

     }

     @override

     public void exceptioncaught(channelhandlercontext ctx, throwable cause) throws exception {

         cause.printstacktrace();

         ctx.close();

     }

}

import cn.ddlover.nettystudy.handler.subreqclienthandler;

import cn.ddlover.nettystudy.protobuf.subscriberespproto;

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

public class subreqclient {

     private static final string host = "localhost" ;

     private static final int port = 8080 ;

     public static void main(string[] args) {

         connect(host, port);

     }

     private static void connect(string host, int port) {

         eventloopgroup group = new nioeventloopgroup();

         bootstrap b = new bootstrap();

         try {

             b.group(group).channel(niosocketchannel. class )

                     .option(channeloption.tcp_nodelay, true)

                     .handler(new channelinitializer<socketchannel>() {

                         @override

                         protected void initchannel(socketchannel socketchannel) throws exception {

                             / / 半包处理

                             socketchannel.pipeline().addlast(new protobufvarint32framedecoder());

                             socketchannel.pipeline().addlast(new protobufdecoder(subscriberespproto.subscriberesp.getdefaultinstance()));

                             socketchannel.pipeline().addlast(new protobufvarint32lengthfieldprepender());

                             socketchannel.pipeline().addlast(new protobufencoder());

                             socketchannel.pipeline().addlast(new subreqclienthandler());

                         }

                     });

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

             f.channel().closefuture().sync();

         } catch (interruptedexception e) {

             e.printstacktrace();

         } finally {

             group.shutdowngracefully();

         }

     }

}

import cn.ddlover.nettystudy.protobuf.subscribereqproto;

import io.netty.channel.channelhandlercontext;

import io.netty.channel.channelinboundhandleradapter;

import java.util.arraylist;

import java.util. list ;

public class subreqclienthandler extends channelinboundhandleradapter {

     @override

     public void channelactive(channelhandlercontext ctx) throws exception {

         for ( int i = 0 ;i< 10 ;i + + ) {

             ctx.write(subreq(i));

         }

         ctx.flush();

     }

     private subscribereqproto.subscribereq subreq( int i) {

         subscribereqproto.subscribereq.builder builder = subscribereqproto.subscribereq.newbuilder();

         builder.setsubreqid(i);

         builder.setusername( "张三" );

         builder.setproductname( "netty book" );

         list <string> address = new arraylist<>();

         address.add( "nanjing yuhuatai" );

         address.add( "beijing liulichang" );

         address.add( "shenzhen hongshulin" );

         builder.addalladdress(address);

         return builder.build();

     }

     @override

     public void channelread(channelhandlercontext ctx, object msg) throws exception {

         system.out.println( "receive server response : [" + msg + "]" );

     }

     @override

     public void exceptioncaught(channelhandlercontext ctx, throwable cause) throws exception {

         cause.printstacktrace();

         ctx.close();

     }

}