长连接、短连接、心跳机制与断线重连

2018-06-12 548

概述

可承遇到，不知什么原因，一个夜晚，机房中，大片的远程调用连接断开。

第二天早上，用户访问高峰，大部分服务器都在获取连接，造成大片网络阻塞。

服务崩溃，惨不忍睹的景象。

本文将从长连接和短连接的概念切入，再到长连接与短连接的区别，以及应用场景，引出心跳机制和断线重连，给出代码实现。

从原理到实践杜绝此类现象。

短连接

概念

client与server通过三次握手建立连接，client发送请求消息，server返回响应，一次连接就完成了。

这时候双方任意都可以发起close操作，不过一般都是client先发起close操作。上述可知，短连接一般只会在 client/server间传递一次请求操作。

短连接的优缺点

管理起来比较简单，存在的连接都是有用的连接，不需要额外的控制手段。

使用场景

通常浏览器访问服务器的时候就是短连接。

对于服务端来说，长连接会耗费服务端的资源，而且用户用浏览器访问服务端相对而言不是很频繁的

如果有几十万，上百万的连接，服务端的压力会非常大，甚至会崩溃。

所以对于并发量大，请求频率低的，建议使用短连接。

长连接

什么是长连接

client向server发起连接，server接受client连接，双方建立连接。

Client与server完成一次读写之后，它们之间的连接并不会主动关闭，后续的读写操作会继续使用这个连接。

长连接的生命周期

正常情况下，一条TCP长连接建立后，只要双不提出关闭请求并且不出现异常情况，这条连接是一直存在的.

操作系统不会自动去关闭它，甚至经过物理网络拓扑的改变之后仍然可以使用。

所以一条连接保持几天、几个月、几年或者更长时间都有可能，只要不出现异常情况或由用户（应用层）主动关闭。

客户端和服务单可一直使用该连接进行数据通信。

长连接的优点

长连接可以省去较多的TCP建立和关闭的操作，减少网络阻塞的影响，

当发生错误时，可以在不关闭连接的情况下进行提示，

减少CPU及内存的使用，因为不需要经常的建立及关闭连接。

长连接的缺点

连接数过多时，影响服务端的性能和并发数量。

使用场景

数据库的连接就是采用TCP长连接.

RPC，远程服务调用，在服务器，一个服务进程频繁调用另一个服务进程，可使用长连接，减少连接花费的时间。

总结

1.对于长连接和短连接的使用是需要根据应用场景来判断的

2.长连接并不是万能的，也是需要维护的，

长连接的实现

心跳机制

应用层协议大多都有HeartBeat机制，通常是客户端每隔一小段时间向服务器发送一个数据包，通知服务器自己仍然在线。

并传输一些可能必要的数据。使用心跳包的典型协议是IM，比如QQ/MSN/飞信等协议。

在TCP的机制里面，本身是存在有心跳包的机制的，也就是TCP的选项：SO_KEEPALIVE。

系统默认是设置的2小时的心跳频率。但是它检查不到机器断电、网线拔出、防火墙这些断线。

而且逻辑层处理断线可能也不是那么好处理。一般，如果只是用于保活还是可以的。

为什么需要心跳机制？

因为网络的不可靠性, 有可能在 TCP 保持长连接的过程中, 由于某些突发情况, 例如网线被拔出, 突然掉电等,

会造成服务器和客户端的连接中断. 在这些突发情况下, 如果恰好服务器和客户端之间没有交互的话, 那么它们是不能在短时间内发现对方已经掉线的.

心跳机制即可解决此类问题。

TCP协议的KeepAlive机制

默认KeepAlive状态是不打开的。

需要将setsockopt将SOL_SOCKET.SO_KEEPALIVE设置为1才是打开KeepAlive状态，

并且可以设置三个参数：

tcp_keepalive_time ，tcp_keepalive_probes ， tcp_keepalive_intvl，

分别表示：连接闲置多久开始发keepalive的ack包、发几个ack包不回复才当对方已断线、两个ack包之间的间隔。

很多网络设备，尤其是NAT路由器，由于其硬件的限制（例如内存、CPU处理能力），无法保持其上的所有连接，因此在必要的时候，会在连接池中选择一些不活跃的连接踢掉。

典型做法是LRU，把最久没有数据的连接给T掉。

通过使用TCP的KeepAlive机制（修改那个time参数），可以让连接每隔一小段时间就产生一些ack包，以降低被踢掉的风险，当然，这样的代价是额外的网络和CPU负担。

