作者 | 張超

責(zé)編 | 伍杏玲

出品 | CSDN（ID：CSDNnews）

最近我在做連接服務(wù)的結(jié)構(gòu)調(diào)整，梳理到其中的一個(gè)功能點(diǎn)：同賬號(hào)連續(xù)登錄，舊的連接會(huì)被新的連接踢掉的功能。

第一個(gè)感覺這是個(gè)簡(jiǎn)單的需求點(diǎn)，不就是新的建立好了，然后把舊的連接踢掉？

可仔細(xì)想想就會(huì)發(fā)現(xiàn)一個(gè)問題：是在新的連接建立好了再踢掉舊的連接還是先踢掉舊的連接，然后再建立新的連接呢？如果存在多個(gè)同時(shí)進(jìn)行的新連接呢，是不是就有數(shù)據(jù)競(jìng)爭(zhēng)了？如果集群要是有多個(gè)機(jī)房，多個(gè)集群該如何高效的實(shí)現(xiàn)這個(gè)功能點(diǎn)？

如何既要考慮功能的實(shí)現(xiàn)，又要考慮工程實(shí)現(xiàn)的復(fù)雜程度以及后續(xù)的維護(hù)成本，是否滿足對(duì)服務(wù)的整體服務(wù)質(zhì)量的要求？在本文中，我和大家一起探討下。

功能需求的描述

1、新連接能夠正常的進(jìn)行操作。

一般的操作是新連接會(huì)覆蓋舊連接，在這種情況下，不會(huì)影響新連接的所有工作。在通常的實(shí)現(xiàn)下，確實(shí)是這樣的，覆蓋之后肯定舊連接沒法正常找到，所以發(fā)送給連接的數(shù)據(jù)能夠正常到達(dá)新的連接上。

但有些系統(tǒng)為了保證連接的活躍性，采用的連接是定時(shí)進(jìn)行刷新自己的連接信息。在這種情況下，就會(huì)出現(xiàn)新連接被覆蓋的情況，具體關(guān)于連接的維護(hù)的后續(xù)的文章中在詳細(xì)的介紹。

2、舊連接不能再發(fā)送或者接收任何的數(shù)據(jù)。

實(shí)現(xiàn)方案

下面介紹幾種解決方案，并進(jìn)行優(yōu)缺點(diǎn)比較。該文章中的數(shù)據(jù)均是基于Erlang版本，其他的編程語言除去第一種之外都是一樣的。

一、利用可比較的程序標(biāo)識(shí)實(shí)現(xiàn)：

%% 新創(chuàng)建連接的邏輯

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]).

%% 設(shè)置新的連接信息,新的連接信息是可以存儲(chǔ)在不同的數(shù)據(jù)庫(kù)中,目前在ejabberd的實(shí)現(xiàn)中就

%% 已經(jīng)支持了mnesia,redis,mysql,postgresql等數(shù)據(jù)庫(kù)的實(shí)現(xiàn)

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),

caseMod:set_session(Session) of

ok ->

caseuse_cache(Mod, LServer) of

true->

ets_cache:delete(?SM_CACHE, {LUser, LServer},

cache_nodes(Mod, LServer));

false->

ok

end;

{ error, _} = Err->

Err

end.

%% 檢查已經(jīng)存在的連接,如果檢測(cè)到連接已經(jīng)是離線的狀態(tài),就直接進(jìn)行刪除相應(yīng)的連接信息

%% 連接是在線的狀態(tài)的話,則保留最大的連接,其他的連接則踢掉(這里通過下面的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),

ifSs == [] -> 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);

(_) -> ok

end,

SIDs)

end.

%% 生成連接的唯一ID,這里加入了時(shí)間戳,而且是單調(diào)遞增的數(shù)據(jù)

%% self加入到sid中,能夠保證sid的唯一性,pid在erlang中是能夠保證唯一性的,具體的

%% 實(shí)現(xiàn)原理有興趣可以自行查找

make_sid ->

{misc:unique_timestamp, self}.

%% misc.erl

