[Java知识库]SpringBoot集成Netty实现消息推送、接收、自动重连

Netty

背景

之前一直想做一个Netty项目，集成消息发送、消息接收、主动推送、身份验证、自动重连等功能方便实际项目使用时直接拿来使用。因为时间问题一直拖延至今o(╯□╰)o，刚好最近公司有相关业务功能，自己大力推行，所以今天趁热打铁一气呵成，到时候可以直接使用。

介绍

Netty 是一个异步事件驱动的网络应用框架用于快速开发可维护的高性能协议服务器和客户端。

Netty 是一个 NIO 客户端服务器框架，可以快速轻松地开发协议服务器和客户端等网络应用程序。 它极大地简化和流线了网络编程，例如 TCP 和 UDP 套接字服务器。

简单并不意味着生成的应用程序会受到可维护性或性能问题的影响。 Netty 是根据从实现许多协议（如 FTP、SMTP、HTTP 以及各种二进制和基于文本的旧协议）中获得的经验精心设计的。 因此，Netty 成功地找到了一种方法，可以在不妥协的情况下实现易于开发、性能、稳定性和灵活性。

特性

设计

1、针对多种传输类型的统一接口 - 阻塞和非阻塞

2、简单但更强大的线程模型

3、真正的无连接的数据报套接字支持

4、链接逻辑支持复用

性能

1、比核心 Java API 更好的吞吐量，较低的延时

2、资源消耗更少，这个得益于共享池和重用

3、减少内存拷贝

安全

1、完整的 SSL / TLS 和 StartTLS 的支持

2、运行在受限的环境例如 Applet 或 OSGI

实战

首先到spring官网初始化下载一个SpringBoot项目，添加三个子模块：common、client、server

server：Netty服务提供者

client：Netty服务连接者

common：通用包、工具类

添加Netty依赖信息，不推荐5版本，因为官方不再支持，所以选择了4版本

<dependency>
   <groupId>io.netty</groupId>
   <artifactId>netty-all</artifactId>
   <version>4.1.70.Final</version>
</dependency>

server

1、编写Netty服务配置

package com.whl.mq.server;

import com.whl.mq.handle.NettyServerChannelInitializer;
import io.netty.bootstrap.ServerBootstrap;
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 lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

/**
 * @author hl.Wu
 *
 * @date 2022/07/17
 * @description: netty server info
 **/
@Slf4j
@Component
public class NioNettyServer {

    @Value("${nio.netty.server.port: 8099}")
    private int port;
    @Autowired
    private NettyServerChannelInitializer nettyServerChannelInitializer;

    @Async
    public void start() {
        log.info("start to netty server port is {}", port);
        // 接收连接
        EventLoopGroup boss = new NioEventLoopGroup();
        // 处理信息
        EventLoopGroup worker = new NioEventLoopGroup();
        try {
            // 定义server
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            // 添加分组
            serverBootstrap.group(boss, worker)
                // 添加通道设置非阻塞
                .channel(NioServerSocketChannel.class)
                // 服务端可连接队列数量
                .option(ChannelOption.SO_BACKLOG, 128)
                // 开启长连接
                .childOption(ChannelOption.SO_KEEPALIVE, Boolean.TRUE)
                // 流程处理
                .childHandler(nettyServerChannelInitializer);
            // 绑定端口
            ChannelFuture cf = serverBootstrap.bind(port).sync();
            // 优雅关闭连接
            cf.channel().closeFuture().sync();
        } catch (Exception e) {
            log.error("connection error",e.getMessage(), e);
        } finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }

}

2、编写业务处理流程?NettyServerChannelInitializer

package com.whl.mq.handle;

import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @author hl.Wu
 * @date 2022/7/14
 **/
@Component
@Slf4j
public class NettyServerChannelInitializer extends ChannelInitializer {
    @Autowired
    private ClientMessageHandler clientMessageHandler;
    @Autowired
    private ClientUserHandler clientUserHandler;
    @Override
    protected void initChannel(Channel channel) {
        // 设置编码类型
        channel.pipeline().addLast("decoder",new StringDecoder(CharsetUtil.UTF_8));
        // 设置解码类型
        channel.pipeline().addLast("encoder",new StringEncoder(CharsetUtil.UTF_8));
        // 用户校验处理逻辑
        channel.pipeline().addLast("ClientTokenHandler", clientUserHandler);
        // 通过校验最终消息业务处理
        channel.pipeline().addLast("ClientMessageHandler",clientMessageHandler);
    }
}

3、编写用户Token身份信息校验逻辑

