文章目录

• 前言
• • 1. 前置准备
  • 2. 消息处理器
  • 3. 重写通道初始化类
  • 4. 核心服务
  • 5. 效果预览
  • 6. 添加通道管理，给指定的客户端发送消息
  • 7. 源码分享

前言

Springboot使用Netty优雅、快速的创建高性能TCP服务器，适合作为开发脚手架进行二次开发。
本博客项目源码地址：

• 项目源码github地址
• 项目源码国内gitee地址

1. 前置准备

• 引入依赖

<dependency>     <groupId>org.springframework.boot</groupId>     <artifactId>spring-boot-starter</artifactId> </dependency> <dependency>     <groupId>org.projectlombok</groupId>     <artifactId>lombok</artifactId>     <optional>true</optional> </dependency>  <dependency>     <groupId>io.netty</groupId>     <artifactId>netty-all</artifactId>     <version>4.1.75.Final</version> </dependency>   <dependency>     <groupId>com.alibaba</groupId>     <artifactId>fastjson</artifactId>     <version>1.2.80</version> </dependency>

• 编写yml配置文件

# tcpnetty:  server:    host: 127.0.0.1    port: 20000    # 传输模式linux上开启会有更高的性能    use-epoll: false# 日记配置logging:  level:    # 开启debug日记打印    com.netty: debug

• 读取YML中的服务配置

/** * 读取YML中的服务配置 * * @author ding */@Configuration@ConfigurationProperties(prefix = ServerProperties.PREFIX)@Datapublic class ServerProperties {    public static final String PREFIX = "netty.server";    /**     * 服务器ip     */    private String ip;    /**     * 服务器端口     */    private Integer port;    /**     * 传输模式linux上开启会有更高的性能     */    private boolean useEpoll;}

2. 消息处理器