unique_timestamp ->

{MS, S, _} = erlang:timestamp,

{MS, S, erlgang:unique_integer([positive, monotonic]) rem1000000}.

原理說明：

1. 設(shè)置新的連接信息是這段代碼中的set_session方法；

2. 查找已經(jīng)存在的連接是代碼中的get_sessions，并沒有列出完整的實(shí)現(xiàn)；

3. 踢掉舊的連接，通過上一步查到的所有連接，根據(jù)連接的大小關(guān)系，除了最大的連接保留之外，其他的所有連接都執(zhí)行踢掉的邏輯。

關(guān)鍵點(diǎn)的說明：

2. 在連接信息中添加了一個(gè)可進(jìn)行比較的字段sid，有了這規(guī)則之后，無論怎樣的踢掉操作都能夠保證競(jìng)爭(zhēng)的雙方保留的結(jié)果是一致的。即使在競(jìng)爭(zhēng)的情況下，也不會(huì)出現(xiàn)刪除錯(cuò)誤的情況。

3. 下面列舉了一些可能的時(shí)序場(chǎng)景：

（1）正常的登錄場(chǎng)景

（2）登錄過程在setSession之后之后被其他的登錄過程打斷：

（3）新的登錄過程插入到另一個(gè)登錄過程中：

（4）雖然業(yè)務(wù)是A是先執(zhí)行setSession但是,但是寫入的sid卻比后寫入的sid要大，比如A在寫入時(shí)網(wǎng)絡(luò)延時(shí)很大等情況都有可能出現(xiàn)如下的情況：

二、利用獨(dú)立的鎖結(jié)構(gòu)實(shí)現(xiàn)

%% 1.獲取新鍵連接的鎖,獲取到相應(yīng)的鎖之后

%% 2.踢掉之前已經(jīng)存在的舊連接

%% 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,

%% 這里沒有列舉獲取鎖的相關(guān)過程,處理完所有的邏輯之后,釋放獲取的鎖

emqx_cm_locker:trans(ClientId, CleanStart);

%% 根據(jù)連接的ClientID找到具體的ChanPid,然后循環(huán)執(zhí)行discard_session

discard_session(ClientId) whenis_binary(ClientId) ->

case lookup_channels(ClientId) of

[] -> ok;

ChanPids ->

lists:foreach(

fun(ChanPid)->

try

discard_session(ClientId, ChanPid)

catch

_:Error:_Stk ->

?LOG(error, "Failed to discard ~p: ~p", [ChanPid, Error])

end

end, ChanPids)

end.

%% 具體的操作踢掉的邏輯

discard_session(ClientId, ChanPid) whennode(ChanPid) == node ->

case get_chan_attrs(ClientId, ChanPid) of

#{conninfo := #{conn_mod := ConnMod}} ->

ConnMod:call(ChanPid, discard);

undefined -> ok

end;

原理說明如下：

1. 根據(jù)連接的標(biāo)識(shí)（ClientID）來獲取對(duì)應(yīng)的鎖，該鎖是排他的，emqx的代碼中的鎖是程序?qū)崿F(xiàn)通過集群中廣播的方式。

在集群中維護(hù)著對(duì)應(yīng)鎖，對(duì)應(yīng)的鎖可以借助其他的數(shù)據(jù)庫(kù)，比如Redis等方式實(shí)現(xiàn)同樣的鎖的功能。不過要注意鎖的釋放，否則將會(huì)影響正常的用戶登錄。

2. 查找已經(jīng)存在的連接，踢掉相應(yīng)的連接。

3. 初始化新創(chuàng)建的連接。

討論方案的對(duì)比

以上兩種實(shí)現(xiàn)是現(xiàn)在使用相對(duì)較多的長(zhǎng)連接服務(wù)的登錄，并且踢掉舊連接的邏輯。

1. 功能實(shí)現(xiàn)：

兩種方案都能完整地實(shí)現(xiàn)新連接踢掉舊連接的過程。

2. 實(shí)現(xiàn)的復(fù)雜度：