ChannelHandler.Sharable：一个ChannelHandler如果使用了@Sharable注解，就可以只在bootstrap中创建一个实例，它就可以被添加到一或多个pipeline中且不存在竞争，这样可以减少同一类handler的new和GC，节省资源，提高效率。

package com.whl.mq.handle;

import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.whl.common.bo.MessageBO;
import com.whl.common.bo.ResponseBO;
import com.google.common.base.Preconditions;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author hl.Wu
 * @date 2022/7/14
 **/
@Slf4j
@Component
@ChannelHandler.Sharable
public class ClientUserHandler extends SimpleChannelInboundHandler<String> {

    /**
     * k：token,v：userId
     */
    public static Map<String,String> userMap = new ConcurrentHashMap<>();

    /**
     * 通道信息 k:userId v:通道信息
     */
    public Map<String, Channel> channelMap = new ConcurrentHashMap<>();

    @Override
    public void channelRead0(ChannelHandlerContext ctx, String msg) {
        // 解析消息对象 校验token信息
        log.info("receive message info is {}", msg);
        String token = (String)JSONUtil.getByPath(JSONUtil.parse(msg), MessageBO.Fields.token);
        // 校验token是否失效
        // TODO 根据业务场景添加身份校验
        Preconditions.checkArgument(userMap.containsKey(token),"抱歉，Token已失效");
        if(!channelMap.containsKey(userMap.get(token))){
            channelMap.put(userMap.get(token), ctx.channel());
        }
        // 开启访问后面的处理逻辑
        ctx.fireChannelRead(msg);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        Channel channel = ctx.channel();
        ResponseBO resp = ResponseBO.builder()
            .code(HttpStatus.FORBIDDEN.toString())
            .success(Boolean.FALSE)
            .build();
        // 返回错误code
        if(cause instanceof IllegalArgumentException){
            resp.setMessage(cause.getMessage());
            channel.writeAndFlush(JSON.toJSONString(resp));
            log.warn("Token 校验未通过,{}", channel.localAddress());
        }
    }
}

4、编写ClientMessageHandler消息最终处理逻辑

package com.whl.mq.handle;

import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.whl.common.bo.MessageBO;
import com.whl.common.bo.ResponseBO;
import io.netty.channel.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Component;

/**
 * @author hl.Wu
 * @date 2022/7/14
 **/
@Component
@Slf4j
@ChannelHandler.Sharable
public class ClientMessageHandler extends SimpleChannelInboundHandler<String> {

    @Override
    public void channelRead0(ChannelHandlerContext ctx, String msg) {
        log.info("接收到的信息数据={}", msg);
        // 返回请求结果
        String requestUid = (String)JSONUtil.getByPath(JSONUtil.parse(msg), MessageBO.Fields.requestUid);
        ResponseBO resp = ResponseBO
            .builder()
            .requestUid(requestUid)
            .code(HttpStatus.OK.toString())
            .success(Boolean.TRUE)
            .message("请求成功")
            .build();
        Channel channel = ctx.channel();
        channel.writeAndFlush(JSONUtil.toJsonStr(resp));
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        Channel channel = ctx.channel();
        ResponseBO resp = ResponseBO.builder()
            .code(HttpStatus.INTERNAL_SERVER_ERROR.toString())
            .success(Boolean.FALSE)
            .build();
        // 返回错误code
        if(cause instanceof IllegalArgumentException){
            resp.setMessage(cause.getMessage());
            channel.writeAndFlush(JSON.toJSONString(resp));
            log.warn("业务异常请排查， ；{}", cause.getMessage(), cause);
        }
        log.error("error message {}",cause.getMessage(),cause);
    }

}

5、编写对外提供Token的接口控制类

package com.whl.mq.controller;

import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.ObjectUtil;
import com.whl.mq.handle.ClientUserHandler;
import io.netty.channel.Channel;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.util.Map;

/**
 * @author hl.Wu
 * @date 2022/7/14
 **/
@RestController
@RequestMapping("/user")
@Api(tags = "用户管理")
public class UserController {
    @Autowired
    private ClientUserHandler clientUserHandler;

    @GetMapping("/token")
    @ApiOperation("获取token信息")
    public String getToken(){
        String token = IdUtil.fastSimpleUUID();
        ClientUserHandler.userMap.put(token,token);
        return token;
    }

    @PostMapping("/tips")
    @ApiOperation("发送提醒")
    public void sendToClient(@RequestParam("tips") String tips, @RequestParam("userId") String userId){
        Map<String, Channel> channelMap = clientUserHandler.channelMap;
        Channel channel = channelMap.get(userId);
        if(ObjectUtil.isNotNull(channel)){
            channel.writeAndFlush(tips);
        }
    }
}

6、最后在启动服务自动加载启动Netty方法，设置异步启动