如何实现心跳机制？

两种方式实现心跳机制:

使用 TCP 协议层面的 keepalive 机制.
在应用层上实现自定义的心跳机制.

虽然在 TCP 协议层面上, 提供了 keepalive 保活机制, 但是使用它有几个缺点:

它不是 TCP 的标准协议, 并且是默认关闭的.
TCP keepalive 机制依赖于操作系统的实现, 默认的 keepalive 心跳时间是 两个小时, 并且对 keepalive 的修改需要系统调用(或者修改系统配置), 灵活性不够.
TCP keepalive 与 TCP 协议绑定, 因此如果需要更换为 UDP 协议时, keepalive 机制就失效了.

使用 TCP 层面的 keepalive 机制比自定义的应用层心跳机制节省流量,

本文的主要介绍应用层方面实现心跳机制，使用netty实现心跳和断线重连。

netty实现心跳机制

netty对心跳机制提供了机制，实现的关键是IdleStateHandler先来看一下他的构造函数

    public IdleStateHandler(
            long readerIdleTime, long writerIdleTime, long allIdleTime,
            TimeUnit unit) {
        this(false, readerIdleTime, writerIdleTime, allIdleTime, unit);
    }

实例化一个 IdleStateHandler 需要提供三个参数:

readerIdleTimeSeconds, 读超时. 即当在指定的时间间隔内没有从 Channel 读取到数据时, 会触发一个 READER_IDLE 的 IdleStateEvent 事件.
writerIdleTimeSeconds, 写超时. 即当在指定的时间间隔内没有数据写入到 Channel 时, 会触发一个 WRITER_IDLE 的 IdleStateEvent 事件.
allIdleTimeSeconds, 读和写都超时. 即当在指定的时间间隔内没有读并且写操作时, 会触发一个 ALL_IDLE 的 IdleStateEvent 事件.

netty心跳流程

1. 客户端成功连接服务端。

2.在客户端中的ChannelPipeline中加入IdleStateHandler，设置写事件触发事件为5s.

3.客户端超过5s未写数据，触发写事件，向服务端发送心跳包，

4.同样，服务端要对心跳包做出响应，其实给客户端最好的回复就是“不回复”，减轻服务端的压力

5.超过三次，1过0s服务端都会收到来自客户端的心跳信息，服务端可以认为客户端挂了，可以close链路。

6.客户端恢复正常，发现链路已断，重新连接服务端。

代码实现

服务端handler:

package com.heartbreak.server;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;

import java.util.Random;

/**
 * @author janti
 * @date 2018/6/10 12:21
 */
public class HeartbeatServerHandler extends SimpleChannelInboundHandler<String> {
    // 失败计数器：未收到client端发送的ping请求
    private int unRecPingTimes = 0;

    // 定义服务端没有收到心跳消息的最大次数
    private static final int MAX_UN_REC_PING_TIMES = 3;

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

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        if (msg!=null && msg.equals("Heartbeat")){
            System.out.println("客户端"+ctx.channel().remoteAddress()+"--心跳信息--");
        }else {
            System.out.println("客户端----请求消息----："+msg);
            String resp = "商品的价格是："+random.nextInt(1000);
            ctx.writeAndFlush(resp);
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state()==IdleState.READER_IDLE){
                System.out.println("===服务端===(READER_IDLE 读超时)");
                // 失败计数器次数大于等于3次的时候，关闭链接，等待client重连
                if (unRecPingTimes >= MAX_UN_REC_PING_TIMES) {
                    System.out.println("===服务端===(读超时，关闭chanel)");
                    // 连续超过N次未收到client的ping消息，那么关闭该通道，等待client重连
                    ctx.close();
                } else {
                    // 失败计数器加1
                    unRecPingTimes++;
                }
            }else {
                super.userEventTriggered(ctx,evt);
            }
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        System.out.println("一个客户端已连接");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        System.out.println("一个客户端已断开连接");
    }
}

服务端server：

package com.heartbreak.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.concurrent.TimeUnit;

/**
 * @author tangj
 * @date 2018/6/10 10:46
 */
public class HeartBeatServer {
    private static int port = 9817;

    public HeartBeatServer(int port) {
        this.port = port;
    }

    ServerBootstrap bootstrap = null;
    ChannelFuture f;