Ejabberd的實(shí)現(xiàn)通過添加一個(gè)可比較的sid的方式，保留最大sid的連接。在存在競(jìng)爭(zhēng)的情況下，依然能保證邏輯的正確性；EMQX采用的是比較常見的加鎖的方式來避免競(jìng)爭(zhēng)帶來的數(shù)據(jù)不一致性問題。

總的來說，實(shí)現(xiàn)的復(fù)雜度Ejabberd相對(duì)復(fù)雜一些，這里不包括EMQX實(shí)現(xiàn)的鎖功能的復(fù)雜度。

3. 理解的難易程度：

Ejabberd的理解難度相對(duì)較大，因?yàn)橐紤]多種情況下的服務(wù)表現(xiàn)是否符合預(yù)期；EMQX通過直接加鎖的方式來避免了這些竟態(tài)的出現(xiàn)。Ejabberd的理解難度更大。

4. 數(shù)據(jù)存儲(chǔ)方面：

Ejabberd的設(shè)計(jì)是同一個(gè)用戶的連接信息在數(shù)據(jù)庫(kù)是有一組的（不同的sid），而EMQX的存儲(chǔ)只會(huì)有一個(gè)（其中 ClientID 為唯一性的key）。

從這方面來看，EMQX在存儲(chǔ)上會(huì)有一定的優(yōu)勢(shì)，在沒有考慮對(duì)鎖的存儲(chǔ)需求下。

5. 集群擴(kuò)展性：

Ejabberd由于做到了無狀態(tài)的新創(chuàng)建連接，能進(jìn)行很好地?cái)U(kuò)展。EMQX由于實(shí)現(xiàn)是在集群中進(jìn)行全局鎖的情況，機(jī)器的擴(kuò)展會(huì)加大獲取鎖的成本，雖然采用并發(fā)調(diào)用的邏輯，但是在網(wǎng)絡(luò)等不穩(wěn)定的條件下，可能會(huì)出現(xiàn)獲取鎖比較慢，或者失敗的情況，所以整體的可擴(kuò)展性會(huì)較差一些。

多機(jī)房、多集群的部署方案

在多機(jī)房、多集群的部署形態(tài)下，由于連接的信息都需要進(jìn)行跨機(jī)房的訪問，會(huì)存在很大的延時(shí)和挑戰(zhàn)。

然而多機(jī)房的部署主要是提供多機(jī)房的活躍互相備份，以保證在其中任何一個(gè)機(jī)房出現(xiàn)問題也不影響整體的服務(wù)。

所以如果是在多機(jī)房的模式下，建議通過負(fù)載的方式來將同一個(gè)賬號(hào)連接到一個(gè)機(jī)房/集群，這樣問題就重新降到解決一個(gè)集群中的踢掉問題。

結(jié)語

登錄連接這個(gè)看似簡(jiǎn)單、但又不簡(jiǎn)單的需求，其中涉及到對(duì)服務(wù)的預(yù)期和要求。我們可采用結(jié)構(gòu)簡(jiǎn)單的加鎖方式來實(shí)現(xiàn)，也可以通過在業(yè)務(wù)層通過自己的選擇策略來進(jìn)行踢出連接的操作。

選擇策略時(shí)要保證分布式情況下保持一致（切忌將除了自己之外的其他連接都踢掉，這樣就會(huì)出現(xiàn)多個(gè)連接的選擇策略不同）。實(shí)現(xiàn)功能的方式可以是多種選擇的，關(guān)鍵看具體的需求是什么樣，選擇一種適合團(tuán)隊(duì)的方式才是最重要的。

傳送門：

EMQX源碼：https://github/emqx/emqx

Ejabberd源碼：https://github/processone/ejabberd

作者簡(jiǎn)介：張超，360 IoT 云連接服務(wù)技術(shù)負(fù)責(zé)人，畢業(yè)于南京大學(xué)。曾從事過游戲開發(fā)、IM 服務(wù)端開發(fā)，目前從事物聯(lián)網(wǎng)接入層相關(guān)工作。