eureka是一種去中心化的服務(wù)治理應(yīng)用，其顯著特點(diǎn)是既可以作為服務(wù)端又可以作為服務(wù)向自己配置的地址進(jìn)行注冊(cè)。那么這篇文章就來探討一下eureka的注冊(cè)流程。
一、eureka的服務(wù)端
eureka的服務(wù)端核心類是eurekabootstrap，該類實(shí)現(xiàn)了一個(gè)servletcontextlistener的監(jiān)聽器。因此我們可以斷定eureka是基于servlet容器實(shí)現(xiàn)的。關(guān)鍵代碼如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
public class eurekabootstrap implements servletcontextlistener { //...省略相關(guān)代碼/** * initializes eureka, including syncing up with other eureka peers and publishing the registry. * * @see * javax.servlet.servletcontextlistener#contextinitialized(javax.servlet.servletcontextevent) */ @override public void contextinitialized(servletcontextevent event) { try { initeurekaenvironment(); initeurekaservercontext(); servletcontext sc = event.getservletcontext(); sc.setattribute(eurekaservercontext.class.getname(), servercontext); } catch (throwable e) { logger.error("cannot bootstrap eureka server :", e); throw new runtimeexception("cannot bootstrap eureka server :", e); } } //省略相關(guān)代碼.....} |
我們可以看到在servletcontext初始化完成時(shí)，會(huì)初始化eureka環(huán)境，然后初始化eurekaservercontext，那么我們?cè)诳匆豢磇niteurekaservercontext方法：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
|
/** * init hook for server context. override for custom logic. */ protected void initeurekaservercontext() throws exception { // ..... applicationinfomanager applicationinfomanager = null; if (eurekaclient == null) { eurekainstanceconfig instanceconfig = iscloud(configurationmanager.getdeploymentcontext()) ? new cloudinstanceconfig() : new mydatacenterinstanceconfig(); applicationinfomanager = new applicationinfomanager( instanceconfig, new eurekaconfigbasedinstanceinfoprovider(instanceconfig).get()); eurekaclientconfig eurekaclientconfig = new defaulteurekaclientconfig(); eurekaclient = new discoveryclient(applicationinfomanager, eurekaclientconfig); } else { applicationinfomanager = eurekaclient.getapplicationinfomanager(); } peerawareinstanceregistry registry; if (isaws(applicationinfomanager.getinfo())) { registry = new awsinstanceregistry( eurekaserverconfig, eurekaclient.geteurekaclientconfig(), servercodecs, eurekaclient ); awsbinder = new awsbinderdelegate(eurekaserverconfig, eurekaclient.geteurekaclientconfig(), registry, applicationinfomanager); awsbinder.start(); } else { registry = new peerawareinstanceregistryimpl( eurekaserverconfig, eurekaclient.geteurekaclientconfig(), servercodecs, eurekaclient ); } //....省略部分代碼 } |
在這個(gè)方法里會(huì)創(chuàng)建許多與eureka服務(wù)相關(guān)的對(duì)象，在這里我列舉了兩個(gè)核心對(duì)象分別是eurekaclient與peerawareinstanceregistry，關(guān)于客戶端部分我們等會(huì)再說，我們現(xiàn)在來看看peerawareinstanceregistry到底是做什么用的，這里我寫貼出關(guān)于這個(gè)類的類圖：
根據(jù)類圖我們可以清晰的發(fā)現(xiàn)peerawareinstanceregistry的最頂層接口為leasemanager與lookupservice,其中l(wèi)ookupservice定義了最基本的發(fā)現(xiàn)示例的行為而leasemanager定義了處理客戶端注冊(cè)，續(xù)約，注銷等操作。那么在這篇文章我們還是重點(diǎn)關(guān)注一下leasemanager的相關(guān)接口的實(shí)現(xiàn)。回過頭來我們?cè)诳磒eerawareinstanceregistry，其實(shí)這個(gè)類用于多個(gè)節(jié)點(diǎn)下復(fù)制相關(guān)信息，比如說一個(gè)節(jié)點(diǎn)注冊(cè)續(xù)約與下線那么通過這個(gè)類將會(huì)相關(guān)復(fù)制(通知)到各個(gè)節(jié)點(diǎn)。我們來看看它是怎么處理客戶端注冊(cè)的：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
/** * registers the information about the {@link instanceinfo} and replicates * this information to all peer eureka nodes. if this is replication event * from other replica nodes then it is not replicated. * * @param info * the {@link instanceinfo} to be registered and replicated. * @param isreplication * true if this is a replication event from other replica nodes, * false otherwise. */@overridepublic void register(final instanceinfo info, final boolean isreplication) { int leaseduration = lease.default_duration_in_secs; if (info.getleaseinfo() != null && info.getleaseinfo().getdurationinsecs() > 0) { leaseduration = info.getleaseinfo().getdurationinsecs(); } super.register(info, leaseduration, isreplication); replicatetopeers(action.register, info.getappname(), info.getid(), info, null, isreplication);} |
我們可以看到它調(diào)用了父類的register方法后又通過replicatetopeers復(fù)制對(duì)應(yīng)的行為到其他節(jié)點(diǎn)，具體如何復(fù)制的先不在這里討論，我們重點(diǎn)來看看注冊(cè)方法，我們?cè)诟割惱镎业絩egister()方法：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
|
/** * registers a new instance with a given duration. * * @see com.netflix.eureka.lease.leasemanager#register(java.lang.object, int, boolean) */ public void register(instanceinfo registrant, int leaseduration, boolean isreplication) {try { read.lock(); map<string, lease<instanceinfo>> gmap = registry.get(registrant.getappname()); register.increment(isreplication); if (gmap == null) { final concurrenthashmap<string, lease<instanceinfo>> gnewmap = new concurrenthashmap<string, lease<instanceinfo>>(); gmap = registry.putifabsent(registrant.getappname(), gnewmap); if (gmap == null) { gmap = gnewmap; } } lease<instanceinfo> existinglease = gmap.get(registrant.getid()); // retain the last dirty timestamp without overwriting it, if there is already a lease if (existinglease != null && (existinglease.getholder() != null)) { long existinglastdirtytimestamp = existinglease.getholder().getlastdirtytimestamp(); long registrationlastdirtytimestamp = registrant.getlastdirtytimestamp(); logger.debug("existing lease found (existing={}, provided={}", existinglastdirtytimestamp, registrationlastdirtytimestamp); // this is a > instead of a >= because if the timestamps are equal, we still take the remote transmitted // instanceinfo instead of the server local copy. if (existinglastdirtytimestamp > registrationlastdirtytimestamp) { logger.warn("there is an existing lease and the existing lease's dirty timestamp {} is greater" + " than the one that is being registered {}", existinglastdirtytimestamp, registrationlastdirtytimestamp); logger.warn("using the existing instanceinfo instead of the new instanceinfo as the registrant"); registrant = existinglease.getholder(); } } else { // the lease does not exist and hence it is a new registration synchronized (lock) { if (this.expectednumberofrenewspermin > 0) { // since the client wants to cancel it, reduce the threshold // (1 // for 30 seconds, 2 for a minute) this.expectednumberofrenewspermin = this.expectednumberofrenewspermin + 2; this.numberofrenewsperminthreshold = (int) (this.expectednumberofrenewspermin * serverconfig.getrenewalpercentthreshold()); } } logger.debug("no previous lease information found; it is new registration"); } lease<instanceinfo> lease = new lease<instanceinfo>(registrant, leaseduration); if (existinglease != null) { lease.setserviceuptimestamp(existinglease.getserviceuptimestamp()); } gmap.put(registrant.getid(), lease); //。。。省略部分代碼} |
通過源代碼，我們來簡要梳理一下流程：
1）首先根據(jù)appname獲取一些列的服務(wù)實(shí)例對(duì)象，如果為null,則新創(chuàng)建一個(gè)map并把當(dāng)前的注冊(cè)應(yīng)用程序信息添加到此map當(dāng)中，這里有一個(gè)lease對(duì)象，這個(gè)類描述了泛型t的時(shí)間屬性，比如說注冊(cè)時(shí)間，服務(wù)啟動(dòng)時(shí)間，最后更新時(shí)間等，大家可以關(guān)注一下它的實(shí)現(xiàn)：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
|
/* * copyright 2012 netflix, inc. * * licensed under the apache license, version 2.0 (the "license"); * you may not use this file except in compliance with the license. * you may obtain a copy of the license at * * http://www.apache.org/licenses/license-2.0 * * unless required by applicable law or agreed to in writing, software * distributed under the license is distributed on an "as is" basis, * without warranties or conditions of any kind, either express or implied. * see the license for the specific language governing permissions and * limitations under the license. */package com.netflix.eureka.lease;import com.netflix.eureka.registry.abstractinstanceregistry;/** * describes a time-based availability of a {@link t}. purpose is to avoid * accumulation of instances in {@link abstractinstanceregistry} as result of ungraceful * shutdowns that is not uncommon in aws environments. * * if a lease elapses without renewals, it will eventually expire consequently * marking the associated {@link t} for immediate eviction - this is similar to * an explicit cancellation except that there is no communication between the * {@link t} and {@link leasemanager}. * * @author karthik ranganathan, greg kim */public class lease<t> { enum action { register, cancel, renew }; public static final int default_duration_in_secs = 90; private t holder; private long evictiontimestamp; private long registrationtimestamp; private long serviceuptimestamp; // make it volatile so that the expiration task would see this quicker private volatile long lastupdatetimestamp; private long duration; public lease(t r, int durationinsecs) { holder = r; registrationtimestamp = system.currenttimemillis(); lastupdatetimestamp = registrationtimestamp; duration = (durationinsecs * 1000); } /** * renew the lease, use renewal duration if it was specified by the * associated {@link t} during registration, otherwise default duration is * {@link #default_duration_in_secs}. */ public void renew() { lastupdatetimestamp = system.currenttimemillis() + duration; } /** * cancels the lease by updating the eviction time. */ public void cancel() { if (evictiontimestamp <= 0) { evictiontimestamp = system.currenttimemillis(); } } /** * mark the service as up. this will only take affect the first time called, * subsequent calls will be ignored. */ public void serviceup() { if (serviceuptimestamp == 0) { serviceuptimestamp = system.currenttimemillis(); } } /** * set the leases service up timestamp. */ public void setserviceuptimestamp(long serviceuptimestamp) { this.serviceuptimestamp = serviceuptimestamp; } /** * checks if the lease of a given {@link com.netflix.appinfo.instanceinfo} has expired or not. */ public boolean isexpired() { return isexpired(0l); } /** * checks if the lease of a given {@link com.netflix.appinfo.instanceinfo} has expired or not. * * note that due to renew() doing the 'wrong" thing and setting lastupdatetimestamp to +duration more than * what it should be, the expiry will actually be 2 * duration. this is a minor bug and should only affect * instances that ungracefully shutdown. due to possible wide ranging impact to existing usage, this will * not be fixed. * * @param additionalleasems any additional lease time to add to the lease evaluation in ms. */ public boolean isexpired(long additionalleasems) { return (evictiontimestamp > 0 || system.currenttimemillis() > (lastupdatetimestamp + duration + additionalleasems)); } /** * gets the milliseconds since epoch when the lease was registered. * * @return the milliseconds since epoch when the lease was registered. */ public long getregistrationtimestamp() { return registrationtimestamp; } /** * gets the milliseconds since epoch when the lease was last renewed. * note that the value returned here is actually not the last lease renewal time but the renewal + duration. * * @return the milliseconds since epoch when the lease was last renewed. */ public long getlastrenewaltimestamp() { return lastupdatetimestamp; } /** * gets the milliseconds since epoch when the lease was evicted. * * @return the milliseconds since epoch when the lease was evicted. */ public long getevictiontimestamp() { return evictiontimestamp; } /** * gets the milliseconds since epoch when the service for the lease was marked as up. * * @return the milliseconds since epoch when the service for the lease was marked as up. */ public long getserviceuptimestamp() { return serviceuptimestamp; } /** * returns the holder of the lease. */ public t getholder() { return holder; }} |
2）根據(jù)當(dāng)前注冊(cè)的id，如果能在map中取到則做以下操作：
2.1）根據(jù)當(dāng)前存在節(jié)點(diǎn)的觸碰時(shí)間和注冊(cè)節(jié)點(diǎn)的觸碰時(shí)間比較，如果前者的時(shí)間晚于后者的時(shí)間，那么當(dāng)前注冊(cè)的實(shí)例就以已存在的實(shí)例為準(zhǔn)
2.2）否則更新其每分鐘期望的續(xù)約數(shù)量及其閾值
3）將當(dāng)前的注冊(cè)節(jié)點(diǎn)存到map當(dāng)中，至此我們的注冊(cè)過程基本告一段落了
二、eureka客戶端
在服務(wù)端servletcontext初始化完畢時(shí)，會(huì)創(chuàng)建discoveryclient。熟悉eureka的朋友，一定熟悉這兩個(gè)屬性：fetchregistry與registerwitheureka。在springcloud中集成eureka獨(dú)立模式運(yùn)行時(shí)，如果這兩個(gè)值不為false，那么啟動(dòng)會(huì)報(bào)錯(cuò)，為什么會(huì)報(bào)錯(cuò)呢？其實(shí)答案就在discoveryclient的構(gòu)造函數(shù)中：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
|
@inject discoveryclient(applicationinfomanager applicationinfomanager, eurekaclientconfig config, abstractdiscoveryclientoptionalargs args, provider<backupregistry> backupregistryprovider) {//....省略部分代碼 if (!config.shouldregisterwitheureka() && !config.shouldfetchregistry()) { logger.info("client configured to neither register nor query for data."); scheduler = null; heartbeatexecutor = null; cacherefreshexecutor = null; eurekatransport = null; instanceregionchecker = new instanceregionchecker(new propertybasedaztoregionmapper(config), clientconfig.getregion()); // this is a bit of hack to allow for existing code using discoverymanager.getinstance() // to work with di'd discoveryclient discoverymanager.getinstance().setdiscoveryclient(this); discoverymanager.getinstance().seteurekaclientconfig(config); inittimestampms = system.currenttimemillis(); logger.info("discovery client initialized at timestamp {} with initial instances count: {}", inittimestampms, this.getapplications().size()); return; // no need to setup up an network tasks and we are done } try { // default size of 2 - 1 each for heartbeat and cacherefresh scheduler = executors.newscheduledthreadpool(2, new threadfactorybuilder() .setnameformat("discoveryclient-%d") .setdaemon(true) .build()); heartbeatexecutor = new threadpoolexecutor( 1, clientconfig.getheartbeatexecutorthreadpoolsize(), 0, timeunit.seconds, new synchronousqueue<runnable>(), new threadfactorybuilder() .setnameformat("discoveryclient-heartbeatexecutor-%d") .setdaemon(true) .build() ); // use direct handoff cacherefreshexecutor = new threadpoolexecutor( 1, clientconfig.getcacherefreshexecutorthreadpoolsize(), 0, timeunit.seconds, new synchronousqueue<runnable>(), new threadfactorybuilder() .setnameformat("discoveryclient-cacherefreshexecutor-%d") .setdaemon(true) .build() ); // use direct handoff eurekatransport = new eurekatransport(); scheduleserverendpointtask(eurekatransport, args); //....省略部分代碼 initscheduledtasks(); //....} |
根據(jù)源代碼，我們可以得出以下結(jié)論：
1）如果shouldregisterwitheureka與shouldfetchregistry都為false,那么直接return。
2）創(chuàng)建發(fā)送心跳與刷新緩存的線程池
3）初始化創(chuàng)建的定時(shí)任務(wù)
那么我們?cè)诳纯磇nitscheduledtasks()方法里有如下代碼：
|
1
2
3
4
5
6
7
8
9
10
11
12
|
// heartbeat timer scheduler.schedule( new timedsupervisortask( "heartbeat", scheduler, heartbeatexecutor, renewalintervalinsecs, timeunit.seconds, expbackoffbound, new heartbeatthread() ), renewalintervalinsecs, timeunit.seconds); |
此處是觸發(fā)一個(gè)定時(shí)執(zhí)行的線程，以秒為單位，根據(jù)renewalintervalinsecs值定時(shí)執(zhí)行發(fā)送心跳，heartbeatthread線程執(zhí)行如下：
|
1
2
3
4
5
6
7
8
9
10
11
|
/** * the heartbeat task that renews the lease in the given intervals. */private class heartbeatthread implements runnable { public void run() { if (renew()) { lastsuccessfulheartbeattimestamp = system.currenttimemillis(); } }} |
我們可以看到run方法里很簡單執(zhí)行renew方法，如果成功記錄一下時(shí)間。renew方法：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
/** * renew with the eureka service by making the appropriate rest call */boolean renew() { eurekahttpresponse<instanceinfo> httpresponse; try { httpresponse = eurekatransport.registrationclient.sendheartbeat(instanceinfo.getappname(), instanceinfo.getid(), instanceinfo, null); logger.debug("{} - heartbeat status: {}", prefix + apppathidentifier, httpresponse.getstatuscode()); if (httpresponse.getstatuscode() == 404) { reregister_counter.increment(); logger.info("{} - re-registering apps/{}", prefix + apppathidentifier, instanceinfo.getappname()); long timestamp = instanceinfo.setisdirtywithtime(); boolean success = register(); if (success) { instanceinfo.unsetisdirty(timestamp); } return success; } return httpresponse.getstatuscode() == 200; } catch (throwable e) { logger.error("{} - was unable to send heartbeat!", prefix + apppathidentifier, e); return false; }} |
在這里發(fā)送心跳如果返回的是404，那么會(huì)執(zhí)行注冊(cè)操作，注意我們根據(jù)返回值httpresponse可以斷定這一切的操作都是基于http請(qǐng)求的，到底是不是呢？我們繼續(xù)看一下register方法：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
/** * register with the eureka service by making the appropriate rest call. */boolean register() throws throwable { logger.info(prefix + apppathidentifier + ": registering service..."); eurekahttpresponse<void> httpresponse; try { httpresponse = eurekatransport.registrationclient.register(instanceinfo); } catch (exception e) { logger.warn("{} - registration failed {}", prefix + apppathidentifier, e.getmessage(), e); throw e; } if (logger.isinfoenabled()) { logger.info("{} - registration status: {}", prefix + apppathidentifier, httpresponse.getstatuscode()); } return httpresponse.getstatuscode() == 204;} |
在這里又調(diào)用了eurekatransport里registrationclient的方法：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
|
private static final class eurekatransport { private closableresolver bootstrapresolver; private transportclientfactory transportclientfactory; private eurekahttpclient registrationclient; private eurekahttpclientfactory registrationclientfactory; private eurekahttpclient queryclient; private eurekahttpclientfactory queryclientfactory; void shutdown() { if (registrationclientfactory != null) { registrationclientfactory.shutdown(); } if (queryclientfactory != null) { queryclientfactory.shutdown(); } if (registrationclient != null) { registrationclient.shutdown(); } if (queryclient != null) { queryclient.shutdown(); } if (transportclientfactory != null) { transportclientfactory.shutdown(); } if (bootstrapresolver != null) { bootstrapresolver.shutdown(); } } } |
在這里我們可以看到，eureka的客戶端是使用http請(qǐng)求進(jìn)行注冊(cè)服務(wù)的，也就是說當(dāng)我們創(chuàng)建discoveryclient就會(huì)向服務(wù)端進(jìn)行實(shí)例的注冊(cè)。
三、服務(wù)端提供的rest服務(wù)
服務(wù)端提供用于處理客戶端注冊(cè)請(qǐng)求的代碼我們已經(jīng)看過了，既然客戶端是通過走h(yuǎn)ttp協(xié)議進(jìn)行注冊(cè)的，那服務(wù)端總要有處理這個(gè)http請(qǐng)求的地址吧，其實(shí)eureka服務(wù)端是采用jax-rs標(biāo)準(zhǔn)提供rest方式進(jìn)行暴露服務(wù)的，我們可以看一下這個(gè)類applicationresoure的addinstance方法:
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
|
/** * registers information about a particular instance for an * {@link com.netflix.discovery.shared.application}. * * @param info * {@link instanceinfo} information of the instance. * @param isreplication * a header parameter containing information whether this is * replicated from other nodes. */@post@consumes({"application/json", "application/xml"})public response addinstance(instanceinfo info, @headerparam(peereurekanode.header_replication) string isreplication) { logger.debug("registering instance {} (replication={})", info.getid(), isreplication); // validate that the instanceinfo contains all the necessary required fields if (isblank(info.getid())) { return response.status(400).entity("missing instanceid").build(); } else if (isblank(info.gethostname())) { return response.status(400).entity("missing hostname").build(); } else if (isblank(info.getipaddr())) { return response.status(400).entity("missing ip address").build(); } else if (isblank(info.getappname())) { return response.status(400).entity("missing appname").build(); } else if (!appname.equals(info.getappname())) { return response.status(400).entity("mismatched appname, expecting " + appname + " but was " + info.getappname()).build(); } else if (info.getdatacenterinfo() == null) { return response.status(400).entity("missing datacenterinfo").build(); } else if (info.getdatacenterinfo().getname() == null) { return response.status(400).entity("missing datacenterinfo name").build(); } // handle cases where clients may be registering with bad datacenterinfo with missing data datacenterinfo datacenterinfo = info.getdatacenterinfo(); if (datacenterinfo instanceof uniqueidentifier) { string datacenterinfoid = ((uniqueidentifier) datacenterinfo).getid(); if (isblank(datacenterinfoid)) { boolean experimental = "true".equalsignorecase(serverconfig.getexperimental("registration.validation.datacenterinfoid")); if (experimental) { string entity = "datacenterinfo of type " + datacenterinfo.getclass() + " must contain a valid id"; return response.status(400).entity(entity).build(); } else if (datacenterinfo instanceof amazoninfo) { amazoninfo amazoninfo = (amazoninfo) datacenterinfo; string effectiveid = amazoninfo.get(amazoninfo.metadatakey.instanceid); if (effectiveid == null) { amazoninfo.getmetadata().put(amazoninfo.metadatakey.instanceid.getname(), info.getid()); } } else { logger.warn("registering datacenterinfo of type {} without an appropriate id", datacenterinfo.getclass()); } } } registry.register(info, "true".equals(isreplication)); return response.status(204).build(); // 204 to be backwards compatible} |
以上就是本文的全部內(nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持服務(wù)器之家。
原文鏈接：http://www.cnblogs.com/niechen/p/9092544.html
