前言

Springboot使用Netty优雅、快速的创建TCP客户端。

1. 前置准备

pom.xml

 <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <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>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </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.75</version>
        </dependency>

application.yml

server:
  port: 10000

spring:
  application:
    name: tcp-client

# tcp
tcp:
  client:
    host: 127.0.0.1
    port: 19999
  # 要连接的服务器ip和端口
  server:
    host: 127.0.0.1
    port: 20000

# 日记配置
logging:
  level:
    com.netty: debug

2. 消息处理器

MessageHandler.java


import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;


/**
 * 消息处理,单例启动
 *
 * @author qiding
 */
@Slf4j
@Component
@ChannelHandler.Sharable
@RequiredArgsConstructor
public class MessageHandler extends SimpleChannelInboundHandler<MqttMessage> {


    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MqttMessage mqttMessage) throws Exception {
        log.debug("\n");
        log.debug("channelId:" + ctx.channel().id());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        log.debug("\n");
        log.debug("开始连接");
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.debug("\n");
        log.info("成功建立连接,channelId：{}", ctx.channel().id());
        super.channelActive(ctx);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        log.info("心跳事件时触发");
        if (evt instanceof IdleStateEvent) {
            log.info("发送心跳");
            // 此处编写发送心跳代码
            // ctx.writeAndFlush("ping");
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}

3. 重写通道初始化类

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

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
@RequiredArgsConstructor
public class ChannelInit extends ChannelInitializer<SocketChannel> {

    private final MessageHandler messageHandler;

    @Override
    protected void initChannel(SocketChannel channel) {
        channel.pipeline()
                // 每隔60s的时间触发一次userEventTriggered的方法，并且指定IdleState的状态位是WRITER_IDLE,事件触发给服务器发送ping消息
                .addLast("idle", new IdleStateHandler(0, 60, 0, TimeUnit.SECONDS))
                // 添加解码器
                .addLast(new StringDecoder())
                // 添加编码器
                .addLast(new StringEncoder())
                // 添加消息处理器
                .addLast("messageHandler", messageHandler);
    }

}

4. 核心服务

接口
ITcpClient.java

import javax.annotation.PreDestroy;

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

服务实现
TcpClient.java

import com.netty.client.channel.ChannelInit;
import io.netty.bootstrap.Bootstrap;
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.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.PreDestroy;

/**
 * 启动 Broker
 *
 * @author qiding
 */
@Component
@Slf4j
@RequiredArgsConstructor
public class TcpClient implements ITcpClient {

    private final ChannelInit channelInit;

    private static final EventLoopGroup WORKER_GROUP = new NioEventLoopGroup();

    private SocketChannel socketChannel;

    private Bootstrap bootstrap;


    @Value("${tcp.client.host}")
    private String clientHost;

    @Value("${tcp.client.port}")
    private Integer clientPort;

    @Value("${tcp.server.host}")
    private String serverHost;

    @Value("${tcp.server.port}")
    private Integer serverPort;

    @Override
    public void start() throws Exception {
        log.info("初始化 MQTT Client ...");
        this.tcpClient();
    }

    @Override
    public void reconnect() throws Exception {
        if (socketChannel != null && socketChannel.isActive()) {
            socketChannel.close();
            this.connect(serverHost, serverPort);
        }

    }

    public void disconnect() {
        if (socketChannel != null && socketChannel.isActive()) {
            socketChannel.close();
        }
    }

    /**
     * mqttBroker初始化
     */
    private void tcpClient() {
        try {
            bootstrap = new Bootstrap()
                    .remoteAddress(clientHost, clientPort)
                    .handler(channelInit)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true);
            bootstrap.group(WORKER_GROUP);
            this.connect(serverHost, serverPort);
        } catch (Exception e) {
            e.printStackTrace();
            WORKER_GROUP.shutdownGracefully();
        }
    }

    /**
     * 连接服务器
     */
    public void connect(String ip, Integer port) throws InterruptedException {
        this.disconnect();
        this.serverHost = ip;
        this.serverPort = port;
        ChannelFuture future = bootstrap.connect(serverHost, serverPort).sync();
        if (future.isSuccess()) {
            socketChannel = (SocketChannel) future.channel();
            log.info("connect server success");
        }
    }
    
    /**
     * 销毁
     */
    @PreDestroy
    @Override
    public void destroy() {
        WORKER_GROUP.shutdownGracefully();
        socketChannel.closeFuture();
    }

    /**
     * 获取频道
     */
    public SocketChannel getSocketChannel() {
        return this.socketChannel;
    }
}

5. 效果预览

在这里插入图片描述

6. 编写http测试接口

为了方便测试，我们编写几个http接口进行测试
HttpApi.java

import com.alibaba.fastjson.JSONObject;
import com.netty.client.server.TcpClient;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

/**
 * 模拟发送api
 *
 * @author qiding
 */
@RequiredArgsConstructor
@RestController
@Slf4j
public class HttpApi {

    private final TcpClient tcpClient;

    /**
     * 消息发布
     */
    @GetMapping("/send")
    public String send(String message) {
        tcpClient.getSocketChannel().writeAndFlush(message);
        return "发送成功";
    }
    
    /**
     * 消息发布
     */
    @PostMapping("/send/json")
    public String send(@RequestBody JSONObject body) {
        tcpClient.getSocketChannel().writeAndFlush(body.toJSONString());
        return "发送成功";
    }

    /**
     * 连接
     */
    @GetMapping("connect")
    public String connect(String ip, Integer port) throws Exception {
        tcpClient.connect(ip, port);
        return "重启指令发送成功";
    }

    /**
     * 重连
     */
    @GetMapping("reconnect")
    public String reconnect() throws Exception {
        tcpClient.reconnect();
        return "重启指令发送成功";
    }
}

http端口为我们开头yml配置的端口，默认9999，建议使用postman或apifox进行http调用

postman示例，http调用发送接口，向服务器发送消息
或直接浏览器输入
http://localhost:9999/send?message=hello

7. 接口列表

消息发送
http://localhost:10000/send?message=hello
连接服务器
http://localhost:10000/connect?ip=127.0.0.1&port=20000
重连
http://localhost:10000/reconnect?ip=127.0.0.1&port=20000
json消息发送示例
http://localhost:9999/send/json

Request URL:  http://localhost:9999/send/json
Request Method: POST
Request Headers:
{
   "Content-Type":"application/json"
}
Request Body:
{
   "test": "hello"
}