import io.netty.channel.ChannelHandler;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.SimpleChannelInboundHandler;import io.netty.handler.timeout.IdleStateEvent;import lombok.RequiredArgsConstructor;import lombok.extern.slf4j.Slf4j;import org.springframework.stereotype.Component;/** * MQTT消息处理,单例启动 * * @author qiding */@Slf4j@Component@ChannelHandler.Sharable@RequiredArgsConstructorpublic class MessageHandler extends SimpleChannelInboundHandler<String> {    @Override    protected void channelRead0(ChannelHandlerContext ctx, String message) throws Exception { log.debug("\n"); log.debug("channelId:" + ctx.channel().id()); log.debug("收到消息:{}", message); // 回复客户端 ctx.writeAndFlush("服务器接收成功！");    }    @Override    public void channelInactive(ChannelHandlerContext ctx) { log.debug("\n"); log.debug("开始连接");    }    @Override    public void channelActive(ChannelHandlerContext ctx) throws Exception { log.debug("\n"); log.debug("成功建立连接,channelId：{}", ctx.channel().id()); super.channelActive(ctx);    }    @Override    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { log.debug("心跳事件时触发"); if (evt instanceof IdleStateEvent) {     log.debug("发送心跳");     IdleStateEvent idleStateEvent = (IdleStateEvent) evt; } else {     super.userEventTriggered(ctx, evt); }    }}

3. 重写通道初始化类

添加我们需要的解码器，这里添加了String解码器和编码器

import com.netty.server.handler.MessageHandler;import io.netty.channel.ChannelInitializer;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 lombok.RequiredArgsConstructor;import org.springframework.stereotype.Component;import java.util.concurrent.TimeUnit;/** * Netty 通道初始化 * * @author qiding */@Component@RequiredArgsConstructorpublic class ChannelInit extends ChannelInitializer<SocketChannel> {    private final MessageHandler messageHandler;    @Override    protected void initChannel(SocketChannel channel) { channel.pipeline()  // 心跳时间  .addLast("idle", new IdleStateHandler(0, 0, 60, TimeUnit.SECONDS))  // 添加解码器  .addLast(new StringDecoder())  // 添加编码器  .addLast(new StringEncoder())  // 添加消息处理器  .addLast("messageHandler", messageHandler);    }}

4. 核心服务

• 接口

public interface ITcpServer {    /**     * 主启动程序，初始化参数     *     * @throws Exception 初始化异常     */    void start() throws Exception;    /**     * 优雅的结束服务器     *     * @throws InterruptedException 提前中断异常     */    @PreDestroy    void destroy() throws InterruptedException;}

• 服务实现

/** * 启动 Server * * @author qiding */@Component@Slf4j@RequiredArgsConstructorpublic class TcpServer implements ITcpServer {    private final ChannelInit channelInit;    private final ServerProperties serverProperties;    private EventLoopGroup bossGroup;    private EventLoopGroup workerGroup;    @Override    public void start() { log.info("初始化 TCP server ..."); bossGroup = serverProperties.isUseEpoll() ? new EpollEventLoopGroup() : new NioEventLoopGroup(); workerGroup = serverProperties.isUseEpoll() ? new EpollEventLoopGroup() : new NioEventLoopGroup(); this.tcpServer();    }    /**     * 初始化     */    private void tcpServer() { try {     new ServerBootstrap()      .group(bossGroup, workerGroup)      .channel(serverProperties.isUseEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)      .localAddress(new InetSocketAddress(serverProperties.getPort()))      // 配置 编码器、解码器、业务处理      .childHandler(channelInit)      // tcp缓冲区      .option(ChannelOption.SO_BACKLOG, 128)      // 将网络数据积累到一定的数量后,服务器端才发送出去,会造成一定的延迟。希望服务是低延迟的,建议将TCP_NODELAY设置为true      .childOption(ChannelOption.TCP_NODELAY, false)      // 保持长连接      .childOption(ChannelOption.SO_KEEPALIVE, true)      // 绑定端口，开始接收进来的连接      .bind().sync();     log.info("tcpServer启动成功！开始监听端口：{}", serverProperties.getPort()); } catch (Exception e) {     e.printStackTrace();     bossGroup.shutdownGracefully();     workerGroup.shutdownGracefully(); }    }    /**     * 销毁     */    @PreDestroy    @Override    public void destroy() { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully();    }}

5. 效果预览

• 启动类添加启动方法

/** * @author ding */@SpringBootApplication@RequiredArgsConstructorpublic class NettyServerApplication implements ApplicationRunner {    private final TcpServer tcpServer;    public static void main(String[] args) { SpringApplication.run(NettyServerApplication.class, args);    }    @Override    public void run(ApplicationArguments args) throws Exception { tcpServer.start();    }}

• 运行
  

• 打开tcp客户端工具进行测试
  

6. 添加通道管理，给指定的客户端发送消息

为了给指定客户端发送消息，我们需要设置一个登录机制，保存登录成功的客户端ID和频道的关系

• 编写通道存储类

/** * 频道信息存储 * 
 * 封装netty的频道存储，客户端id和频道双向绑定 * * @author qiding */@Slf4jpublic class ChannelStore {    /**     * 频道绑定 key     */    private final static AttributeKey<Object> CLIENT_ID = AttributeKey.valueOf("clientId");    /**     * 客户端和频道绑定     */    private final static ConcurrentHashMap<String, ChannelId> CLIENT_CHANNEL_MAP = new ConcurrentHashMap<>(16);    /**     * 存储频道     */    public final static ChannelGroup CHANNEL_GROUP = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);    /**     * 重入锁     */    private static final Lock LOCK = new ReentrantLock();    /**     * 获取单机连接数量     */    public static int getLocalConnectCount() { return CHANNEL_GROUP.size();    }    /**     * 获取绑定的通道数量（测试用）     */    public static int getBindCount() { return CLIENT_CHANNEL_MAP.size();    }    /**     * 绑定频道和客户端id     *     * @param ctx      连接频道     * @param clientId 用户id     */    public static void bind(ChannelHandlerContext ctx, String clientId) { LOCK.lock(); try {     // 释放旧的连接     closeAndClean(clientId);     // 绑定客户端id到频道上     ctx.channel().attr(CLIENT_ID).set(clientId);     // 双向保存客户端id和频道     CLIENT_CHANNEL_MAP.put(clientId, ctx.channel().id());     // 保存频道     CHANNEL_GROUP.add(ctx.channel()); } finally {     LOCK.unlock(); }    }    /**     * 是否已登录     */    public static boolean isAuth(ChannelHandlerContext ctx) { return !StringUtil.isNullOrEmpty(getClientId(ctx));    }    /**     * 获取客户端id     *     * @param ctx 连接频道     */    public static String getClientId(ChannelHandlerContext ctx) { return ctx.channel().hasAttr(CLIENT_ID) ? (String) ctx.channel().attr(CLIENT_ID).get() : "";    }    /**     * 获取频道     *     * @param clientId 客户端id     */    public static Channel getChannel(String clientId) { return Optional.of(CLIENT_CHANNEL_MAP.containsKey(clientId))  .filter(Boolean::booleanValue)  .map(b -> CLIENT_CHANNEL_MAP.get(clientId))  .map(CHANNEL_GROUP::find)  .orElse(null);    }    /**     * 释放连接和资源     * CLIENT_CHANNEL_MAP 需要释放     * CHANNEL_GROUP 不需要释放，netty会自动帮我们移除     *     * @param clientId 客户端id     */    public static void closeAndClean(String clientId) { // 清除绑定关系 Optional.of(CLIENT_CHANNEL_MAP.containsKey(clientId))  .filter(Boolean::booleanValue)  .ifPresent(oldChannel -> CLIENT_CHANNEL_MAP.remove(clientId)); // 若存在旧连接，则关闭旧连接，相同clientId，不允许重复连接 Optional.ofNullable(getChannel(clientId))  .ifPresent(ChannelOutboundInvoker::close);    }    public static void closeAndClean(ChannelHandlerContext ctx) { closeAndClean(getClientId(ctx));    }}

• 配置登录机制

我们在消息处理器 MessageHandler 中修改channelRead0方法，模拟登录

    @Override    protected void channelRead0(ChannelHandlerContext ctx, String message) throws Exception { log.debug("\n"); log.debug("channelId:" + ctx.channel().id()); log.debug("收到消息:{}", message); // 判断是否未登录 if (!ChannelStore.isAuth(ctx)) {     // 登录逻辑自行实现，我这里为了演示把第一次发送的消息作为客户端ID     String clientId = message.trim();     ChannelStore.bind(ctx, clientId);     log.debug("登录成功");     ctx.writeAndFlush("login successfully");     return; } // 回复客户端 ctx.writeAndFlush("ok");    }    /**     * 指定客户端发送     *     * @param clientId 其它已成功登录的客户端     * @param message  消息     */    public void sendByClientId(String clientId, String message) { Channel channel = ChannelStore.getChannel(clientId); channel.writeAndFlush(message);    }

调用sendByClientId即可给已登录的其它客户端发送消息了。

7. 源码分享

• Springboot-cli开发脚手架，集合各种常用框架使用案例，完善的文档，致力于让开发者快速搭建基础环境并让应用跑起来。
• 项目源码国内gitee地址
• 项目源码github地址