1. 创建SpringBoot父工程
创建一个SpringBoot工程,然后创建三个子模块
整体工程目录:一个server服务(netty服务器),两个client服务(netty客户端)
pom文件引入netty依赖,springboot依赖
<?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>
<packaging>pom</packaging>
<modules>
<module>server</module>
<module>client1</module>
<module>client2</module>
</modules>
<!--继承 SpringBoot 框架的一个父项目,所有自己开发的 Spring Boot 都必须的继承-->
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.1.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>org.example</groupId>
<artifactId>nettyTest</artifactId>
<version>1.0-SNAPSHOT</version>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>8</source>
<target>8</target>
</configuration>
</plugin>
</plugins>
</build>
<dependencies>
<!--SpringBoot 框架 web 项目起步依赖,通过该依赖自动关联其它依赖,不需要我们一个一个去添加-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.34.Final</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.16.18</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.1.23</version>
</dependency>
</dependencies>
</project>
2. 创建Netty服务器
NettySpringBootApplication
package server;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class NettySpringBootApplication {
public static void main(String[] args) {
SpringApplication.run(NettySpringBootApplication.class, args);
}
}
NettyServiceHandler
package server.netty;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
public class NettyServiceHandler extends SimpleChannelInboundHandler<String> {
private static final ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
// 获取到当前与服务器连接成功的channel
Channel channel = ctx.channel();
group.add(channel);
System.out.println(channel.remoteAddress() + " 上线," + "在线数量:" + group.size());
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
// 获取到当前要断开连接的Channel
Channel channel = ctx.channel();
System.out.println(channel.remoteAddress() + "下线," + "在线数量:" + group.size());
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
Channel channel = ctx.channel();
System.out.println("netty客户端" + channel.remoteAddress() + "发送过来的消息:" + msg);
group.forEach(ch -> { // JDK8 提供的lambda表达式
if (ch != channel) {
ch.writeAndFlush(channel.remoteAddress() + ":" + msg + "\n");
}
});
}
public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable throwable) throws Exception {
throwable.printStackTrace();
channelHandlerContext.close();
}
}
SocketInitializer
package server.netty;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.bytes.ByteArrayDecoder;
import io.netty.handler.codec.bytes.ByteArrayEncoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.springframework.stereotype.Component;
@Component
public class SocketInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
// // 添加对byte数组的编解码,netty提供了很多编解码器,你们可以根据需要选择
// pipeline.addLast(new ByteArrayDecoder());
// pipeline.addLast(new ByteArrayEncoder());
//添加一个基于行的解码器
pipeline.addLast(new LineBasedFrameDecoder(2048));
pipeline.addLast(new StringDecoder());
pipeline.addLast(new StringEncoder());
// 添加上自己的处理器
pipeline.addLast(new NettyServiceHandler());
}
}
NettyServer
package server.netty;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
@Component
@Slf4j
public class NettyServer {
private final static Logger logger = LoggerFactory.getLogger(NettyServer.class);
@Resource
private SocketInitializer socketInitializer;
@Getter
private ServerBootstrap serverBootstrap;
/**
* netty服务监听端口
*/
@Value("${netty.port:6666}")
private int port;
/**
* 主线程组数量
*/
@Value("${netty.bossThread:1}")
private int bossThread;
/**
* 启动netty服务器
*/
public void start() {
this.init();
this.serverBootstrap.bind(this.port);
logger.info("Netty started on port: {} (TCP) with boss thread {}", this.port, this.bossThread);
}
/**
* 初始化netty配置
*/
private void init() {
NioEventLoopGroup bossGroup = new NioEventLoopGroup(this.bossThread);
NioEventLoopGroup workerGroup = new NioEventLoopGroup();
this.serverBootstrap = new ServerBootstrap();
this.serverBootstrap.group(bossGroup, workerGroup) // 两个线程组加入进来
.channel(NioServerSocketChannel.class) // 配置为nio类型
.option(ChannelOption.SO_BACKLOG, 128) //设置线程队列等待连接个数
.childOption(ChannelOption.SO_KEEPALIVE, true)
.childHandler(this.socketInitializer); // 加入自己的初始化器
}
}
NettyStartListener
package server.netty;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/**
* 监听Spring容器启动完成,完成后启动Netty服务器
* @author Gjing
**/
@Component
public class NettyStartListener implements ApplicationRunner {
@Resource
private NettyServer nettyServer;
@Override
public void run(ApplicationArguments args) throws Exception {
this.nettyServer.start();
}
}
application.yml
server:
port: 8000
netty:
port: 6666
bossThread: 1
3. 创建Netty客户端
Client1
package client;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class Client1 {
public static void main(String[] args) {
SpringApplication.run(Client1.class, args);
}
}
NettyClientHandler
package client.netty;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
public class NettyClientHandler extends SimpleChannelInboundHandler<String> {
public void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) {
System.out.println("收到服务端消息:" + channelHandlerContext.channel().remoteAddress() + "的消息:" + msg);
}
public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable throwable) throws Exception {
channelHandlerContext.close();
}
}
SocketInitializer
package client.netty;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
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 org.springframework.stereotype.Component;
@Component
public class SocketInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
pipeline.addLast(new LineBasedFrameDecoder(2048));
pipeline.addLast(new StringDecoder());
pipeline.addLast(new StringEncoder());
pipeline.addLast(new NettyClientHandler()); //加入自己的处理器
}
}
NettyClient
package client.netty;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
@Component
@Slf4j
public class NettyClient {
private final static Logger logger = LoggerFactory.getLogger(NettyClient.class);
@Resource
private SocketInitializer socketInitializer;
@Getter
private Bootstrap bootstrap;
@Getter
private Channel channel;
/**
* netty服务监听端口
*/
@Value("${netty.port:6666}")
private int port;
@Value("${netty.host:127.0.0.1}")
private String host;
/**
* 启动netty
*/
public void start() {
this.init();
this.channel = this.bootstrap.connect(host, port).channel();
logger.info("Netty connect on port: {}, the host {}, the channel {}", this.port, this.host, this.channel);
try {
InputStreamReader is = new InputStreamReader(System.in, StandardCharsets.UTF_8);
BufferedReader br = new BufferedReader(is);
while (true) {
System.out.println("输入:");
this.channel.writeAndFlush(br.readLine() + "\r\n");
}
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 初始化netty配置
*/
private void init() {
EventLoopGroup group = new NioEventLoopGroup();
this.bootstrap = new Bootstrap();
//设置线程组
bootstrap.group(group)
.channel(NioSocketChannel.class) //设置客户端的通道实现类型
.handler(this.socketInitializer);
}
}
NettyStartListener
package client.netty;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
@Component
public class NettyStartListener implements ApplicationRunner {
@Resource
private NettyClient nettyClient;
@Override
public void run(ApplicationArguments args) throws Exception {
this.nettyClient.start();
}
}
application.yml
server:
port: 8001
netty:
port: 6666
host: "127.0.0.1"
然后按照相同的方法创建client2
4. 测试