聊聊心跳机制及netty心跳实现

　　我们在使用netty的时候会使用一个参数，ChannelOption.SO_KEEPALIVE为true, 设置好了之后再Linux系统才会对keepalive生效，但是linux里边需要配置几个参数，tcp_keepalive_time, tcp_keepalive_invl, tcp_keepalive_probes，如果不配置的时候都会是默认值。

　　tcp_keepalive_time 即给一个TCP连接发送心跳包最后的时间间隔某一段时间后继续发送心跳包，允许空闲的时间，然后再次发送心跳包，默认时间为7200秒，即2个小时发一次心跳包。

tcp_keepalive_invl，发送存活探测时候未收到对方回执的时候，需要间隔一段时间继续发送。默认为75秒。

　　tcp_keepalive_probes，如果发了存活探测的时候没有收到对方的回执，那么需要继续发送探测的次数，此时默认值为9次，也就是未收到回执的时候需要发送9次。

　　再理一次，间隔tcp_keepalive_time之后发送心跳探测，如果未收到对方回执的时候，需要间隔tcp_keepalive_invl设置的时间继续发送，一共需要发送tcp_keepalive_probes的次数。　　

这个是Linux系统的配置，如果要使用Linux的此功能需要设置SO_KEEPALIVE为true，同时设置其他几个参数。系统默认的SO_KEEPALIVE为false。因为这些情况的差异，所以netty提供了自己实现心跳的机制。

　　netty有心跳的实现方法 IdleStateHandler，其中有读空闲时间，写空闲时间，读写空闲时间，只要有一个满足条件会触发userEventTriggered方法。

public IdleStateHandler(

            int readerIdleTimeSeconds,

            int writerIdleTimeSeconds,

            int allIdleTimeSeconds)

　　定义个消息内容吧，长度为Type的长度1 + 实际内容的长度5 = 6。Length为2个字节，Type为1个类型。

  +----------+----------+----------------+

  |  Length  |Type(byte)| Actual Content |

  |   0x06   |    1     |    "HELLO"     |

  +----------+----------+----------------+

　　定义公共的inbound方法，用于进行channelRead, sendPing, sendPong, userEventTriggered 方法。

package com.hqs.heartbeat.common;

import io.netty.buffer.ByteBuf;

import io.netty.channel.ChannelHandlerContext;

import io.netty.channel.SimpleChannelInboundHandler;

import io.netty.handler.timeout.IdleStateEvent;

import java.util.concurrent.atomic.AtomicInteger;

/**

 * @author huangqingshi

 * @Date 2019-05-11

 */

public abstract class CustomeHeartbeatHandler extends SimpleChannelInboundHandler<ByteBuf> {

    public static final byte PING = 1;

    public static final byte PONG = 2;

    public static final byte CUSTOM_MSG = 3;

    protected String name;

    private AtomicInteger heartbeatCount = new AtomicInteger(0);

    public CustomeHeartbeatHandler(String name) {

        this.name = name;

    }

    @Override

    protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {

        if(byteBuf.getByte(2) == PING) {

            sendPong(channelHandlerContext);

        } else if(byteBuf.getByte(2) == PONG) {

            System.out.println("get pong msg from " + channelHandlerContext

                    .channel().remoteAddress());

        } else {

            handleData(channelHandlerContext, byteBuf);

        }

    }

    protected abstract void handleData(ChannelHandlerContext channelHandlerContext,

                                       ByteBuf byteBuf);

    @Override

    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        System.out.println("channel read : " + msg);

        ByteBuf byteBuf = (ByteBuf) msg;

        System.out.println(byteBuf.getByte(2));

