在接入spring-cloud-gateway時，可能有需求進行緩存json-body數據或者form-urlencoded數據的情況。
由于spring-cloud-gateway是以webflux為基礎的響應式架構設計，所以在原有zuul基礎上遷移過來的過程中，傳統的編程思路，并不適合于reactor stream的開發。
網絡上有許多緩存案例，但是在測試過程中出現各種bug問題，在緩存body時，需要考慮整體的響應式操作，才能更合理的緩存數據
下面提供緩存json-body數據或者form-urlencoded數據的具體實現方案，該方案經測試，滿足各方面需求，以及避免了網絡上其他緩存方案所出現的問題
定義一個gatewaycontext類，用于存儲請求中緩存的數據
|
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
|
import lombok.getter;import lombok.setter;import lombok.tostring;import org.springframework.util.linkedmultivaluemap;import org.springframework.util.multivaluemap;@getter@setter@tostringpublic class gatewaycontext { public static final string cache_gateway_context = "cachegatewaycontext"; /** * cache json body */ private string cachebody; /** * cache formdata */ private multivaluemap<string, string> formdata; /** * cache reqeust path */ private string path;} |
實現globalfilter和ordered接口用于緩存請求數據
1 . 該示例只支持緩存下面3種mediatype
- application_json--json數據
- application_json_utf8--json數據
- application_form_urlencoded--formdata表單數據
2 . 經驗總結:
- 在緩存body時,不能夠在filter內部直接進行緩存，需要按照響應式的處理方式，在異步操作路途上進行緩存body，由于body只能讀取一次，所以要讀取完成后要重新封裝新的request和exchange才能保證請求正常傳遞到下游
- 在緩存formdata時，formdata也只能讀取一次，所以在讀取完畢后，需要重新封裝request和exchange,這里要注意，如果對formdata內容進行了修改，則必須重新定義header中的content-length已保證傳輸數據的大小一致
|
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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
|
import com.choice.cloud.architect.usergate.option.filterorderenum;import com.choice.cloud.architect.usergate.support.gatewaycontext;import io.netty.buffer.bytebufallocator;import lombok.extern.slf4j.slf4j;import org.springframework.cloud.gateway.filter.gatewayfilterchain;import org.springframework.cloud.gateway.filter.globalfilter;import org.springframework.core.ordered;import org.springframework.core.io.bytearrayresource;import org.springframework.core.io.buffer.databuffer;import org.springframework.core.io.buffer.databufferutils;import org.springframework.core.io.buffer.nettydatabufferfactory;import org.springframework.http.httpheaders;import org.springframework.http.mediatype;import org.springframework.http.codec.httpmessagereader;import org.springframework.http.server.reactive.serverhttprequest;import org.springframework.http.server.reactive.serverhttprequestdecorator;import org.springframework.util.multivaluemap;import org.springframework.web.reactive.function.server.handlerstrategies;import org.springframework.web.reactive.function.server.serverrequest;import org.springframework.web.server.serverwebexchange;import reactor.core.publisher.flux;import reactor.core.publisher.mono;import java.io.unsupportedencodingexception;import java.net.urlencoder;import java.nio.charset.charset;import java.nio.charset.standardcharsets;import java.util.list;import java.util.map;@slf4jpublic class gatewaycontextfilter implements globalfilter, ordered { /** * default httpmessagereader */ private static final list<httpmessagereader<?>> messagereaders = handlerstrategies.withdefaults().messagereaders(); @override public mono<void> filter(serverwebexchange exchange, gatewayfilterchain chain) { /** * save request path and serviceid into gateway context */ serverhttprequest request = exchange.getrequest(); string path = request.getpath().pathwithinapplication().value(); gatewaycontext gatewaycontext = new gatewaycontext(); gatewaycontext.getallrequestdata().addall(request.getqueryparams()); gatewaycontext.setpath(path); /** * save gateway context into exchange */ exchange.getattributes().put(gatewaycontext.cache_gateway_context,gatewaycontext); httpheaders headers = request.getheaders(); mediatype contenttype = headers.getcontenttype(); long contentlength = headers.getcontentlength(); if(contentlength>0){ if(mediatype.application_json.equals(contenttype) || mediatype.application_json_utf8.equals(contenttype)){ return readbody(exchange, chain,gatewaycontext); } if(mediatype.application_form_urlencoded.equals(contenttype)){ return readformdata(exchange, chain,gatewaycontext); } } log.debug("[gatewaycontext]contenttype:{},gateway context is set with {}",contenttype, gatewaycontext); return chain.filter(exchange); } @override public int getorder() { return integer.min_value; } /** * readformdata * @param exchange * @param chain * @return */ private mono<void> readformdata(serverwebexchange exchange,gatewayfilterchain chain,gatewaycontext gatewaycontext){ httpheaders headers = exchange.getrequest().getheaders(); return exchange.getformdata() .doonnext(multivaluemap -> { gatewaycontext.setformdata(multivaluemap); log.debug("[gatewaycontext]read formdata:{}",multivaluemap); }) .then(mono.defer(() -> { charset charset = headers.getcontenttype().getcharset(); charset = charset == null? standardcharsets.utf_8:charset; string charsetname = charset.name(); multivaluemap<string, string> formdata = gatewaycontext.getformdata(); /** * formdata is empty just return */ if(null == formdata || formdata.isempty()){ return chain.filter(exchange); } stringbuilder formdatabodybuilder = new stringbuilder(); string entrykey; list<string> entryvalue; try { /** * remove system param ,repackage form data */ for (map.entry<string, list<string>> entry : formdata.entryset()) { entrykey = entry.getkey(); entryvalue = entry.getvalue(); if (entryvalue.size() > 1) { for(string value : entryvalue){ formdatabodybuilder.append(entrykey).append("=").append(urlencoder.encode(value, charsetname)).append("&"); } } else { formdatabodybuilder.append(entrykey).append("=").append(urlencoder.encode(entryvalue.get(0), charsetname)).append("&"); } } }catch (unsupportedencodingexception e){ //ignore urlencode exception } /** * substring with the last char '&' */ string formdatabodystring = ""; if(formdatabodybuilder.length()>0){ formdatabodystring = formdatabodybuilder.substring(0, formdatabodybuilder.length() - 1); } /** * get data bytes */ byte[] bodybytes = formdatabodystring.getbytes(charset); int contentlength = bodybytes.length; serverhttprequestdecorator decorator = new serverhttprequestdecorator( exchange.getrequest()) { /** * change content-length * @return */ @override public httpheaders getheaders() { httpheaders httpheaders = new httpheaders(); httpheaders.putall(super.getheaders()); if (contentlength > 0) { httpheaders.setcontentlength(contentlength); } else { httpheaders.set(httpheaders.transfer_encoding, "chunked"); } return httpheaders; } /** * read bytes to flux<databuffer> * @return */ @override public flux<databuffer> getbody() { return databufferutils.read(new bytearrayresource(bodybytes),new nettydatabufferfactory(bytebufallocator.default),contentlength); } }; serverwebexchange mutateexchange = exchange.mutate().request(decorator).build(); log.debug("[gatewaycontext]rewrite form data :{}",formdatabodystring); return chain.filter(mutateexchange); })); } /** * readjsonbody * @param exchange * @param chain * @return */ private mono<void> readbody(serverwebexchange exchange,gatewayfilterchain chain,gatewaycontext gatewaycontext){ /** * join the body */ return databufferutils.join(exchange.getrequest().getbody()) .flatmap(databuffer -> { /** * read the body flux<databuffer> */ databufferutils.retain(databuffer); flux<databuffer> cachedflux = flux.defer(() -> flux.just(databuffer.slice(0, databuffer.readablebytecount()))); /** * repackage serverhttprequest */ serverhttprequest mutatedrequest = new serverhttprequestdecorator(exchange.getrequest()) { @override public flux<databuffer> getbody() { return cachedflux; } }; /** * mutate exchage with new serverhttprequest */ serverwebexchange mutatedexchange = exchange.mutate().request(mutatedrequest).build(); /** * read body string with default messagereaders */ return serverrequest.create(mutatedexchange, messagereaders) .bodytomono(string.class) .doonnext(objectvalue -> { gatewaycontext.setcachebody(objectvalue); log.debug("[gatewaycontext]read jsonbody:{}",objectvalue); }).then(chain.filter(mutatedexchange)); }); }} |
在后續filter中，可以直接從serverexchange中獲取gatewaycontext，就可以獲取到緩存的數據，如果需要緩存其他數據，則可以根據自己的需求，添加到gatewaycontext中即可
復制代碼 代碼如下:
gatewaycontext gatewaycontext = exchange.getattribute(gatewaycontext.cache_gateway_context);
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持服務器之家。
原文鏈接：https://segmentfault.com/a/1190000017898354