    // 检测chanel是否接受过心跳数据时间间隔（单位秒）
    private static final int READ_WAIT_SECONDS = 10;

    public static void main(String args[]) {
        HeartBeatServer heartBeatServer = new HeartBeatServer(port);
        heartBeatServer.startServer();
    }

    public void startServer() {
        EventLoopGroup bossgroup = new NioEventLoopGroup();
        EventLoopGroup workergroup = new NioEventLoopGroup();
        try {
            bootstrap = new ServerBootstrap();
            bootstrap.group(bossgroup, workergroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new HeartBeatServerInitializer());
            // 服务器绑定端口监听
            f = bootstrap.bind(port).sync();
            System.out.println("server start ,port: "+port);
            // 监听服务器关闭监听，此方法会阻塞
            f.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            bossgroup.shutdownGracefully();
            workergroup.shutdownGracefully();
        }
    }


    private class HeartBeatServerInitializer extends ChannelInitializer<SocketChannel> {

        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            // 监听读操作,读超时时间为5秒，超过5秒关闭channel;
            pipeline.addLast("ping", new IdleStateHandler(READ_WAIT_SECONDS, 0, 0, TimeUnit.SECONDS));
            pipeline.addLast("decoder", new StringDecoder());
            pipeline.addLast("encoder", new StringEncoder());

            pipeline.addLast("handler", new HeartbeatServerHandler());
        }
    }

}

客户端handler

package com.heartbreak.client;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.EventLoop;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.CharsetUtil;
import io.netty.util.ReferenceCountUtil;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * @author tangj
 * @date 2018/6/11 22:55
 */
public class HeartBeatClientHandler extends SimpleChannelInboundHandler<String>{
    private HeartBeatClient client;

    private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:dd");

    private static final ByteBuf HEARTBEAT_SEQUENCE = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Heartbeat",
            CharsetUtil.UTF_8));

    public HeartBeatClientHandler(HeartBeatClient client) {
        this.client = client;
    }
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        System.out.println("收到服务端回复："+msg);
        if (msg.equals("Heartbeat")) {
            ctx.write("has read message from server");
            ctx.flush();
        }
        ReferenceCountUtil.release(msg);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.WRITER_IDLE) {
                ctx.writeAndFlush(HEARTBEAT_SEQUENCE.duplicate());
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        System.err.println("客户端与服务端断开连接,断开的时间为："+format.format(new Date()));
        // 定时线程 断线重连
        final EventLoop eventLoop = ctx.channel().eventLoop();
        eventLoop.schedule(new Runnable() {
            @Override
            public void run() {
                client.doConncet();
            }
        }, 10, TimeUnit.SECONDS);
    }


}

客户端启动:

package com.heartbreak.client;

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.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @author tangj
 * @date 2018/6/10 16:18
 */
public class HeartBeatClient {

    private Random random = new Random();
    public Channel channel;
    public Bootstrap bootstrap;

    protected String host = "127.0.0.1";
    protected int port = 9817;

    public static void main(String args[]) throws Exception {
        HeartBeatClient client = new HeartBeatClient();
        client.run();
        client.sendData();

    }

    public void run() throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new SimpleClientInitializer(HeartBeatClient.this));
            doConncet();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 发送数据
     * @throws Exception
     */
    public void sendData() throws Exception {
        BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
        while (true){
            String cmd = in.readLine();
            switch (cmd){
                case "close" :
                    channel.close();
                    break;
                default:
                channel.writeAndFlush(in.readLine());
                    break;
            }
        }
    }

    /**
     * 连接服务端
     */
    public void doConncet() {
        if (channel != null && channel.isActive()) {
            return;
        }
        ChannelFuture channelFuture = bootstrap.connect(host, port);
        channelFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture futureListener) throws Exception {
                if (channelFuture.isSuccess()) {
                    channel = futureListener.channel();
                    System.out.println("connect server successfully");
                } else {
                    System.out.println("Failed to connect to server, try connect after 10s");
                    futureListener.channel().eventLoop().schedule(new Runnable() {
                        @Override
                        public void run() {
                            doConncet();
                        }
                    }, 10, TimeUnit.SECONDS);
                }
            }
        });

    }


    private class SimpleClientInitializer extends ChannelInitializer<SocketChannel> {

        private HeartBeatClient client;

        public SimpleClientInitializer(HeartBeatClient client) {
            this.client = client;
        }

        @Override
        protected void initChannel(SocketChannel socketChannel) throws Exception {
            ChannelPipeline pipeline = socketChannel.pipeline();
            pipeline.addLast(new IdleStateHandler(0, 5, 0));
            pipeline.addLast("encoder", new StringEncoder());
            pipeline.addLast("decoder", new StringDecoder());
            pipeline.addLast("handler", new HeartBeatClientHandler(client));
        }
    }


}