package com.whl.mq;

import com.whl.mq.server.NioNettyServer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.scheduling.annotation.EnableAsync;

@SpringBootApplication
@EnableAsync
@Slf4j
public class NettyServerApplication implements CommandLineRunner {

	public static void main(String[] args) {
		ConfigurableApplicationContext context = SpringApplication.run(NettyServerApplication.class, args);
		// start to netty server
		context.getBean(NioNettyServer.class).start();
	}

	@Override
	public void run(String... args) {
		log.info("========================server start success========================");
	}

}

Client

1、编写client连接Netty服务配置

PostConstruct：服务启动后会调用该方法，不能含有任何参数
PreDestroy：服务关闭前会调用该方法，不能含有任何参数

package com.whl.client.client;

import com.whl.client.handle.NettyClientChannelInitializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.TimeUnit;

/**
 * @author hl.Wu
 * @date 2021/11/9
 *
 * @description: netty client info
 **/
@Slf4j
@Component
public class NioNettyClient {

    @Value("${netty.server.host:localhost}")
    private String host;

    @Value("${netty.server.port:8099}")
    private int port;

    private SocketChannel socketChannel;

    /**
     * work 线程组用于数据处理
     */
    private EventLoopGroup work = new NioEventLoopGroup();

    @Autowired
    private NettyClientChannelInitializer nettyClientChannelInitializer;

    /**
     * 发送消息
     *
     * @param msg
     */
    public void sendMsg(String msg) {
        if(!socketChannel.isActive()){
            // 如果失去连接，重新创建新的连接
            log.info("****************服务失去连接，开始创建新的连接****************");
            start();
        }
        // 发送消息
        socketChannel.writeAndFlush(msg);
    }

    @PostConstruct
    public void start() {
        work = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(work)
                // 设置通道
                .channel(NioSocketChannel.class)
                // 日志处理格式
                .handler(new LoggingHandler(LogLevel.INFO))
                // 禁用nagle算法
                .option(ChannelOption.TCP_NODELAY, true)
                // 保持长连接
                .option(ChannelOption.SO_KEEPALIVE, Boolean.TRUE)
                // 流程处理
                .handler(nettyClientChannelInitializer);
            // start to channel
            ChannelFuture future = bootstrap.connect(host, port).sync();
            future.addListener((ChannelFutureListener) future1 -> {
                if (future1.isSuccess()) {
                    log.info("**********************服务连接成功**********************");
                } else {
                    log.warn("**********************服务连接失败，20s后开始重新连接服务器**********************");
                    // 20s后重新连接
                    future1.channel().eventLoop().schedule(() -> this.start(), 20, TimeUnit.SECONDS);
                }
            });
            socketChannel = (SocketChannel) future.channel();
        } catch (Exception e) {
            log.error("connection error", e.getMessage(), e);
        }
    }

    @PreDestroy
    private void close() {
        if(socketChannel != null){
            socketChannel.close();
        }
        work.shutdownGracefully();
    }
}

2、编写NettyClientChannelInitializer接收服务端响应消息业务处理流程

注：

这个实例中我没有使用注解方式，因为心跳处理逻辑会引用重新连接服务的bean，如果使用注解方式会出现循环依赖错误

package com.whl.client.handle;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
 * @author hl.Wu
 * @date 2022/7/14
 **/
@Component
@Slf4j
public class NettyClientChannelInitializer extends ChannelInitializer {

    private final ResponseChannelHandler responseChannelHandler;

    private final ClientHeartbeatHandler clientChannelHandler;

    public NettyClientChannelInitializer(ResponseChannelHandler responseChannelHandler,
        ClientHeartbeatHandler clientChannelHandler) {
        this.responseChannelHandler = responseChannelHandler;
        this.clientChannelHandler = clientChannelHandler;
    }

    @Override
    protected void initChannel(Channel channel) {
        channel.pipeline().addLast("decoder",new StringDecoder(CharsetUtil.UTF_8));
        channel.pipeline().addLast("encoder",new StringEncoder(CharsetUtil.UTF_8));
        channel.pipeline().addLast("responseChannelHandler",responseChannelHandler);
        channel.pipeline().addLast("clientChannelHandler",clientChannelHandler);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("error message {}",cause.getMessage(),cause);
        super.exceptionCaught(ctx, cause);
    }
}

3、编写心跳处理逻辑ClientHeartbeatHandler

package com.whl.client.handle;

import com.alibaba.fastjson.JSON;
import com.whl.common.bo.MessageBO;
import com.whl.client.client.NioNettyClient;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * 心跳监测
 *
 * @author hl.Wu
 * @date 2022/7/15
 **/
