Springboot整合Netty，自定義協(xié)議實(shí)現(xiàn)
Springboot整合Netty
新建springboot項(xiàng)目，并在項(xiàng)目以來中導(dǎo)入netty包，用fastjson包處理jsonStr。
<!-- netty -->
? ? ? ?<dependency>
? ? ? ? ? ?<groupId>io.netty</groupId>
? ? ? ? ? ?<artifactId>netty-all</artifactId>
? ? ? ? ? ?<version>4.1.42.Final</version>
? ? ? ?</dependency>
?
? ? ? ?<!-- Json處理 -->
? ? ? ?<dependency>
? ? ? ? ? ?<groupId>com.alibaba.fastjson2</groupId>
? ? ? ? ? ?<artifactId>fastjson2</artifactId>
? ? ? ? ? ?<version>2.0.16</version>
? ? ? ?</dependency>創(chuàng)建netty相關(guān)配置信息文件
-
yml配置文件——application.yml
# netty 配置
netty:
?# boss線程數(shù)量
boss: 4
?# worker線程數(shù)量
worker: 2
?# 連接超時(shí)時(shí)間
timeout: 6000
?# 服務(wù)器主端口
port: 18023
?# 服務(wù)器備用端口
portSalve: 18026
?# 服務(wù)器地址
host: 127.0.0.1-
netty配置實(shí)體類——NettyProperties與yml配置文件綁定 通過
@ConfigurationProperties(prefix = "netty")注解讀取配置文件中的netty配置，通過反射注入值，需要在實(shí)體類中提供對應(yīng)的setter和getter方法。
@ConfigurationProperties(prefix = "netty")對應(yīng)的實(shí)體類屬性名稱不要求一定相同，只需保證“set”字符串拼接配置文件的屬性和setter方法名相同即可。
@Configuration
@ConfigurationProperties(prefix = "netty")
public class NettyProperties {
?
? ?/**
? ? * boss線程數(shù)量
? ? */
? ?private Integer boss;
?
? ?/**
? ? * worker線程數(shù)量
? ? */
? ?private Integer worker;
?
? ?/**
? ? * 連接超時(shí)時(shí)間
? ? */
? ?private Integer timeout = 30000;
?
? ?/**
? ? * 服務(wù)器主端口
? ? */
? ?private Integer port = 18023;
?
? ?/**
? ? * 服務(wù)器備用端口
? ? */
? ?private Integer portSalve = 18026;
?
? ?/**
? ? * 服務(wù)器地址 默認(rèn)為本地
? ? */
? ?private String host = "127.0.0.1";
// setter、getter 。。。。
}-
對netty進(jìn)行配置，綁定netty相關(guān)配置設(shè)置 Netty通常由一個(gè)Bootstrap開始，主要作用是配置整個(gè)Netty程序，串聯(lián)各個(gè)組件，Netty中Bootstrap類是客戶端程序的啟動引導(dǎo)類，ServerBootstrap是服務(wù)端啟動引導(dǎo)類。
@Configuration
@EnableConfigurationProperties
public class NettyConfig {
? ?final NettyProperties nettyProperties;
?
? ?public NettyConfig(NettyProperties nettyProperties) {
? ? ? ?this.nettyProperties = nettyProperties;
? }
?
? ?/**
? ? * boss線程池-進(jìn)行客戶端連接
? ? *
? ? * @return
? ? */
? ?@Bean
? ?public NioEventLoopGroup boosGroup() {
? ? ? ?return new NioEventLoopGroup(nettyProperties.getBoss());
? }
?
? ?/**
? ? * worker線程池-進(jìn)行業(yè)務(wù)處理
? ? *
? ? * @return
? ? */
? ?@Bean
? ?public NioEventLoopGroup workerGroup() {
? ? ? ?return new NioEventLoopGroup(nettyProperties.getWorker());
? }
?
? ?/**
? ? * 服務(wù)端啟動器，監(jiān)聽客戶端連接
? ? *
? ? * @return
? ? */
? ?@Bean
? ?public ServerBootstrap serverBootstrap() {
? ? ? ?ServerBootstrap serverBootstrap = new ServerBootstrap()
? ? ? ? ? ? ? ?// 指定使用的線程組
? ? ? ? ? ? ? .group(boosGroup(), workerGroup())
? ? ? ? ? ? ? ?// 指定使用的通道
? ? ? ? ? ? ? .channel(NioServerSocketChannel.class)
? ? ? ? ? ? ? ?// 指定連接超時(shí)時(shí)間
? ? ? ? ? ? ? .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, nettyProperties.getTimeout())
? ? ? ? ? ? ? ?// 指定worker處理器
? ? ? ? ? ? ? .childHandler(new NettyServerHandler());
? ? ? ?return serverBootstrap;
? }
}-
worker處理器，初始化通道以及配置對應(yīng)管道的處理器 自定義了
##@##分割符，通過DelimiterBasedFrameDecoder來處理拆包沾包問題； 通過MessageDecodeHandler將接收消息解碼處理成對象實(shí)例； 通過MessageEncodeHandler將發(fā)送消息增加分割符后并編碼； 最后通過ServerListenerHandler根據(jù)消息類型對應(yīng)處理不同消息。
public class NettyServerHandler extends ChannelInitializer<SocketChannel> {
? ?@Override
? ?protected void initChannel(SocketChannel socketChannel) throws Exception {
? ? ? ?// 數(shù)據(jù)分割符
? ? ? ?String delimiterStr = "##@##";
? ? ? ?ByteBuf delimiter = Unpooled.copiedBuffer(delimiterStr.getBytes());
? ? ? ?ChannelPipeline pipeline = socketChannel.pipeline();
? ? ? ?// 使用自定義處理拆包/沾包，并且每次查找的最大長度為1024字節(jié)
? ? ? ?pipeline.addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
? ? ? ?// 將上一步解碼后的數(shù)據(jù)轉(zhuǎn)碼為Message實(shí)例
? ? ? ?pipeline.addLast(new MessageDecodeHandler());
? ? ? ?// 對發(fā)送客戶端的數(shù)據(jù)進(jìn)行編碼，并添加數(shù)據(jù)分隔符
? ? ? ?pipeline.addLast(new MessageEncodeHandler(delimiterStr));
? ? ? ?// 對數(shù)據(jù)進(jìn)行最終處理
? ? ? ?pipeline.addLast(new ServerListenerHandler());
? }
}-
數(shù)據(jù)解碼 數(shù)據(jù)解碼和編碼都采用UTF8格式
public class MessageDecodeHandler extends ByteToMessageDecoder {
?
? ?@Override
? ?protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf in, List<Object> list) throws Exception {
? ? ? ?ByteBuf frame = in.retainedDuplicate();
? ? ? ?final String content = frame.toString(CharsetUtil.UTF_8);
? ? ? ?Message message = new Message(content);
? ? ? ?list.add(message);
? ? ? ?in.skipBytes(in.readableBytes());
? }
}-
數(shù)據(jù)解碼轉(zhuǎn)換的實(shí)例 Message類用于承載消息、轉(zhuǎn)JsonString
public class Message {
? ?/**
? ? * 數(shù)據(jù)長度
? ? */
? ?private Integer len;
?
? ?/**
? ? * 接收的通訊數(shù)據(jù)body
? ? */
? ?private String content;
?
? ?/**
? ? * 消息類型
? ? */
? ?private Integer msgType;
?
? ?public Message(Object object) {
? ? ? ?String str = object.toString();
? ? ? ?JSONObject jsonObject = JSONObject.parseObject(str);
? ? ? ?msgType = Integer.valueOf(jsonObject.getString("msg_type"));
? ? ? ?content = jsonObject.getString("body");
? ? ? ?len = str.length();
? }
?
? ?public String toJsonString() {
? ? ? ?return "{" +
? ? ? ? ? ? ? ?"\"msg_type\": " + msgType + ",\n" +
? ? ? ? ? ? ? ?"\"body\": " + content +
? ? ? ? ? ? ? ?"}";
? }
// setter、getter 。。。。
}-
數(shù)據(jù)編碼 netty服務(wù)端回復(fù)消息時(shí)，對消息轉(zhuǎn)JsonString增加分割符，并進(jìn)行編碼。
public class MessageEncodeHandler extends MessageToByteEncoder<Message> {
? ?// 數(shù)據(jù)分割符
? ?String delimiter;
?
? ?public MessageEncodeHandler(String delimiter) {
? ? ? ?this.delimiter = delimiter;
? }
?
? ?@Override
? ?protected void encode(ChannelHandlerContext channelHandlerContext, Message message, ByteBuf out) throws Exception {
? ? ? ?out.writeBytes((message.toJsonString() + delimiter).getBytes(CharsetUtil.UTF_8));
? }
}-
數(shù)據(jù)處理器，針對不同類型數(shù)據(jù)分類處理 在處理不同接收數(shù)據(jù)時(shí)使用了枚舉類型，在使用switch時(shí)可以做下處理，具體參考代碼，這里只演示如何操作，并沒實(shí)現(xiàn)數(shù)據(jù)處理業(yè)務(wù)類。
public class ServerListenerHandler extends SimpleChannelInboundHandler<Message> {
? ?private static final Logger log = LoggerFactory.getLogger(ServerListenerHandler.class);
?
? ?/**
? ? * 設(shè)備接入連接時(shí)處理
? ? *
? ? * @param ctx
? ? */
? ?@Override
? ?public void handlerAdded(ChannelHandlerContext ctx) {
? ? ? ?log.info("有新的連接：[{}]", ctx.channel().id().asLongText());
? }
?
? ?/**
? ? * 數(shù)據(jù)處理
? ? *
? ? * @param ctx
? ? * @param msg
? ? */
? ?@Override
? ?protected void channelRead0(ChannelHandlerContext ctx, Message msg) {
? ? ? ?// 獲取消息實(shí)例中的消息體
? ? ? ?String content = msg.getContent();
? ? ? ?// 對不同消息類型進(jìn)行處理
? ? ? ?MessageEnum type = MessageEnum.getStructureEnum(msg);
? ? ? ?switch (type) {
? ? ? ? ? ?case CONNECT:
? ? ? ? ? ? ? ?// TODO 心跳消息處理
? ? ? ? ? ?case STATE:
? ? ? ? ? ? ? ?// TODO 設(shè)備狀態(tài)
? ? ? ? ? ?default:
? ? ? ? ? ? ? ?System.out.println(type.content + "消息內(nèi)容" + content);
? ? ? }
? }
?
? ?/**
? ? * 設(shè)備下線處理
? ? *
? ? * @param ctx
? ? */
? ?@Override
? ?public void handlerRemoved(ChannelHandlerContext ctx) {
? ? ? ?log.info("設(shè)備下線了:{}", ctx.channel().id().asLongText());
? }
?
? ?/**
? ? * 設(shè)備連接異常處理
? ? *
? ? * @param ctx
? ? * @param cause
? ? */
? ?@Override
? ?public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
? ? ? ?// 打印異常
? ? ? ?log.info("異常：{}", cause.getMessage());
? ? ? ?// 關(guān)閉連接
? ? ? ?ctx.close();
? }
}-
數(shù)據(jù)類型枚舉類
public enum MessageEnum {
? ?CONNECT(1, "心跳消息"),
? ?STATE(2, "設(shè)備狀態(tài)");
?
? ?public final Integer type;
? ?public final String content;
?
? ?MessageEnum(Integer type, String content) {
? ? ? ?this.type = type;
? ? ? ?this.content = content;
? }
?
? ?// case中判斷使用
? ?public static MessageEnum getStructureEnum(Message msg) {
? ? ? ?Integer type = Optional.ofNullable(msg)
? ? ? ? ? ? ? .map(Message::getMsgType)
? ? ? ? ? ? ? .orElse(0);
? ? ? ?if (type == 0) {
? ? ? ? ? ?return null;
? ? ? } else {
? ? ? ? ? ?List<MessageEnum> objectEnums = Arrays.stream(MessageEnum.values())
? ? ? ? ? ? ? ? ? .filter((item) -> item.getType() == type)
? ? ? ? ? ? ? ? ? .distinct()
? ? ? ? ? ? ? ? ? .collect(Collectors.toList());
? ? ? ? ? ?if (objectEnums.size() > 0) {
? ? ? ? ? ? ? ?return objectEnums.get(0);
? ? ? ? ? }
? ? ? ? ? ?return null;
? ? ? }
? }
// setter、getter。。。。
}到此Netty整個(gè)配置已經(jīng)完成，但如果要跟隨springboot一起啟動，仍需要做一些配置。
-
netty啟動類配置
@Component
public class NettyServerBoot {
? ?private static final Logger log = LoggerFactory.getLogger(NettyServerBoot.class);
? ?@Resource
? ?NioEventLoopGroup boosGroup;
? ?@Resource
? ?NioEventLoopGroup workerGroup;
? ?final ServerBootstrap serverBootstrap;
? ?final NettyProperties nettyProperties;
?
? ?public NettyServerBoot(ServerBootstrap serverBootstrap, NettyProperties nettyProperties) {
? ? ? ?this.serverBootstrap = serverBootstrap;
? ? ? ?this.nettyProperties = nettyProperties;
? }
?
?
? ?/**
? ? * 啟動netty
? ? *
? ? * @throws InterruptedException
? ? */
? ?@PostConstruct
? ?public void start() throws InterruptedException {
? ? ? ?// 綁定端口啟動
? ? ? ?serverBootstrap.bind(nettyProperties.getPort()).sync();
? ? ? ?// 備用端口
? ? ? ?serverBootstrap.bind(nettyProperties.getPortSalve()).sync();
? ? ? ?log.info("啟動Netty: {},{}", nettyProperties.getPort(), nettyProperties.getPortSalve());
? }
?
? ?/**
? ? * 關(guān)閉netty
? ? */
? ?@PreDestroy
? ?public void close() {
? ? ? ?log.info("關(guān)閉Netty");
? ? ? ?boosGroup.shutdownGracefully();
? ? ? ?workerGroup.shutdownGracefully();
? }
}增加NettyServerBoot配置后，啟動application時(shí)，netty服務(wù)端會跟隨一起啟動。 
文章來源：http://www.zghlxwxcb.cn/news/detail-735954.html
同時(shí)，在springboot關(guān)閉前，會先銷毀netty服務(wù)。文章來源地址http://www.zghlxwxcb.cn/news/detail-735954.html
到了這里，關(guān)于Springboot整合Netty，自定義協(xié)議實(shí)現(xiàn)的文章就介紹完了。如果您還想了解更多內(nèi)容，請?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