运行结果：

1.客户端长时间未发送心跳包，服务端关闭连接

server start ,port: 9817
一个客户端已连接
===服务端===(READER_IDLE 读超时)
===服务端===(READER_IDLE 读超时)
===服务端===(READER_IDLE 读超时)
===服务端===(READER_IDLE 读超时)
===服务端===(读超时，关闭chanel)
一个客户端已断开连接

2.客户端发送心跳包，服务端和客户端保持心跳信息

一个客户端已连接
客户端/127.0.0.1:55436--心跳信息--
客户端/127.0.0.1:55436--心跳信息--
客户端/127.0.0.1:55436--心跳信息--
客户端/127.0.0.1:55436--心跳信息--

3.服务单宕机，断开连接，客户端进行重连

客户端与服务端断开连接,断开的时间为：2018-06-12 23:47:12
Failed to connect to server, try connect after 10s
Failed to connect to server, try connect after 10s
Failed to connect to server, try connect after 10s
connect server successfully

代码地址：

LearnTCP

本文首发于个人网站：http://www.janti.cn

参考：

浅析 Netty 实现心跳机制与断线重连

http://tldp.org/HOWTO/html_single/TCP-Keepalive-HOWTO

个人博客网站 http://www.janti.cn

微信关注我们

原文链接：https://yq.aliyun.com/articles/609265

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

HTTP 、Socket 、 TCP 七层OSI网络模型，这里只介绍五层常用网络模型，想知道全部七层详细内容自行查询。应用层：HTTP FTP SMTP DNS Telnet 传输层：TCP UDP 网络层：IP ICMP 数据链路层：ARP等物理层：1000BASE-SX等 socket是用来连接传输层和应用层，使得应用层可以直接和传输层做交互。 socket本身不属于网络协议，socket可以直接操控tcp，这样可以实现自己的应用层协议，例如聊天室就是，socket可以直接和tcp打交道，实现与http同级别的网络协议。 image.png image.png 上图左侧是server端，右侧是client端

2018-06-12

659

java基础巩固-浅析String源码及其不可变性

字符串可以说是广泛应用在日常编程中，jdk从1.0就提供了String类来创建和操作字符串。同时它也是不可改变类(基本类型的包装类都不可改变)的典型代表。源码查看(基于1.8) public final class String implements java.io.Serializable, Comparable<String>, CharSequence { private final char value[]; //这边只是引用并不是真正对象 ... } //首先string类创建的对象是不可变的(一个对象在创建完成后不能再改变它状态，说明是不可变的，并发程序最喜欢不可变量了)， //里面最主要的成员为char类型的数组几个构造方法 //空的构造方法例如 String a = new String(); a为""空字符 public String() { this.value = new char[0]; } //带参构造方法将源的hash和value赋给目标String public String(String original) { this.value ...

2018-06-13

627

发表评论

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。

长连接、短连接、心跳机制与断线重连

概述

短连接

长连接

长连接的实现

心跳机制

为什么需要心跳机制？

TCP协议的KeepAlive机制

如何实现心跳机制？

netty实现心跳机制

netty心跳流程

代码实现

TCP长连接与短连接、心跳机制

浅析 Netty 实现心跳机制与断线重连

Python socket 编程理解

java基础巩固-浅析String源码及其不可变性

相关文章

发表评论

资源下载

优质分享App

Mario

Nacos

Rocky Linux

欢迎您来访！

长连接 、短连接、心跳机制与断线重连

概述

短连接

长连接

长连接的实现

心跳机制

为什么需要心跳机制？

TCP协议的KeepAlive机制

如何实现心跳机制？

netty实现心跳机制

netty心跳流程

代码实现

TCP长连接与短连接、心跳机制

浅析 Netty 实现心跳机制与断线重连

Python socket 编程理解

java基础巩固-浅析String源码及其不可变性

相关文章

发表评论

资源下载

优质分享App

Mario

Nacos

Rocky Linux

欢迎您来访！

长连接、短连接、心跳机制与断线重连