作者 | 张超

责编 | 伍杏玲

出品 | CSDN（ID：CSDNnews）

最近我在做连接服务的结构调整，梳理到其中的一个功能点：同账号连续登录，旧的连接会被新的连接踢掉的功能。

第一个感觉这是个简单的需求点，不就是新的建立好了，然后把旧的连接踢掉？

可仔细想想就会发现一个问题：是在新的连接建立好了再踢掉旧的连接还是先踢掉旧的连接，然后再建立新的连接呢？如果存在多个同时进行的新连接呢，是不是就有数据竞争了？如果集群要是有多个机房，多个集群该如何高效的实现这个功能点？

如何既要考虑功能的实现，又要考虑工程实现的复杂程度以及后续的维护成本，是否满足对服务的整体服务质量的要求？在本文中，我和大家一起探讨下。

功能需求的描述

同一个账号（这里的同一个账号，指的是系统的唯一识别码，如果系统支持同一个账号的多端登录， 则指的式特定的某一个端）登录系统时，后登录的连接保留，而之前建立的连接将默认被踢掉（断开连接，只是断开的原因是踢掉）。踢掉旧的连接之后主要达到下面的两个效果：

1、新连接能够正常的进行操作。

一般的操作是新连接会覆盖旧连接，在这种情况下，不会影响新连接的所有工作。在通常的实现下，确实是这样的，覆盖之后肯定旧连接没法正常找到，所以发送给连接的数据能够正常到达新的连接上。

但有些系统为了保证连接的活跃性，采用的连接是定时进行刷新自己的连接信息。在这种情况下，就会出现新连接被覆盖的情况，具体关于连接的维护的后续的文章中在详细的介绍。

2、旧连接不能再发送或者接收任何的数据。

如果不踢掉的话，旧的连接至少能够进行数据上传，该数据在特定的场景下就会带来业务上的不一致性，甚至是错误；当然这种情况下收到数据也是有可能的，1中描述的实现策略是一种情况，我接触的发布订阅的模型就是新的订阅默认是不会覆盖旧的订阅，所以是两者共存。

实现方案

下面介绍几种解决方案，并进行优缺点比较。该文章中的数据均是基于Erlang版本，其他的编程语言除去第一种之外都是一样的。

一、利用可比较的程序标识实现：

