项目中需要使用到TCP/IP协议完成数据的发送与接收。如果只是用以前写的简单的socket套接字方法，每次接收发送消息都会创建新的socket再关闭socket，造成资源浪费。于是使用netty框架完成java网络通信。Netty框架的内容很多，这里只是代码展示其中的一个功能。
代码仓库
这里使用的是Springboot+Netty框架，使用maven搭建项目。这里是在一个项目中搭建服务端与客户端，所以端口一样。还可以使用TCP/UTP工具自己搭建服务端和客户端，只要在yml文件中修改ip和端口就好。
pom.xml
<?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>com.hzx.testmaven15netty</groupId> <artifactId>testmaven15netty</artifactId> <version>1.0-SNAPSHOT</version> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.3.0.RELEASE</version> </parent> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter</artifactId> </dependency> <dependency> <groupId>io.netty</groupId> <artifactId>netty-all</artifactId> <version>4.1.31.Final</version> </dependency> </dependencies></project>
application.yml
server: port: 8080# 作为客户端请求的服务端地址netty: tcp: server: # 作为客户端请求的服务端地址 host: 127.0.0.1 # 作为客户端请求的服务端端口 port: 7000 client: # 作为服务端开放给客户端的端口 port: 7000
服务端
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 io.netty.util.concurrent.Future;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.beans.factory.annotation.Value;import org.springframework.stereotype.Component;import java.util.Map;import java.util.concurrent.ConcurrentHashMap;@Componentpublic class NettyTcpServer { private static final Logger LOGGER = LoggerFactory.getLogger(NettyTcpServer.class); // boss事件轮询线程组 // 处理Accept连接事件的线程，这里线程数设置为1即可，netty处理链接事件默认为单线程，过度设置反而浪费cpu资源 private EventLoopGroup boss = new NioEventLoopGroup(1); //worker事件轮询线程组 //处理handler的工作线程，其实也就是处理IO读写 。线程数据默认为 CPU 核心数乘以2 private EventLoopGroup worker = new NioEventLoopGroup(); @Autowired ServerChannelInitializer serverChannelInitializer; @Value("${netty.tcp.client.port}") private Integer port; // 与客户端建立连接后得到的通道对象 private Channel channel; /** * 存储client的channel * key:ip value:Channel */ public static Map<String, Channel> map = new ConcurrentHashMap<String, Channel>(); /** * 开启Netty tcp server服务 * * @return */ public ChannelFuture start() { // 启动类 ServerBootstrap serverBootstrap = new ServerBootstrap(); serverBootstrap.group(boss, worker)//组配置，初始化ServerBootstrap的线程组 .channel(NioServerSocketChannel.class)//构造channel通道工厂 bossGroup的通道，只是负责连接 .childHandler(serverChannelInitializer) //设置通道处理者ChannelHandlerWorkerGroup的处理器 .option(ChannelOption.SO_BACKLOG, 1024)//socket参数，当服务器请求处理程全满时，用于临时存放已完成三次握手请求的队列的最大长度。如果未设置或所设置的值小于1，Java将使用默认值50。 .childOption(ChannelOption.SO_KEEPALIVE, true);//启用心跳保活机制，tcp，默认2小时发一次心跳 //Future：异步任务的生命周期，可用来获取任务结果 ChannelFuture channelFuture1 = serverBootstrap.bind(port).syncUninterruptibly(); // 绑定端口 开启监听 同步等待 if (channelFuture1 != null && channelFuture1.isSuccess()) { channel = channelFuture1.channel();// 获取通道 LOGGER.info("Netty tcp server start success,port={}",port); }else { LOGGER.error("Netty tcp server start fail"); } return channelFuture1; } /** * 停止Netty tcp server服务 */ public void destroy(){ if (channel != null) { channel.close(); } try { Future<?> future = worker.shutdownGracefully().await(); if (!future.isSuccess()) { LOGGER.error("netty tcp workerGroup shutdown fail,{}",future.cause()); } } catch (InterruptedException e) { LOGGER.error(e.toString()); } LOGGER.info("Netty tcp server shutdown success"); }}
import io.netty.channel.ChannelInitializer;import io.netty.channel.ChannelPipeline;import io.netty.channel.socket.SocketChannel;import io.netty.handler.codec.string.StringDecoder;import io.netty.handler.codec.string.StringEncoder;import io.netty.handler.timeout.IdleStateHandler;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;import java.util.concurrent.TimeUnit;@Componentpublic class ServerChannelInitializer extends ChannelInitializer<SocketChannel> { @Autowired ServerChannelHandler serverChannelHandler; @Override protected void initChannel(SocketChannel socketChannel) throws Exception { ChannelPipeline pipeline = socketChannel.pipeline(); //IdleStateHandler心跳机制,如果超时触发Handle中userEventTrigger()方法 pipeline.addLast("idleStateHandler", new IdleStateHandler(15,0,0, TimeUnit.MINUTES)); // 字符串编解码器 pipeline.addLast( new StringDecoder(), new StringEncoder() ); // 自定义Handler pipeline.addLast("serverChannelHandler",serverChannelHandler); }}
import io.netty.channel.ChannelHandler;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.SimpleChannelInboundHandler;import io.netty.handler.timeout.IdleState;import io.netty.handler.timeout.IdleStateEvent;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.stereotype.Component;@Component@ChannelHandler.Sharablepublic class ServerChannelHandler extends SimpleChannelInboundHandler<Object> { private static final Logger LOGGER = LoggerFactory.getLogger(ServerChannelHandler.class); /** * 拿到传过来的msg数据，开始处理 * @param channelHandlerContext * @param msg * @throws Exception */ @Override protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object msg) throws Exception { LOGGER.info("Netty tcp server receive message: {}",msg); channelHandlerContext.writeAndFlush(" response message "+msg).syncUninterruptibly(); } /** * 活跃的、有效的通道 * 第一次连接成功后进入的方法 * @param ctx * @throws Exception */ @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { super.channelActive(ctx); LOGGER.info("tcp client "+getRemoteAddress(ctx)+" connect success"); NettyTcpServer.map.put(getIPString(ctx),ctx.channel()); } /** * 不活动的通道 * 连接丢失后执行的方法（client端可据此实现断线重连） * @param ctx * @throws Exception */ @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { // 删除Channel Map中失效的Client NettyTcpServer.map.remove(getIPString(ctx)); ctx.close(); } /** * 异常处理 * @param ctx * @param cause * @throws Exception */ @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { super.exceptionCaught(ctx, cause); // 发生异常 关闭连接 LOGGER.error("引擎{}的通道发生异常，断开连接",getRemoteAddress(ctx)); ctx.close(); } /** * 心跳机制 超时处理 * @param ctx * @param evt * @throws Exception */ @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { String socketString = ctx.channel().remoteAddress().toString(); if (evt instanceof IdleStateEvent) { IdleStateEvent event = (IdleStateEvent) evt; if (event.state()== IdleState.READER_IDLE) { LOGGER.info("Client: "+socketString+" READER_IDLE读超时"); ctx.disconnect(); }else if (event.state()==IdleState.WRITER_IDLE){ LOGGER.info("Client: "+socketString+" WRITER_IDLE写超时"); ctx.disconnect(); }else if (event.state()==IdleState.ALL_IDLE){ LOGGER.info("Client: "+socketString+" ALL_IDLE总超时"); ctx.disconnect(); } } } /** * 获取client对象：ip+port * @param channelHandlerContext * @return */ public String getRemoteAddress(ChannelHandlerContext channelHandlerContext){ String socketString = ""; socketString = channelHandlerContext.channel().remoteAddress().toString(); return socketString; } /** * 获取client的ip * @param channelHandlerContext * @return */ public String getIPString(ChannelHandlerContext channelHandlerContext){ String ipString = ""; String socketString = channelHandlerContext.channel().remoteAddress().toString(); int colonAt = socketString.indexOf(":"); ipString = socketString.substring(1,colonAt); return ipString; }}
客户端
import io.netty.bootstrap.Bootstrap;import io.netty.channel.Channel;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelOption;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.nio.NioSocketChannel;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.beans.factory.annotation.Value;import org.springframework.stereotype.Component;import javax.xml.ws.Holder;@Componentpublic class NettyTcpClient { private static final Logger LOGGER = LoggerFactory.getLogger(NettyTcpClient.class); @Value("${netty.tcp.server.host}") String HOST; @Value("${netty.tcp.server.port}") int PORT; @Autowired ClientChannelInitializer clientChannelInitializer; private Channel channel; /** * 初始化 `Bootstrap` 客户端引导程序 * @return */ private final Bootstrap getBootstrap(){ Bootstrap bootstrap = new Bootstrap(); NioEventLoopGroup group = new NioEventLoopGroup(); bootstrap.group(group) .channel(NioSocketChannel.class)//通道连接者 .handler(clientChannelInitializer)//通道处理者 .option(ChannelOption.SO_KEEPALIVE,true);// 心跳报活 return bootstrap; } /** * 建立连接，获取连接通道对象 */ public void connect(){ ChannelFuture channelFuture = getBootstrap().connect(HOST, PORT).syncUninterruptibly(); if (channelFuture != null&&channelFuture.isSuccess()) { channel = channelFuture.channel(); LOGGER.info("connect tcp server host = {},port = {} success", HOST,PORT); }else { LOGGER.error("connect tcp server host = {},port = {} fail",HOST,PORT); } } /** * 向服务器发送消息 */ public void sendMessage(Object msg) throws InterruptedException { if (channel != null) { channel.writeAndFlush(msg).sync(); }else { LOGGER.warn("消息发送失败，连接尚未建立"); } }}
import io.netty.channel.ChannelInitializer;import io.netty.channel.ChannelPipeline;import io.netty.channel.socket.SocketChannel;import io.netty.handler.codec.string.StringDecoder;import io.netty.handler.codec.string.StringEncoder;import io.netty.handler.timeout.IdleStateHandler;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;import java.util.concurrent.TimeUnit;@Componentpublic class ClientChannelInitializer extends ChannelInitializer<SocketChannel> { @Autowired ClientChannelHandler clientChannelHandler; @Override protected void initChannel(SocketChannel socketChannel) throws Exception { ChannelPipeline pipeline = socketChannel.pipeline(); pipeline.addLast("idleStateHandler", new IdleStateHandler(15,0,0, TimeUnit.MINUTES)); pipeline.addLast(new StringDecoder(),new StringEncoder()); pipeline.addLast("clientChannelHandler",clientChannelHandler); }}
import io.netty.channel.ChannelHandler;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.SimpleChannelInboundHandler;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.stereotype.Component;@Component@ChannelHandler.Sharablepublic class ClientChannelHandler extends SimpleChannelInboundHandler<Object> { private static final Logger LOGGER = LoggerFactory.getLogger(ClientChannelHandler.class); @Override protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object msg) throws Exception { LOGGER.info("Netty tcp client receive msg : " + msg); }}
启动类
import com.netty.client.NettyTcpClient;import com.netty.server.NettyTcpServer;import io.netty.channel.ChannelFuture;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.CommandLineRunner;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplicationpublic class StartApplication implements CommandLineRunner { public static void main(String[] args) throws Exception { SpringApplication.run(StartApplication.class, args); } @Autowired NettyTcpServer nettyTcpServer; @Autowired NettyTcpClient nettyTcpClient; /** * Callback used to run the bean. * * @param args incoming main method arguments * @throws Exception on error */ @Override public void run(String... args) throws Exception { ChannelFuture start = nettyTcpServer.start(); nettyTcpClient.connect(); for (int i = 0; i < 10; i++) { nettyTcpClient.sendMessage("hello world "+i); } start.channel().closeFuture().syncUninterruptibly(); }}
使用循环让客户端向服务端发送10条数据
运行结果
“C:\Program Files\Java\jdk1.8.0_271\bin\java.exe” -XX:TieredStopAtLevel=1 -noverify -Dspring.output.ansi.enabled=always -Dcom.sun.management.jmxremote -Dspring.jmx.enabled=true -Dspring.liveBeansView.mbeanDomain -Dspring.application.admin.enabled=true “-javaagent:D:\IDEA\IntelliJ IDEA 2019.1.1\lib\idea_rt.jar=62789:D:\IDEA\IntelliJ IDEA 2019.1.1\bin” -Dfile.encoding=UTF-8 -classpath “C:\Program Files\Java\jdk1.8.0_271\jre\lib\charsets.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\deploy.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\ext\access-bridge-64.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\ext\cldrdata.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\ext\dnsns.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\ext\jaccess.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\ext\jfxrt.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\ext\localedata.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\ext\nashorn.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\ext\sqljdbc4-4.0.0.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\ext\sunec.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\ext\sunjce_provider.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\ext\sunmscapi.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\ext\sunpkcs11.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\ext\zipfs.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\javaws.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\jce.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\jfr.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\jfxswt.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\jsse.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\management-agent.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\plugin.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\resources.jar;C:\Program Files\Java\jdk1.8.0_271\jre\lib\rt.jar;D:\Java Code\testmaven15netty\target\classes;D:\Maven\myreprository\org\springframework\boot\spring-boot-starter\2.3.0.RELEASE\spring-boot-starter-2.3.0.RELEASE.jar;D:\Maven\myreprository\org\springframework\boot\spring-boot\2.3.0.RELEASE\spring-boot-2.3.0.RELEASE.jar;D:\Maven\myreprository\org\springframework\spring-context\5.2.6.RELEASE\spring-context-5.2.6.RELEASE.jar;D:\Maven\myreprository\org\springframework\spring-aop\5.2.6.RELEASE\spring-aop-5.2.6.RELEASE.jar;D:\Maven\myreprository\org\springframework\spring-beans\5.2.6.RELEASE\spring-beans-5.2.6.RELEASE.jar;D:\Maven\myreprository\org\springframework\spring-expression\5.2.6.RELEASE\spring-expression-5.2.6.RELEASE.jar;D:\Maven\myreprository\org\springframework\boot\spring-boot-autoconfigure\2.3.0.RELEASE\spring-boot-autoconfigure-2.3.0.RELEASE.jar;D:\Maven\myreprository\org\springframework\boot\spring-boot-starter-logging\2.3.0.RELEASE\spring-boot-starter-logging-2.3.0.RELEASE.jar;D:\Maven\myreprository\ch\qos\logback\logback-classic\1.2.3\logback-classic-1.2.3.jar;D:\Maven\myreprository\ch\qos\logback\logback-core\1.2.3\logback-core-1.2.3.jar;D:\Maven\myreprository\org\slf4j\slf4j-api\1.7.30\slf4j-api-1.7.30.jar;D:\Maven\myreprository\org\apache\logging\log4j\log4j-to-slf4j\2.13.2\log4j-to-slf4j-2.13.2.jar;D:\Maven\myreprository\org\apache\logging\log4j\log4j-api\2.13.2\log4j-api-2.13.2.jar;D:\Maven\myreprository\org\slf4j\jul-to-slf4j\1.7.30\jul-to-slf4j-1.7.30.jar;D:\Maven\myreprository\jakarta\annotation\jakarta.annotation-api\1.3.5\jakarta.annotation-api-1.3.5.jar;D:\Maven\myreprository\org\springframework\spring-core\5.2.6.RELEASE\spring-core-5.2.6.RELEASE.jar;D:\Maven\myreprository\org\springframework\spring-jcl\5.2.6.RELEASE\spring-jcl-5.2.6.RELEASE.jar;D:\Maven\myreprository\org\yaml\snakeyaml\1.26\snakeyaml-1.26.jar;D:\Maven\myreprository\io\netty\netty-all\4.1.31.Final\netty-all-4.1.31.Final.jar” com.netty.StartApplication
. ____ _ __ _ _/\\ / ___’_ __ _ _(_)_ __ __ _ \ \ \ \( ( )\___ | ‘_ | ‘_| | ‘_ \/ _` | \ \ \ \\\/ ___)| |_)| | | | | || (_| | ) ) ) ) ‘ |____| .__|_| |_|_| |_\__, | / / / /=========|_|==============|___/=/_/_/_/:: Spring Boot :: (v2.3.0.RELEASE)
2021-07-13 08:32:17.161 INFO 18068 — [ main] com.netty.StartApplication : Starting StartApplication on LAPTOP-H9JFQJGF with PID 18068 (D:\Java Code\testmaven15netty\target\classes started by huangzixiao in D:\Java Code\testmaven15netty)2021-07-13 08:32:17.165 INFO 18068 — [ main] com.netty.StartApplication : No active profile set, falling back to default profiles: default2021-07-13 08:32:18.371 INFO 18068 — [ main] com.netty.StartApplication : Started StartApplication in 1.706 seconds (JVM running for 2.672)2021-07-13 08:32:18.931 INFO 18068 — [ main] com.netty.server.NettyTcpServer : Netty tcp server start success,port=70002021-07-13 08:32:19.016 INFO 18068 — [ main] com.netty.client.NettyTcpClient : connect tcp server host = 127.0.0.1,port = 7000 success2021-07-13 08:32:19.078 INFO 18068 — [ntLoopGroup-3-1] com.netty.server.ServerChannelHandler : tcp client /127.0.0.1:52653 connect success2021-07-13 08:32:19.100 INFO 18068 — [ntLoopGroup-3-1] com.netty.server.ServerChannelHandler : Netty tcp server receive message: hello world 02021-07-13 08:32:19.100 INFO 18068 — [ntLoopGroup-3-1] com.netty.server.ServerChannelHandler : Netty tcp server receive message: hello world 12021-07-13 08:32:19.100 INFO 18068 — [ntLoopGroup-4-1] com.netty.client.ClientChannelHandler : Netty tcp client receive msg : response message hello world 02021-07-13 08:32:19.100 INFO 18068 — [ntLoopGroup-3-1] com.netty.server.ServerChannelHandler : Netty tcp server receive message: hello world 22021-07-13 08:32:19.100 INFO 18068 — [ntLoopGroup-4-1] com.netty.client.ClientChannelHandler : Netty tcp client receive msg : response message hello world 12021-07-13 08:32:19.100 INFO 18068 — [ntLoopGroup-3-1] com.netty.server.ServerChannelHandler : Netty tcp server receive message: hello world 32021-07-13 08:32:19.100 INFO 18068 — [ntLoopGroup-4-1] com.netty.client.ClientChannelHandler : Netty tcp client receive msg : response message hello world 22021-07-13 08:32:19.100 INFO 18068 — [ntLoopGroup-3-1] com.netty.server.ServerChannelHandler : Netty tcp server receive message: hello world 42021-07-13 08:32:19.100 INFO 18068 — [ntLoopGroup-4-1] com.netty.client.ClientChannelHandler : Netty tcp client receive msg : response message hello world 32021-07-13 08:32:19.104 INFO 18068 — [ntLoopGroup-3-1] com.netty.server.ServerChannelHandler : Netty tcp server receive message: hello world 52021-07-13 08:32:19.104 INFO 18068 — [ntLoopGroup-4-1] com.netty.client.ClientChannelHandler : Netty tcp client receive msg : response message hello world 42021-07-13 08:32:19.104 INFO 18068 — [ntLoopGroup-3-1] com.netty.server.ServerChannelHandler : Netty tcp server receive message: hello world 62021-07-13 08:32:19.104 INFO 18068 — [ntLoopGroup-4-1] com.netty.client.ClientChannelHandler : Netty tcp client receive msg : response message hello world 52021-07-13 08:32:19.104 INFO 18068 — [ntLoopGroup-3-1] com.netty.server.ServerChannelHandler : Netty tcp server receive message: hello world 72021-07-13 08:32:19.104 INFO 18068 — [ntLoopGroup-4-1] com.netty.client.ClientChannelHandler : Netty tcp client receive msg : response message hello world 6 response message hello world 72021-07-13 08:32:19.104 INFO 18068 — [ntLoopGroup-3-1] com.netty.server.ServerChannelHandler : Netty tcp server receive message: hello world 82021-07-13 08:32:19.104 INFO 18068 — [ntLoopGroup-4-1] com.netty.client.ClientChannelHandler : Netty tcp client receive msg : response message hello world 82021-07-13 08:32:19.104 INFO 18068 — [ntLoopGroup-3-1] com.netty.server.ServerChannelHandler : Netty tcp server receive message: hello world 92021-07-13 08:32:19.104 INFO 18068 — [ntLoopGroup-4-1] com.netty.client.ClientChannelHandler : Netty tcp client receive msg : response message hello world 9
到此这篇关于Netty框架实现TCP/IP通信的文章就介绍到这了,更多相关Netty框架实现TCP/IP通信内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持！
不敢面对自己的不完美，总是担心自己的失败，
