Netty4.x整合SpringBoot2.x使用Protobuf3详解
前言
本篇文章主要介绍的是SpringBoot整合Netty以及使用Protobuf进行数据传输的相关内容。Protobuf会介绍下用法，至于Netty在netty 之 telnet HelloWorld 详解中已经介绍过了，这里就不再过多细说了。
Protobuf
介绍
Protocol Buffer是Google的语言中立的，平台中立的，可扩展机制的，用于序列化结构化数据 – 对比XML，但更小，更快，更简单。您可以定义数据的结构化，然后可以使用特殊生成的源代码轻松地在各种数据流中使用各种语言编写和读取结构化数据。
官网地址: https://developers.google.com/protocol-buffers/
使用
这里的使用就只介绍Java相关的使用。具体protobuf3的使用可以看Protobuf 语言指南(proto3)。 首先我们需要在src/main文件夹下建立一个proto文件夹，然后在该文件夹新建一个user.proto文件，此文件定义我们需要传输的文件。
注：使用grpc方式编译.proto时，会默认扫描src/main/proto文件夹下的protobuf文件。
例如我们需要定义一个用户的信息，包含的字段主要有编号、名称、年龄。 那么该protobuf文件的格式如下: 注：这里使用的是proto3，相关的注释我已写了，这里便不再过多讲述了。需要注意一点的是proto文件和生成的Java文件名称不能一致!
1 //proto3语法注解:如果您不这样做，protobuf编译器将假定您正在使用proto2,这必须是文件的第一个非空的非注释行。
2 syntax = "proto3";
3 //生成的包名
4 option java_package = "com.sanshengshui.netty.protobuf";
5 //生成的java名
6 option java_outer_classname = "UserMsg";
7
8 message User{
9 //ID
10 int32 id = 1;
11 //姓名
12 string name = 2;
13 //年龄
14 int32 age = 3;
15 //状态
16 int32 state = 4;
17 }
创建好该文件之后，我们cd到该工程的根目录下，执行mvn clean compile,输入完之后，回车即可在target文件夹中看到已经生成好的Java文件，然后直接在工程中使用此protobuf文件就可以了。因为能自动扫描到此类。详情请看下图:
注：生成protobuf的文件软件和测试的protobuf文件我也整合到该项目中了，可以直接获取的。
Java文件生成好之后，我们再来看怎么使用。 这里我就直接贴代码了，并且将注释写在代码中，应该更容易理解些吧。。。 代码示例:
@RunWith(JUnit4.class)
@Slf4j
public class NettySpringbootProtostuffApplicationTests {
@Test
public void ProtobufTest() throws IOException {
UserMsg.User.Builder userInfo = UserMsg.User.newBuilder();
userInfo.setId(1);
userInfo.setName("mushuwei");
userInfo.setName("24");
UserMsg.User user = userInfo.build();
// 将数据写到输出流
ByteArrayOutputStream output = new ByteArrayOutputStream();
user.writeTo(output);
// 将数据序列化后发送
byte[] byteArray = output.toByteArray();
// 接收到流并读取
ByteArrayInputStream input = new ByteArrayInputStream(byteArray);
// 反序列化
UserMsg.User userInfo2 = UserMsg.User.parseFrom(input);
log.info("id:" + userInfo2.getId());
log.info("name:" + userInfo2.getName());
log.info("age:" + userInfo2.getAge());
}
}
注：这里说明一点，因为protobuf是通过二进制进行传输，所以需要注意下相应的编码。还有使用protobuf也需要注意一下一次传输的最大字节长度。
输出结果：
17:28:07.914 [main] INFO com.sanshengshui.nettyspringbootprotostuff.NettySpringbootProtostuffApplicationTests - id:1 17:28:07.919 [main] INFO com.sanshengshui.nettyspringbootprotostuff.NettySpringbootProtostuffApplicationTests - name:24 17:28:07.919 [main] INFO com.sanshengshui.nettyspringbootprotostuff.NettySpringbootProtostuffApplicationTests - age:0
Netty整合springboot并使用protobuf进行数据传输
说明：如果想直接获取工程那么可以直接跳到底部，通过链接下载工程代码。
开发准备
环境要求
JDK:：1.8
Netty:：4.0或以上(不包括5)
Protobuf：3.0或以上
如果对Netty不熟的话，可以看看我之前写的netty 之 telnet HelloWorld 详解。大神请无视~。~ 地址：https://www.cnblogs.com/sanshengshui/p/9726306.html
首先还是Maven的相关依赖:
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<java.version>1.8</java.version>
<netty-all.version>4.1.29.Final</netty-all.version>
<protobuf.version>3.6.1</protobuf.version>
<grpc.version>1.15.0</grpc.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!--netty jar包导入-->
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>${netty-all.version}</version>
</dependency>
<!--使用grpc优雅的编译protobuf-->
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
<version>${protobuf.version}</version>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-netty</artifactId>
<version>${grpc.version}</version>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-protobuf</artifactId>
<version>${grpc.version}</version>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-stub</artifactId>
<version>${grpc.version}</version>
</dependency>
<!--lombok用于日志,实体类的重复代码书写-->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<scope>provided</scope>
</dependency>
</dependencies>
添加了相应的maven依赖之后！我们还需要添加grpc优雅的编译protobuf的插件:
<build>
<extensions>
<extension>
<groupId>kr.motd.maven</groupId>
<artifactId>os-maven-plugin</artifactId>
<version>1.5.0.Final</version>
</extension>
</extensions>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>2.5.1</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-resources-plugin</artifactId>
<version>2.7</version>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-source-plugin</artifactId>
<version>2.2.1</version>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
<version>3.0.2</version>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>3.0.0</version>
</plugin>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-dependency-plugin</artifactId>
<executions>
<execution>
<id>copy-protoc</id>
<phase>generate-sources</phase>
<goals>
<goal>copy</goal>
</goals>
<configuration>
<artifactItems>
<artifactItem>
<groupId>com.google.protobuf</groupId>
<artifactId>protoc</artifactId>
<version>${protobuf.version}</version>
<classifier>${os.detected.classifier}</classifier>
<type>exe</type>
<overWrite>true</overWrite>
<outputDirectory>${project.build.directory}</outputDirectory>
</artifactItem>
</artifactItems>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.xolstice.maven.plugins</groupId>
<artifactId>protobuf-maven-plugin</artifactId>
<version>0.5.0</version>
<configuration>
<!--
The version of protoc must match protobuf-java. If you don't depend on
protobuf-java directly, you will be transitively depending on the
protobuf-java version that grpc depends on.
-->
<protocArtifact>com.google.protobuf:protoc:${protobuf.version}:exe:${os.detected.classifier}
</protocArtifact>
<pluginId>grpc-java</pluginId>
<pluginArtifact>io.grpc:protoc-gen-grpc-java:1.0.0:exe:${os.detected.classifier}
</pluginArtifact>
</configuration>
<executions>
<execution>
<goals>
<goal>compile</goal>
<goal>compile-custom</goal>
<goal>test-compile</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
此外我们还需要对application.yml配置文件作一点修改:
server:
enabled: true
bind_address: 0.0.0.0
bind_port: 9876
netty:
#不进行内存泄露的检测
leak_detector_level: DISABLED
boss_group_thread_count: 1
worker_group_thread_count: 12
#最大负载大小
max_payload_size: 65536
项目结构
netty-springboot-protobuf
├── client
├── NettyClient.class -- 客户端启动类
├── NettyClientHandler.class -- 客户端逻辑处理类
├── NettyClientHandler.class -- 客户端初始化类
├── server
├── NettyServer.class -- 服务端启动类
├── NettyServerHandler -- 服务端逻辑处理类
├── NettyServerInitializer -- 服务端初始化类
├── proto
├── user.proto -- protobuf文件
代码编写
代码模块主要分为服务端和客户端。 主要实现的业务逻辑： 服务端启动成功之后，客户端也启动成功，这时服务端会发送一条protobuf格式的信息给客户端，然后客户端给予相应的应答。客户端与服务端连接成功之后，客户端每个一段时间会发送心跳指令给服务端，告诉服务端该客户端还存过中，如果客户端没有在指定的时间发送信息，服务端会关闭与该客户端的连接。当客户端无法连接到服务端之后，会每隔一段时间去尝试重连，只到重连成功!
服务端
首先是编写服务端的启动类，相应的注释在代码中写得很详细了，这里也不再过多讲述了。不过需要注意的是，在之前的我写的Netty文章中，是通过main方法直接启动服务端，因此是直接new一个对象的。而在和SpringBoot整合之后，我们需要将Netty交给springBoot去管理，所以这里就用了相应的注解。 代码如下:
1 @Service("nettyServer")
2 @Slf4j
3 public class NettyServer {
4 /**
5 * 通过springboot读取静态资源,实现netty配置文件的读写
6 */
7
8 @Value("${server.bind_port}")
9 private Integer port;
10
11 @Value("${server.netty.boss_group_thread_count}")
12 private Integer bossGroupThreadCount;
13
14 @Value("${server.netty.worker_group_thread_count}")
15 private Integer workerGroupThreadCount;
16
17 @Value("${server.netty.leak_detector_level}")
18 private String leakDetectorLevel;
19
20 @Value("${server.netty.max_payload_size}")
21 private Integer maxPayloadSize;
22
23 private ChannelFuture channelFuture;
24 private EventLoopGroup bossGroup;
25 private EventLoopGroup workerGroup;
26
27
28 @PostConstruct
29 public void init() throws Exception {
30 log.info("Setting resource leak detector level to {}",leakDetectorLevel);
31 ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.valueOf(leakDetectorLevel.toUpperCase()));
32
33 log.info("Starting Server");
34 //创建boss线程组 用于服务端接受客户端的连接
35 bossGroup = new NioEventLoopGroup(bossGroupThreadCount);
36 // 创建 worker 线程组 用于进行 SocketChannel 的数据读写
37 workerGroup = new NioEventLoopGroup(workerGroupThreadCount);
38 // 创建 ServerBootstrap 对象
39 ServerBootstrap b = new ServerBootstrap();
40 //设置使用的EventLoopGroup
41 b.group(bossGroup, workerGroup)
42 //设置要被实例化的为 NioServerSocketChannel 类
43 .channel(NioServerSocketChannel.class)
44 // 设置 NioServerSocketChannel 的处理器
45 .handler(new LoggingHandler(LogLevel.INFO))
46 // 设置连入服务端的 Client 的 SocketChannel 的处理器
47 .childHandler(new NettyServerInitializer());
48 // 绑定端口，并同步等待成功，即启动服务端
49 channelFuture = b.bind(port).sync();
50
51 log.info("Server started!");
52
53 }
54
55 @PreDestroy
56 public void shutdown() throws InterruptedException {
57 log.info("Stopping Server");
58 try {
59 // 监听服务端关闭，并阻塞等待
60 channelFuture.channel().closeFuture().sync();
61 } finally {
62 // 优雅关闭两个 EventLoopGroup 对象
63 workerGroup.shutdownGracefully();
64 bossGroup.shutdownGracefully();
65 }
66 log.info("server stopped!");
67
68 }
69
70 }
服务端主类编写完毕之后，我们再来设置下相应的过滤条件。 这里需要继承Netty中ChannelInitializer类，然后重写initChannel该方法，进行添加相应的设置，如心跳超时设置，传输协议设置，以及相应的业务实现类。 代码如下:
public class NettyServerInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline ph = ch.pipeline();
//入参说明: 读超时时间、写超时时间、所有类型的超时时间、时间格式
ph.addLast(new IdleStateHandler(5, 0, 0, TimeUnit.SECONDS));
// 解码和编码，应和客户端一致
//传输的协议 Protobuf
ph.addLast(new ProtobufVarint32FrameDecoder());
ph.addLast(new ProtobufDecoder(UserMsg.User.getDefaultInstance()));
ph.addLast(new ProtobufVarint32LengthFieldPrepender());
ph.addLast(new ProtobufEncoder());
//业务逻辑实现类
ph.addLast("nettyServerHandler", new NettyServerHandler());
}
}
服务相关的设置的代码写完之后，我们再来编写主要的业务代码。 使用Netty编写业务层的代码，我们需要继承ChannelInboundHandlerAdapter 或SimpleChannelInboundHandler类，在这里顺便说下它们两的区别吧。 继承SimpleChannelInboundHandler类之后，会在接收到数据后会自动release掉数据占用的Bytebuffer资源。并且继承该类需要指定数据格式。 而继承ChannelInboundHandlerAdapter则不会自动释放，需要手动调用ReferenceCountUtil.release()等方法进行释放。继承该类不需要指定数据格式。 所以在这里，个人推荐服务端继承ChannelInboundHandlerAdapter，手动进行释放，防止数据未处理完就自动释放了。而且服务端可能有多个客户端进行连接，并且每一个客户端请求的数据格式都不一致，这时便可以进行相应的处理。 客户端根据情况可以继承SimpleChannelInboundHandler类。好处是直接指定好传输的数据格式，就不需要再进行格式的转换了。
代码如下:
@Slf4j
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
/** 空闲次数 */
private AtomicInteger idle_count = new AtomicInteger(1);
/** 发送次数 */
private AtomicInteger count = new AtomicInteger(1);
/**
* 建立连接时，发送一条消息
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
log.info("连接的客户端地址:" + ctx.channel().remoteAddress());
UserMsg.User user = UserMsg.User.newBuilder().setId(1).setAge(24).setName("穆书伟").setState(0).build();
ctx.writeAndFlush(user);
super.channelActive(ctx);
}
/**
* 超时处理 如果5秒没有接受客户端的心跳，就触发; 如果超过两次，则直接关闭;
*/
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object obj) throws Exception {
if (obj instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent) obj;
// 如果读通道处于空闲状态，说明没有接收到心跳命令
if (IdleState.READER_IDLE.equals(event.state())) {
log.info("已经5秒没有接收到客户端的信息了");
if (idle_count.get() > 1) {
log.info("关闭这个不活跃的channel");
ctx.channel().close();
}
idle_count.getAndIncrement();
}
} else {
super.userEventTriggered(ctx, obj);
}
}
/**
* 业务逻辑处理
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.info("第" + count.get() + "次" + ",服务端接受的消息:" + msg);
try {
// 如果是protobuf类型的数据
if (msg instanceof UserMsg.User) {
UserMsg.User user = (UserMsg.User) msg;
if (user.getState() == 1) {
log.info("客户端业务处理成功!");
} else if(user.getState() == 2){
log.info("接受到客户端发送的心跳!");
}else{
log.info("未知命令!");
}
} else {
log.info("未知数据!" + msg);
return;
}
} catch (Exception e) {
e.printStackTrace();
} finally {
ReferenceCountUtil.release(msg);
}
count.getAndIncrement();
}
/**
* 异常处理
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
还有个服务端的启动类，之前是通过main方法直接启动， 不过这里改成了通过springBoot进行启动，差别不大。 代码如下：
@SpringBootApplication
@ComponentScan({"com.sanshengshui.netty.server"})
public class NettyServerApp {
/**
* @param args
*/
public static void main(String[] args) {
SpringApplication.run(NettyServerApp.class);
}
}
到这里服务端相应的代码就编写完毕了
版权声明：本文为sanshengshui原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。