        super.channelRead(ctx, msg);

    }

    protected void sendPong(ChannelHandlerContext channelHandlerContext) {

        ByteBuf buf = channelHandlerContext.alloc().buffer(3);

        buf.writeShort(3);

        buf.writeByte(PONG);

        channelHandlerContext.writeAndFlush(buf);

        heartbeatCount.incrementAndGet();

        System.out.println("send pong message to " + channelHandlerContext.channel().remoteAddress());

    }

    protected void sendPing(ChannelHandlerContext channelHandlerContext) {

        ByteBuf buf = channelHandlerContext.alloc().buffer(3);

        buf.writeShort(3);

        buf.writeByte(PING);

        channelHandlerContext.writeAndFlush(buf);

        heartbeatCount.incrementAndGet();

        System.out.println("send ping message to " + channelHandlerContext.channel().remoteAddress());

    }

    @Override

    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {

        if(evt instanceof IdleStateEvent){

            IdleStateEvent e = (IdleStateEvent) evt;

            switch (e.state()) {

                case ALL_IDLE:

                    handlALLIdle(ctx);

                    break;

                case READER_IDLE:

                    handlReadIdle(ctx);

                    break;

                case WRITER_IDLE:

                    handlWriteIdle(ctx);

                    break;

                 default:

                     break;

            }

        }

    }

    protected void handlReadIdle(ChannelHandlerContext channelHandlerContext) {

        System.out.println("READ_IDLE---");

    }

    protected void handlWriteIdle(ChannelHandlerContext channelHandlerContext) {

        System.out.println("WRITE_IDLE---");

    }

    protected void handlALLIdle(ChannelHandlerContext channelHandlerContext) {

        System.out.println("ALL_IDLE---");

    }

    @Override

    public void channelActive(ChannelHandlerContext ctx) throws Exception {

        System.out.println("channel:" + ctx.channel().remoteAddress() + " is active");

    }

    @Override

    public void channelInactive(ChannelHandlerContext ctx) throws Exception {

        System.out.println("channel:" + ctx.channel().remoteAddress() + " is inactive");

    }

}

　　定义Server的方法，设置读超时为10秒，采用固定长度方法进行内容分割：LengthFieldBasedFrameDecoder(1024, 0, 2, -2, 0)，长度为1K 。一个主线程接收请求，四个线程处理请求。端口号设置为9999。

package com.hqs.heartbeat.server;

import io.netty.bootstrap.ServerBootstrap;

import io.netty.channel.Channel;

import io.netty.channel.ChannelInitializer;

import io.netty.channel.ChannelPipeline;

import io.netty.channel.nio.NioEventLoopGroup;

import io.netty.channel.socket.SocketChannel;

import io.netty.channel.socket.nio.NioServerSocketChannel;

import io.netty.handler.codec.LengthFieldBasedFrameDecoder;

import io.netty.handler.timeout.IdleStateHandler;

/**

 * @author huangqingshi

 * @Date 2019-05-11

 */

public class Server {

    public static void main(String[] args) {

        NioEventLoopGroup boss = new NioEventLoopGroup(1);

        NioEventLoopGroup worker = new NioEventLoopGroup(4);

        try {

            ServerBootstrap bootstrapServer = new ServerBootstrap();

            bootstrapServer.group(boss, worker).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {

                @Override

                protected void initChannel(SocketChannel ch) throws Exception {

                    ChannelPipeline channelPipeline = ch.pipeline();

                    channelPipeline.addLast(new IdleStateHandler(10, 0, 0));

                    channelPipeline.addLast(new LengthFieldBasedFrameDecoder(1024, 0,2, -2, 0));

                    channelPipeline.addLast(new ServerHandler());

                }

            });

            Channel channel = bootstrapServer.bind(9999).sync().channel();

            channel.closeFuture().sync();

        } catch (Exception e) {

            throw new RuntimeException(e);

        } finally {

            boss.shutdownGracefully();

            worker.shutdownGracefully();

        }

    }

}

　　Server的handler的处理方法：

package com.hqs.heartbeat.server;

import com.hqs.heartbeat.common.CustomeHeartbeatHandler;

import io.netty.buffer.ByteBuf;

import io.netty.buffer.Unpooled;

import io.netty.channel.ChannelHandlerContext;

/**

 * @author huangqingshi

 * @Date 2019-05-11

 */

public class ServerHandler extends CustomeHeartbeatHandler {

    public ServerHandler() {

        super("server");

    }

    @Override

    protected void handleData(ChannelHandlerContext channelHandlerContext,

                              ByteBuf byteBuf) {

        byte[] data = new byte[byteBuf.readableBytes() - 3];

        ByteBuf responseBuf = Unpooled.copiedBuffer(byteBuf);

        byteBuf.skipBytes(3);

        byteBuf.readBytes(data);

        String content = new String(data);

        System.out.println(name + " get content : " + content);

        channelHandlerContext.writeAndFlush(responseBuf);

    }

    @Override

    protected void handlReadIdle(ChannelHandlerContext channelHandlerContext) {

        super.handlReadIdle(channelHandlerContext);

        System.out.println(" client " + channelHandlerContext.channel().remoteAddress() + " reader timeout close it --");

        channelHandlerContext.close();

    }

}

　　定义Client类，所有超时时间为5秒，如果5秒没有读写的话则发送ping，如果失去连接之后inactive了就会重新连接，采用10秒出发一次。

