netty编程之结合springboot一起使用

写在前面

源码。
本文看下netty结合springboot如何使用。

1：netty server部分

server类（不要main，后续通过springboot来启动咯!）：

package com.dahuyou.netty.springboot.server;import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;import java.net.InetSocketAddress;@Component("nettyServer")
public class NettyServer {private Logger logger = LoggerFactory.getLogger(NettyServer.class);//配置服务端NIO线程组private final EventLoopGroup parentGroup = new NioEventLoopGroup(); //NioEventLoopGroup extends MultithreadEventLoopGroup Math.max(1, SystemPropertyUtil.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));private final EventLoopGroup childGroup = new NioEventLoopGroup();private Channel channel;public ChannelFuture bing(InetSocketAddress address) {ChannelFuture channelFuture = null;try {ServerBootstrap b = new ServerBootstrap();b.group(parentGroup, childGroup).channel(NioServerSocketChannel.class)    //非阻塞模式.option(ChannelOption.SO_BACKLOG, 128).childHandler(new MyChannelInitializer());channelFuture = b.bind(address).syncUninterruptibly();channel = channelFuture.channel();} catch (Exception e) {logger.error(e.getMessage());} finally {if (null != channelFuture && channelFuture.isSuccess()) {logger.info("netty server start done. {大忽悠有限没责任公司}");} else {logger.error("netty server start error. {大忽悠有限没责任公司}");}}return channelFuture;}public void destroy() {if (null == channel) return;channel.close();parentGroup.shutdownGracefully();childGroup.shutdownGracefully();}public Channel getChannel() {return channel;}}

MyChannelInitializer:

package com.dahuyou.netty.springboot.server;import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;import java.nio.charset.Charset;public class MyChannelInitializer extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel channel) {// 解码转String，注意调整自己的编码格式GBK、UTF-8channel.pipeline().addLast(new StringDecoder(Charset.forName("GBK")));// 解码转String，注意调整自己的编码格式GBK、UTF-8channel.pipeline().addLast(new StringEncoder(Charset.forName("GBK")));// 在管道中添加我们自己的接收数据实现方法channel.pipeline().addLast(new MyServerHandler());}}

MyServerHandler:

package com.dahuyou.netty.springboot.server;import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.socket.SocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import java.text.SimpleDateFormat;
import java.util.Date;public class MyServerHandler extends ChannelInboundHandlerAdapter {private Logger logger = LoggerFactory.getLogger(MyServerHandler.class);/*** 当客户端主动链接服务端的链接后，这个通道就是活跃的了。也就是客户端与服务端建立了通信通道并且可以传输数据*/@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {SocketChannel channel = (SocketChannel) ctx.channel();logger.info("链接报告开始");logger.info("链接报告信息：有一客户端链接到本服务端");logger.info("链接报告IP:{}", channel.localAddress().getHostString());logger.info("链接报告Port:{}", channel.localAddress().getPort());logger.info("链接报告完毕");//通知客户端链接建立成功String str = "通知客户端链接建立成功" + " " + new Date() + " " + channel.localAddress().getHostString() + "\r\n";ctx.writeAndFlush(str);}/*** 当客户端主动断开服务端的链接后，这个通道就是不活跃的。也就是说客户端与服务端的关闭了通信通道并且不可以传输数据*/@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {logger.info("客户端断开链接{}", ctx.channel().localAddress().toString());}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {//接收msg消息{与上一章节相比，此处已经不需要自己进行解码}logger.info(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + " 服务端接收到消息：" + msg);//通知客户端链消息发送成功String str = "服务端收到：" + new Date() + " " + msg + "\r\n";ctx.writeAndFlush(str);}/*** 抓住异常，当发生异常的时候，可以做一些相应的处理，比如打印日志、关闭链接*/@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {ctx.close();logger.info("异常信息：\r\n" + cause.getMessage());}}

2：springboot部分

springboot app:

package com.dahuyou.netty.springboot.web;import com.dahuyou.netty.springboot.server.NettyServer;
import io.netty.channel.ChannelFuture;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.ComponentScan;
import java.net.InetSocketAddress;// CommandLineRunner接口方法会在springboot启动完毕之后调用org.springframework.boot.SpringApplication.callRunners
@SpringBootApplication
@ComponentScan("com.dahuyou.netty.springboot")
public class NettyApplication implements CommandLineRunner {@Value("${netty.host}")private String host;@Value("${netty.port}")private int port;@Autowiredprivate NettyServer nettyServer;public static void main(String[] args) {SpringApplication.run(NettyApplication.class, args);}@Overridepublic void run(String... args) throws Exception {InetSocketAddress address = new InetSocketAddress(host, port);ChannelFuture channelFuture = nettyServer.bing(address);// 在jvm钩子中停止netty（jvm正常退出时回调用）Runtime.getRuntime().addShutdownHook(new Thread(() -> nettyServer.destroy()));channelFuture.channel().closeFuture().syncUninterruptibly();}}

这里通过CommandLineRunner接口来启动netty程序，另外为了获取netty的相关状态还定义了一个controller：

package com.dahuyou.netty.springboot.web;import com.dahuyou.netty.springboot.server.NettyServer;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;@RestController
@RequestMapping(value = "/nettyserver", method = RequestMethod.GET)
public class NettyController {@Resourceprivate NettyServer nettyServer;@RequestMapping("/localAddress")public String localAddress() {return "nettyServer localAddress " + nettyServer.getChannel().localAddress();}@RequestMapping("/isOpen")public String isOpen() {return "nettyServer isOpen： " + nettyServer.getChannel().isOpen();}}

3：test case部分

代码：

package com.dahuyou.netty.springboot.test;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.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import java.nio.charset.Charset;
import java.text.SimpleDateFormat;
import java.util.Date;public class ApiTest {public static void main(String[] args) {EventLoopGroup workerGroup = new NioEventLoopGroup();try {Bootstrap b = new Bootstrap();b.group(workerGroup);b.channel(NioSocketChannel.class);b.option(ChannelOption.AUTO_READ, true);b.handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel channel) throws Exception {// 基于换行符号channel.pipeline().addLast(new LineBasedFrameDecoder(1024));// 解码转String，注意调整自己的编码格式GBK、UTF-8channel.pipeline().addLast(new StringDecoder(Charset.forName("GBK")));// 解码转String，注意调整自己的编码格式GBK、UTF-8channel.pipeline().addLast(new StringEncoder(Charset.forName("GBK")));// 在管道中添加我们自己的接收数据实现方法channel.pipeline().addLast(new ChannelInboundHandlerAdapter() {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {//接收msg消息{与上一章节相比，此处已经不需要自己进行解码}System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + " 客户端接收到消息：" + msg);}});}});ChannelFuture f = b.connect("127.0.0.1", 7397).sync();System.out.println("netty client start done. {大忽悠有限没责任公司}");//向服务端发送信息f.channel().writeAndFlush("你好，SpringBoot启动的netty服务端，这里是大忽悠有限没责任公司，你可不要被我忽悠了哦！！！”\r\n");f.channel().writeAndFlush("你好，SpringBoot启动的netty服务端，这里是大忽悠有限没责任公司，你可不要被我忽悠了哦！！！”\r\n");f.channel().writeAndFlush("你好，SpringBoot启动的netty服务端，这里是大忽悠有限没责任公司，你可不要被我忽悠了哦！！！”\r\n");f.channel().writeAndFlush("你好，SpringBoot启动的netty服务端，这里是大忽悠有限没责任公司，你可不要被我忽悠了哦！！！”\r\n");f.channel().writeAndFlush("你好，SpringBoot启动的netty服务端，这里是大忽悠有限没责任公司，你可不要被我忽悠了哦！！！”\r\n");f.channel().closeFuture().syncUninterruptibly();} catch (InterruptedException e) {e.printStackTrace();} finally {workerGroup.shutdownGracefully();}}}

运行client输出:
在这里插入图片描述
服务端输出：

等等，好像有问题，在client明明是write了几句话的，就这样的：

怎么当作一句话来接收了呢？其实，这就是发生了粘包了,知道了问题所在，解决也就简单了，增加一个相关的解码器即可，因为在每句话的结尾我们都增加了换行符，所以这里我们直接使用netty提供的基于换行符的解码器LineBasedFrameDecoder就行,即修改MyChannelInitializer类即可：
在这里插入图片描述
接着重新启动测试，server输出就正常了：

写在后面

参考文章列表

springboot之项目搭建并say hi 。

扩展

在以上的client测试类中，我们在write数据的时候，每句话的结尾都增加了换行符\r\n，这显然是一个重复的工作了。是否可以将这个工作只做一遍呢，可以的，使用netty提供的outboundhandler就可以很轻松的来解决这个问题了，首先我们需要先来定义一个ChannelOutboundHandler:

package com.dahuyou.netty.springboot.test;import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.ChannelPromise;/*** 同意增加换行符的outbound handler*/
public class MyAddNewLineOutMsgHandler extends ChannelOutboundHandlerAdapter {@Override // 调用ctx的writeAndFlush方法时会被该方法拦截执行public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {System.out.println("MyAddNewLineOutMsgHandler--统一增加换行符咯！！！");msg += "\r\n";super.write(ctx, msg, promise);}}