%% 新创建连接的逻辑
open_session(SID, User, Server, Resource, Priority, Info) ->set_session(SID, User, Server, Resource, Priority, Info),check_for_sessions_to_replace(User, Server, Resource),JID = jid:make(User, Server, Resource),ejabberd_hooks:run(sm_register_connection_hook,JID#jid.lserver, [SID, JID, Info]).%% 设置新的连接信息,新的连接信息是可以存储在不同的数据库中,目前在ejabberd的实现中就
%% 已经支持了mnesia,redis,mysql,postgresql等数据库的实现
set_session(SID, User, Server, Resource, Priority, Info) ->LUser = jid:nodeprep(User),LServer = jid:nameprep(Server),LResource = jid:resourceprep(Resource),US = {LUser, LServer},USR = {LUser, LServer, LResource},set_session(#session{sid = SID, usr = USR, us = US,priority = Priority, info = Info}).-spec set_session(#session{}) -> ok | {error, any()}.
set_session(#session{us = {LUser, LServer}} = Session) ->Mod = get_sm_backend(LServer),case Mod:set_session(Session) ofok ->case use_cache(Mod, LServer) oftrue ->ets_cache:delete(?SM_CACHE, {LUser, LServer},cache_nodes(Mod, LServer));false ->okend;{error, _} = Err ->Errend.%% 检查已经存在的连接,如果检测到连接已经是离线的状态,就直接进行删除相应的连接信息
%% 连接是在线的状态的话,则保留最大的连接,其他的连接则踢掉(这里通过下面的replaced消息处理)
check_for_sessions_to_replace(User, Server, Resource) ->LUser = jid:nodeprep(User),LServer = jid:nameprep(Server),LResource = jid:resourceprep(Resource),check_existing_resources(LUser, LServer, LResource),check_max_sessions(LUser, LServer).-spec check_existing_resources(binary(), binary(), binary()) -> ok.
check_existing_resources(LUser, LServer, LResource) ->Mod = get_sm_backend(LServer),Ss = get_sessions(Mod, LUser, LServer, LResource),if Ss == [] -> ok;true ->SIDs = [SID || #session{sid = SID} <- Ss],MaxSID = lists:max(SIDs),lists:foreach(fun ({_, Pid} = S) when S /= MaxSID ->ejabberd_c2s:route(Pid, replaced);(_) -> okend,SIDs)end.%% 生成连接的唯一ID,这里加入了时间戳,而且是单调递增的数据
%% self()加入到sid中,能够保证sid的唯一性,pid在erlang中是能够保证唯一性的,具体的
%% 实现原理有兴趣可以自行查找
make_sid() ->{misc:unique_timestamp(), self()}.%% misc.erl
unique_timestamp() ->{MS, S, _} = erlang:timestamp(),{MS, S, erlgang:unique_integer([positive, monotonic]) rem 1000000}.

原理说明：

1. 设置新的连接信息是这段代码中的set_session()方法；

2. 查找已经存在的连接是代码中的get_sessions()，并没有列出完整的实现；

3. 踢掉旧的连接，通过上一步查到的所有连接，根据连接的大小关系，除了最大的连接保留之外，其他的所有连接都执行踢掉的逻辑。

关键点的说明：

1. 连接的唯一识别从单一的账号信息，添加了一个唯一识别码（即代码中的sid），有了这个识别标志之后，在写入新的连接数据同时，不会覆盖原有的旧连接，进而保证了后续的连接清理工作。

2. 在连接信息中添加了一个可进行比较的字段sid，有了这规则之后，无论怎样的踢掉操作都能够保证竞争的双方保留的结果是一致的。即使在竞争的情况下，也不会出现删除错误的情况。

3. 下面列举了一些可能的时序场景：

（1）正常的登录场景

（2）登录过程在setSession之后之后被其他的登录过程打断：

（3）新的登录过程插入到另一个登录过程中：

（4）虽然业务是A是先执行setSession但是,但是写入的sid却比后写入的sid要大，比如A在写入时网络延时很大等情况都有可能出现如下的情况：

二、利用独立的锁结构实现

%% 1. 获取新键连接的锁,获取到相应的锁之后
%% 2. 踢掉之前已经存在的旧连接
%% 3. 初始化新的连接
open_session(true, ClientInfo = #{clientid := ClientId}, ConnInfo) ->CleanStart = fun(_) ->ok = discard_session(ClientId),%% 具体的初始化新连接的具体逻辑Session = emqx_session:init(ClientInfo, ConnInfo),{ok, #{session => Session, present => false}}end,%% 这里没有列举获取锁的相关过程,处理完所有的逻辑之后,释放获取的锁emqx_cm_locker:trans(ClientId, CleanStart);%% 根据连接的ClientID找到具体的ChanPid,然后循环执行discard_session
discard_session(ClientId) when is_binary(ClientId) ->case lookup_channels(ClientId) of[] -> ok;ChanPids ->lists:foreach(fun(ChanPid) ->trydiscard_session(ClientId, ChanPid)catch_:Error:_Stk ->?LOG(error, "Failed to discard ~p: ~p", [ChanPid, Error])endend, ChanPids)end.%% 具体的操作踢掉的逻辑
discard_session(ClientId, ChanPid) when node(ChanPid) == node() ->case get_chan_attrs(ClientId, ChanPid) of#{conninfo := #{conn_mod := ConnMod}} ->ConnMod:call(ChanPid, discard);undefined -> okend;

原理说明如下：

1. 根据连接的标识（ClientID）来获取对应的锁，该锁是排他的，emqx的代码中的锁是程序实现通过集群中广播的方式。

在集群中维护着对应锁，对应的锁可以借助其他的数据库，比如Redis等方式实现同样的锁的功能。不过要注意锁的释放，否则将会影响正常的用户登录。

2. 查找已经存在的连接，踢掉相应的连接。

3. 初始化新创建的连接。

讨论方案的对比

以上两种实现是现在使用相对较多的长连接服务的登录，并且踢掉旧连接的逻辑。

1. 功能实现：

两种方案都能完整地实现新连接踢掉旧连接的过程。

2. 实现的复杂度：

Ejabberd的实现通过添加一个可比较的sid的方式，保留最大sid的连接。在存在竞争的情况下，依然能保证逻辑的正确性；EMQX采用的是比较常见的加锁的方式来避免竞争带来的数据不一致性问题。

总的来说，实现的复杂度Ejabberd相对复杂一些，这里不包括EMQX实现的锁功能的复杂度。

3. 理解的难易程度：

Ejabberd的理解难度相对较大，因为要考虑多种情况下的服务表现是否符合预期；EMQX通过直接加锁的方式来避免了这些竟态的出现。Ejabberd的理解难度更大。

4. 数据存储方面：

Ejabberd的设计是同一个用户的连接信息在数据库是有一组的（不同的sid），而EMQX的存储只会有一个（其中 ClientID 为唯一性的key）。

从这方面来看，EMQX在存储上会有一定的优势，在没有考虑对锁的存储需求下。

5. 集群扩展性：

Ejabberd由于做到了无状态的新创建连接，能进行很好地扩展。EMQX由于实现是在集群中进行全局锁的情况，机器的扩展会加大获取锁的成本，虽然采用并发调用的逻辑，但是在网络等不稳定的条件下，可能会出现获取锁比较慢，或者失败的情况，所以整体的可扩展性会较差一些。

多机房、多集群的部署方案

在多机房、多集群的部署形态下，由于连接的信息都需要进行跨机房的访问，会存在很大的延时和挑战。

然而多机房的部署主要是提供多机房的活跃互相备份，以保证在其中任何一个机房出现问题也不影响整体的服务。

所以如果是在多机房的模式下，建议通过负载的方式来将同一个账号连接到一个机房/集群，这样问题就重新降到解决一个集群中的踢掉问题。

结语

登录连接这个看似简单、但又不简单的需求，其中涉及到对服务的预期和要求。我们可采用结构简单的加锁方式来实现，也可以通过在业务层通过自己的选择策略来进行踢出连接的操作。

选择策略时要保证分布式情况下保持一致（切忌将除了自己之外的其他连接都踢掉，这样就会出现多个连接的选择策略不同）。实现功能的方式可以是多种选择的，关键看具体的需求是什么样，选择一种适合团队的方式才是最重要的。

传送门：

EMQX源码：https://github.com/emqx/emqx

Ejabberd源码：https://github.com/processone/ejabberd

作者简介：张超，360 IoT 云连接服务技术负责人，毕业于南京大学。曾从事过游戏开发、IM 服务端开发，目前从事物联网接入层相关工作。

【End】

热 文 推 荐 

☞抗住 60 亿次攻击，起底阿里云安全的演进之路 | 问底中国 IT 技术演进

☞罗永浩回应被“Sharklet 科技解约”；12306 已屏蔽多个抢票软件；FreeDB 将关闭 | 极客头条

微浏览器无处不在

☞雷军回家

☞小米创始人离职，背后这 3 点值得深思

☞迁移学习前沿研究亟需新鲜血液，深度学习理论不能掉链子

☞漫画：互联网公司会议观察

☞俄罗斯“扎克伯格”：创建区块链版“微信” 27岁身价已达2.5亿美元

点击阅读原文，即刻参加！

你点的每个“在看”，我都认真当成了喜欢