@Component
@Slf4j
@ChannelHandler.Sharable
public class ClientHeartbeatHandler extends ChannelInboundHandlerAdapter {

    private final NioNettyClient nioNettyClient;

    public ClientHeartbeatHandler(NioNettyClient nioNettyClient) {
        this.nioNettyClient = nioNettyClient;
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if(evt instanceof IdleStateEvent){
            IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
            if (idleStateEvent.state() == IdleState.WRITER_IDLE) {
                log.info("已经10s没有发送消息给服务端");
                //发送心跳消息，并在发送失败时关闭该连接
                MessageBO message = new MessageBO();
                message.setHeartbeat(Boolean.TRUE);
                ctx.writeAndFlush(JSON.toJSONString(message)).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        //如果运行过程中服务端挂了,执行重连机制
        log.info("start to reconnect netty server");
        ctx.channel().eventLoop().schedule(() -> nioNettyClient.start(), 3L, TimeUnit.SECONDS);
        super.channelInactive(ctx);
    }

}

4、编写接收服务端响应消息最终处理逻辑ResponseChannelHandler

package com.whl.client.handle;

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
 * @author hl.Wu
 * @date 2022/7/15
 **/
@Component
@Slf4j
@ChannelHandler.Sharable
public class ResponseChannelHandler extends SimpleChannelInboundHandler<String> {
    protected void channelRead0(ChannelHandlerContext ctx, String msg) {
        log.info("response info is {}", msg);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
    }
}

5、最后编写发送消息接口

package com.whl.client.controller;

import com.alibaba.fastjson.JSON;
import com.whl.common.bo.MessageBO;
import com.whl.client.client.NioNettyClient;
import io.swagger.annotations.Api;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author hl.Wu
 * @date 2022/7/14
 **/
@RestController
@RequestMapping("/client")
@Api(tags = "客户端管理")
@Slf4j
public class ClientController {
    @Autowired
    private NioNettyClient nioNettyClient;

    @PostMapping("/send")
    public void send(@RequestBody MessageBO message) {
        log.info(JSON.toJSONString(message));
        nioNettyClient.sendMsg(JSON.toJSONString(message));
    }
}

common

1、编写公共实体对象

package com.whl.common.bo;

import java.io.Serializable;

/**
 * @author hl.Wu
 * @date 2022/7/14
 **/
public abstract class BaseBO implements Serializable {

    protected String toLog() {
        return null;
    }
}

package com.whl.common.bo;

import lombok.Data;
import lombok.experimental.FieldNameConstants;

import java.util.List;

/**
 * @author hl.Wu
 * @date 2022/7/14
 **/
@Data
@FieldNameConstants
public class MessageBO extends BaseBO{

    /**
     * 请求唯一标识id
     */
    private String requestUid;

    private String token;

    private Boolean heartbeat;

    private List<VerificationEmsBO> data;
}

package com.whl.common.bo;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * @author hl.Wu
 * @date 2022/7/14
 **/
@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class ResponseBO extends BaseBO{

    private String requestUid;
    /**
     * 响应code
     */
    private String code;

    /**
     * 响应提示信息
     */
    private String message;

    /**
     * 是否成功
     */
    private Boolean success;
}

package com.whl.common.bo;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * 验证码短信
 *
 * @author hl.Wu
 * @date 2022/7/14
 **/
@Data
@AllArgsConstructor
@NoArgsConstructor
public class VerificationEmsBO extends BaseBO{

    /**
     * 手机号码
     */
    private String mobile;

    /**
     * 短信内容
     */
    private String message;
}

?测试

1、启动server服务

访问server服务，http://localhost:8081/doc.html

2、启动Client服务

访问client服务，http://localhost:8082/doc.html

3、请求服务端token接口?

4、将返回的token信息组装至client发送接口

5、查看后台接收日志

client

?server

?6、测试失败重连机制

操作：重新启动服务端，服务启动成功后再次请求client发送接口，实现重连

?7、测试服务端主动给客户端推送消息

操作：调用服务端发送提醒接口，userId使用客户端访问成功时的token

client日志

总结

大功告成，测试所有功能一切正常，完美运行。

1、身份验证、心跳机制只是简单处理，实际项目中可以使用Redis进行优化。

2、用户信息存放过于简单，没有使用数据库

3、因为项目中添加了RocketMQ，所以需要先安装RocketMQ才能运行成功

哈哈！一天一点儿小知识，今天你学会了吗？

参考地址

Netty: Home

项目源码

https://gitee.com/mackjie/whl-netty

GitHub - mackjie/whl-netty: SpringBoot集成Netty实现消息推送、接收、自动重连

分支：develop-nomq