eureka是一種去中心化的服務治理應用，其顯著特點是既可以作為服務端又可以作為服務向自己配置的地址進行注冊。那么這篇文章就來探討一下eureka的注冊流程。
一、eureka的服務端
eureka的服務端核心類是eurekabootstrap，該類實現了一個servletcontextlistener的監聽器。因此我們可以斷定eureka是基于servlet容器實現的。關鍵代碼如下：
|
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 { //...省略相關代碼/** * 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); } } //省略相關代碼.....} |
我們可以看到在servletcontext初始化完成時，會初始化eureka環境，然后初始化eurekaservercontext，那么我們在看一看initeurekaservercontext方法：
|
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 ); } //....省略部分代碼 } |
在這個方法里會創建許多與eureka服務相關的對象，在這里我列舉了兩個核心對象分別是eurekaclient與peerawareinstanceregistry，關于客戶端部分我們等會再說，我們現在來看看peerawareinstanceregistry到底是做什么用的，這里我寫貼出關于這個類的類圖：
根據類圖我們可以清晰的發現peerawareinstanceregistry的最頂層接口為leasemanager與lookupservice,其中lookupservice定義了最基本的發現示例的行為而leasemanager定義了處理客戶端注冊，續約，注銷等操作。那么在這篇文章我們還是重點關注一下leasemanager的相關接口的實現。回過頭來我們在看peerawareinstanceregistry，其實這個類用于多個節點下復制相關信息，比如說一個節點注冊續約與下線那么通過這個類將會相關復制(通知)到各個節點。我們來看看它是怎么處理客戶端注冊的：
|
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);} |
我們可以看到它調用了父類的register方法后又通過replicatetopeers復制對應的行為到其他節點，具體如何復制的先不在這里討論，我們重點來看看注冊方法，我們在父類里找到register()方法：
|
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）首先根據appname獲取一些列的服務實例對象，如果為null,則新創建一個map并把當前的注冊應用程序信息添加到此map當中，這里有一個lease對象，這個類描述了泛型t的時間屬性，比如說注冊時間，服務啟動時間，最后更新時間等，大家可以關注一下它的實現：
|
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）根據當前注冊的id，如果能在map中取到則做以下操作：
2.1）根據當前存在節點的觸碰時間和注冊節點的觸碰時間比較，如果前者的時間晚于后者的時間，那么當前注冊的實例就以已存在的實例為準
2.2）否則更新其每分鐘期望的續約數量及其閾值
3）將當前的注冊節點存到map當中，至此我們的注冊過程基本告一段落了
二、eureka客戶端
在服務端servletcontext初始化完畢時，會創建discoveryclient。熟悉eureka的朋友，一定熟悉這兩個屬性：fetchregistry與registerwitheureka。在springcloud中集成eureka獨立模式運行時，如果這兩個值不為false，那么啟動會報錯，為什么會報錯呢？其實答案就在discoveryclient的構造函數中：
|
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(); //....} |
根據源代碼，我們可以得出以下結論：
1）如果shouldregisterwitheureka與shouldfetchregistry都為false,那么直接return。
2）創建發送心跳與刷新緩存的線程池
3）初始化創建的定時任務
那么我們在看看initscheduledtasks()方法里有如下代碼：
|
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); |
此處是觸發一個定時執行的線程，以秒為單位，根據renewalintervalinsecs值定時執行發送心跳，heartbeatthread線程執行如下：
|
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方法里很簡單執行renew方法，如果成功記錄一下時間。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; }} |
在這里發送心跳如果返回的是404，那么會執行注冊操作，注意我們根據返回值httpresponse可以斷定這一切的操作都是基于http請求的，到底是不是呢？我們繼續看一下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;} |
在這里又調用了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請求進行注冊服務的，也就是說當我們創建discoveryclient就會向服務端進行實例的注冊。
三、服務端提供的rest服務
服務端提供用于處理客戶端注冊請求的代碼我們已經看過了，既然客戶端是通過走http協議進行注冊的，那服務端總要有處理這個http請求的地址吧，其實eureka服務端是采用jax-rs標準提供rest方式進行暴露服務的，我們可以看一下這個類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} |
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持服務器之家。
原文鏈接：http://www.cnblogs.com/niechen/p/9092544.html