package com.hqs.heartbeat.client;

import com.hqs.heartbeat.common.CustomeHeartbeatHandler;

import io.netty.bootstrap.Bootstrap;

import io.netty.buffer.ByteBuf;

import io.netty.channel.*;

import io.netty.channel.nio.NioEventLoopGroup;

import io.netty.channel.socket.SocketChannel;

import io.netty.channel.socket.nio.NioSocketChannel;

import io.netty.handler.codec.LengthFieldBasedFrameDecoder;

import io.netty.handler.timeout.IdleStateHandler;

import java.util.Random;

import java.util.concurrent.TimeUnit;

/**

 * @author huangqingshi

 * @Date 2019-05-11

 */

public class Client {

    private NioEventLoopGroup workGroup = new NioEventLoopGroup(4);

    private Channel channel;

    private Bootstrap bootstrap;

    public static void main(String[] args) throws InterruptedException {

        Client client = new Client();

        client.start();

        client.sendData();

    }

    public void start() {

        try {

            bootstrap = new Bootstrap();

            bootstrap.group(workGroup).channel(NioSocketChannel.class).option(ChannelOption.SO_KEEPALIVE, true)

                    .handler(new ChannelInitializer<SocketChannel>() {

                        @Override

                        protected void initChannel(SocketChannel ch) throws Exception {

                            ChannelPipeline channelPipeline = ch.pipeline()

                                    .addLast(new IdleStateHandler(0,0,5))

                                    .addLast(new LengthFieldBasedFrameDecoder(1024, 0, 2, -2, 0))

                                    .addLast(new ClientHandler(Client.this));

                        }

                    });

            doConnect();

        } catch (Exception e) {

            throw new RuntimeException(e);

        }

    }

    public void sendData() throws InterruptedException {

        Random random = new Random(System.currentTimeMillis());

        for(int i = 0; i < 10000; i++) {

            if(channel != null && channel.isActive()) {

                String content = "client msg " + i;

                ByteBuf byteBuf = channel.alloc().buffer(3 + content.getBytes().length);

                byteBuf.writeShort(3 + content.getBytes().length);

                byteBuf.writeByte(CustomeHeartbeatHandler.CUSTOM_MSG);

                byteBuf.writeBytes(content.getBytes());

                channel.writeAndFlush(byteBuf);

            }

            Thread.sleep(random.nextInt(20000));

        }

    }

    public void doConnect() {

        if(channel != null && channel.isActive()) {

            return;

        }

        ChannelFuture future = bootstrap

                .connect("127.0.0.1", 9999);

        future.addListener(new ChannelFutureListener() {

            @Override

            public void operationComplete(ChannelFuture future) throws Exception {

                if(future.isSuccess()) {

                    channel = future.channel();

                    System.out.println("connect to server successfully");

                } else {

                    System.out.println("Failed to connect to server, try after 10s");

                    future.channel().eventLoop().schedule(new Runnable() {

                        @Override

                        public void run() {

                            doConnect();

                        }

                    }, 10, TimeUnit.SECONDS);

                }

            }

        });

    }

}

　　定义clientHandler方法，读取时跳过长度+类型 2+1 三个字节，然后获取消息。连接断开之后则进行重连。

package com.hqs.heartbeat.client;

import com.hqs.heartbeat.common.CustomeHeartbeatHandler;

import io.netty.buffer.ByteBuf;

import io.netty.channel.ChannelHandlerContext;

/**

 * @author huangqingshi

 * @Date 2019-05-11

 */

public class ClientHandler extends CustomeHeartbeatHandler {

    private Client client;

    public ClientHandler(Client client) {

        super("client");

        this.client = client;

    }

    @Override

    protected void handleData(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) {

        byte[] data = new byte[byteBuf.readableBytes() - 3];

        byteBuf.skipBytes(3);

        byteBuf.readBytes(data);

        String content = new String(data);

        System.out.println(name + " get content:" + content);

    }

    @Override

    protected void handlALLIdle(ChannelHandlerContext channelHandlerContext) {

        super.handlALLIdle(channelHandlerContext);

        sendPing(channelHandlerContext);

    }

    @Override

    public void channelInactive(ChannelHandlerContext ctx) throws Exception {

        super.channelInactive(ctx);

        client.doConnect();

    }

}

　　好了，总体的netty心跳实现机制就这么多，希望能帮助到大家。

　　github地址：https://github.com/stonehqs/heartbeat

巴特西

聊聊心跳机制及netty心跳实现

最新文章

热门文章