netty使用protobuf和心跳检查、断线重连

1、下载安装protoc ，本人使用3.6.1版本

https://github.com/protocolbuffers/protobuf/releases

下载protoc，选择对应的操作系统

2、user.proto文件内容如下

syntax ="proto3";

// 生成的包名

option java_package="com.pancm.protobuf";

//生成的java名 java类名不要和proto文件名一样

option java_outer_classname ="UserInfo";

message UserMsg {

// ID

int32 id =1;

// 姓名

string name =2;

// 年龄

int32 age =3;

// 状态

int32 state =4;

}

3、netty server端和client端pom文件都添加 protobuf依赖

<artifactId>grpc-netty</artifactId>

</dependency>

</dependency>

<artifactId>grpc-protobuf</artifactId>

</dependency>

<groupId>com.google.protobuf</groupId>

<artifactId>protobuf-java</artifactId>

</dependency>

4、netty 服务端和客户端pom文件在build标签添加

<groupId>kr.motd.maven</groupId>

<artifactId>os-maven-plugin</artifactId>

<version>1.5.0.Final</version>

</extension>

</extensions>

5、netty 服务端和客户端pom文件添加 plugin

<groupId>org.xolstice.maven.plugins</groupId>

<artifactId>protobuf-maven-plugin</artifactId>

<protocExecutable>E:\protoc-3.6.1-win32\bin\protoc.exe</protocExecutable>

</configuration>

<goals>

<goal>compile</goal>

</goals>

</execution>

</executions>

</plugin>

6、netty服务端编译安装，生成UserInfo.java

7、netty服务端实现

package com.tcp.protobuf;

import com.pancm.protobuf.UserInfo;

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;

import io.netty.handler.timeout.IdleStateHandler;

import java.util.concurrent.TimeUnit;

public class TcpServerProto {

private static final StringIP ="127.0.0.1";

private static final int PORT =9999;

/**

* 用于分配处理业务线程的线程组个数

protected static final int BIZGROUPSIZE = Runtime.getRuntime().availableProcessors() *2;// 默认

/**

* 业务出现线程大小

protected static final int BIZTHREADSIZE =4;

* NioEventLoopGroup实际上就是个线程池,

* NioEventLoopGroup在后台启动了n个NioEventLoop来处理Channel事件,

* 每一个NioEventLoop负责处理m个Channel,

* NioEventLoopGroup从NioEventLoop数组里挨个取出NioEventLoop来处理Channel

private static final EventLoopGroupbossGroup =new NioEventLoopGroup(BIZGROUPSIZE);

private static final EventLoopGroupworkerGroup =new NioEventLoopGroup(BIZTHREADSIZE);

public static void run() {

try {

ServerBootstrap b =new ServerBootstrap();

b.group(bossGroup,workerGroup)

.option(ChannelOption.SO_BACKLOG,1024)// 设置tcp缓冲区

.childOption(ChannelOption.SO_KEEPALIVE,true)

.channel(NioServerSocketChannel.class)

.childHandler(new ChannelInitializer() {

@Override

public void initChannel(SocketChannel ch)throws Exception {

ChannelPipeline pipeline = ch.pipeline();

//入参说明: 读超时时间、写超时时间、所有类型的超时时间、时间格式

//超时处理如果30秒没有接受客户端的心跳

ch.pipeline().addLast(new IdleStateHandler(30,0,0, TimeUnit.SECONDS));

// 解码和编码，应和客户端一致

//传输的协议 Protobuf

ch.pipeline().addLast(new ProtobufVarint32FrameDecoder());

ch.pipeline().addLast(new ProtobufDecoder(UserInfo.UserMsg.getDefaultInstance()));

ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());

ch.pipeline().addLast(new ProtobufEncoder());

// 处理网络IO

pipeline.addLast(new NettyServerHandlerProto());

}

});

// 绑定端口同步等待绑定成功

ChannelFuture f = b.bind(PORT).sync();// (7)

// 等到服务端监听端口关闭

f.channel().closeFuture().sync();

System.out.println("TCP服务器已启动");

}catch (InterruptedException e) {

System.out.println(e);

}

protected static void shutdown() {

// 优雅释放线程资源

workerGroup.shutdownGracefully();

bossGroup.shutdownGracefully();

}

8、NettyServerHandlerProto

package com.tcp.protobuf;

import com.pancm.protobuf.UserInfo;

import io.netty.channel.ChannelHandlerContext;

import io.netty.channel.ChannelInboundHandlerAdapter;

import io.netty.handler.timeout.IdleState;

import io.netty.handler.timeout.IdleStateEvent;

import io.netty.util.ReferenceCountUtil;

/**

* @Description

* @Date 2019/10/17 15:56

* @Author zsj

public class NettyServerHandlerProtoextends ChannelInboundHandlerAdapter {

/**

* 空闲次数

private int idle_count=1;

/**

* 发送次数

private int count =1;

/**

* 建立连接时，发送一条消息

@Override

public void channelActive(ChannelHandlerContext ctx)throws Exception {

System.out.println("连接的客户端地址:" + ctx.channel().remoteAddress());

UserInfo.UserMsg userMsg = UserInfo.UserMsg.newBuilder().setId(1).setAge(18).setName("xuwujing").setState(0)

.build();

ctx.writeAndFlush(userMsg);

super.channelActive(ctx);

}

/**

* 超时处理如果30秒没有接受客户端的心跳，就触发; 如果超过两次，则直接关闭;

@Override

public void userEventTriggered(ChannelHandlerContext ctx, Object obj)throws Exception {

if (objinstanceof IdleStateEvent) {

IdleStateEvent event = (IdleStateEvent) obj;

if (IdleState.READER_IDLE.equals(event.state())) {// 如果读通道处于空闲状态，说明没有接收到心跳命令

System.out.println("已经30秒没有接收到客户端的信息了");

if (idle_count >1) {

System.out.println("关闭这个不活跃的channel");

ctx.channel().close();

}

idle_count++;

}

}else {

super.userEventTriggered(ctx, obj);

}

/**

* 业务逻辑处理

@Override

public void channelRead(ChannelHandlerContext ctx, Object msg)throws Exception {

System.out.println("第" +count +"次" +",服务端接受的消息:" + msg);

try {

// 如果是protobuf类型的数据

if (msginstanceof UserInfo.UserMsg) {

UserInfo.UserMsg userState = (UserInfo.UserMsg) msg;

if (userState.getState() ==1) {

System.out.println("客户端业务处理成功!");

}else if (userState.getState() ==2) {

System.out.println("接受到客户端发送的心跳!");

}else {

System.out.println("未知命令!");

}

}else {

System.out.println("未知数据!" + msg);

return;

}

}catch (Exception e) {

e.printStackTrace();

}finally {

ReferenceCountUtil.release(msg);

}

count++;

}

/**

* 异常处理

@Override

public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)throws Exception {

cause.printStackTrace();

ctx.close();

}

客户端实现待续。。。。