Skip to content

Commit 7665844

Browse files
zhou-haozhou-hao
committed
refactor: 优化
1 parent 79b2257 commit 7665844

4 files changed

Lines changed: 202 additions & 49 deletions

File tree

‎jetlinks-components/network-component/http-component/src/main/java/org/jetlinks/community/network/http/server/vertx/VertxHttpExchange.java‎

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import io.netty.util.ReferenceCountUtil;
2222
import io.vertx.core.MultiMap;
2323
import io.vertx.core.buffer.Buffer;
24+
import io.vertx.core.buffer.impl.BufferImpl;
2425
import io.vertx.core.http.HttpMethod;
2526
import io.vertx.core.http.HttpServerRequest;
2627
import io.vertx.core.http.HttpServerResponse;
@@ -29,6 +30,7 @@
2930
import lombok.Getter;
3031
import lombok.Setter;
3132
import lombok.extern.slf4j.Slf4j;
33+
import org.hswebframework.web.exception.BusinessException;
3234
import org.jetlinks.core.message.codec.http.Header;
3335
import org.jetlinks.core.message.codec.http.HttpRequestMessage;
3436
import org.jetlinks.core.message.codec.http.HttpResponseMessage;
@@ -44,6 +46,7 @@
4446
import reactor.core.publisher.Mono;
4547

4648
import java.net.InetSocketAddress;
49+
import java.time.Duration;
4750
import java.util.*;
4851
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
4952
import java.util.function.Function;
@@ -77,6 +80,8 @@ public class VertxHttpExchange implements HttpExchange, HttpResponse, HttpReques
7780

7881
private volatile Boolean alreadyResponse = false;
7982

83+
private int handleCount;
84+
8085
public VertxHttpExchange(HttpServerRequest httpServerRequest,
8186
HttpServerConfig config) {
8287

@@ -88,10 +93,9 @@ public VertxHttpExchange(HttpServerRequest httpServerRequest,
8893
if (httpServerRequest.method() == HttpMethod.GET) {
8994
body = Mono.just(Unpooled.EMPTY_BUFFER);
9095
} else {
91-
9296
Mono<ByteBuf> buffer = Mono
9397
.fromCompletionStage(this.httpServerRequest.body().toCompletionStage())
94-
.map(Buffer::getByteBuf);
98+
.map(buf->((BufferImpl) buf).getByteBuf());
9599

96100
if (MultiPart.isMultiPart(getContentType())) {
97101
body = MultiPart
@@ -100,12 +104,22 @@ public VertxHttpExchange(HttpServerRequest httpServerRequest,
100104
.thenReturn(Unpooled.EMPTY_BUFFER)
101105
.cache();
102106
} else {
103-
body = buffer;
107+
body = buffer.cache();
104108
}
105109

106110
}
107111
}
108112

113+
114+
void mark() {
115+
handleCount++;
116+
}
117+
118+
int handleCount() {
119+
return handleCount;
120+
}
121+
122+
109123
@Override
110124
@Generated
111125
public String requestId() {

‎jetlinks-components/network-component/http-component/src/main/java/org/jetlinks/community/network/http/server/vertx/VertxHttpServer.java‎

Lines changed: 153 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -15,23 +15,32 @@
1515
*/
1616
package org.jetlinks.community.network.http.server.vertx;
1717

18+
import io.netty.handler.codec.http.websocketx.WebSocketCloseStatus;
19+
import io.vertx.core.http.HttpClosedException;
1820
import lombok.*;
1921
import lombok.extern.slf4j.Slf4j;
22+
import org.hswebframework.web.exception.I18nSupportException;
23+
import org.jetlinks.core.lang.SeparatedCharSequence;
24+
import org.jetlinks.core.lang.SharedPathString;
2025
import org.jetlinks.core.topic.Topic;
2126
import org.jetlinks.community.network.DefaultNetworkType;
2227
import org.jetlinks.community.network.NetworkType;
2328
import org.jetlinks.community.network.http.server.HttpExchange;
2429
import org.jetlinks.community.network.http.server.HttpServer;
2530
import org.jetlinks.community.network.http.server.WebSocketExchange;
31+
import org.jetlinks.core.utils.TopicUtils;
32+
import org.springframework.http.HttpMethod;
2633
import org.springframework.http.HttpStatus;
2734
import reactor.core.Disposable;
2835
import reactor.core.Disposables;
2936
import reactor.core.publisher.Flux;
3037
import reactor.core.publisher.FluxSink;
31-
import reactor.core.publisher.Mono;
3238

3339
import java.net.InetSocketAddress;
3440
import java.util.Collection;
41+
import java.util.Map;
42+
import java.util.concurrent.ConcurrentHashMap;
43+
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
3544
import java.util.stream.Collectors;
3645
import java.util.stream.Stream;
3746

@@ -45,12 +54,22 @@
4554
@Slf4j
4655
public class VertxHttpServer implements HttpServer {
4756

57+
private static final SharedPathString WEBSOCKET_PREFIX = SharedPathString.of("/ws");
58+
59+
private static final Map<HttpMethod, SeparatedCharSequence> HTTP_PREFIX_CACHE = new ConcurrentHashMap<>();
60+
4861
private Collection<io.vertx.core.http.HttpServer> httpServers;
4962

5063
private HttpServerConfig config;
5164

5265
private String id;
5366

67+
private static final AtomicLongFieldUpdater<VertxHttpServer> PENDING =
68+
AtomicLongFieldUpdater.newUpdater(VertxHttpServer.class, "pending");
69+
private volatile long pending;
70+
71+
private int maxConcurrency = -1;
72+
5473
private final Topic<FluxSink<HttpExchange>> route = Topic.createRoot();
5574
private final Topic<FluxSink<WebSocketExchange>> websocketRoute = Topic.createRoot();
5675

@@ -79,59 +98,151 @@ public void setHttpServers(Collection<io.vertx.core.http.HttpServer> httpServers
7998
for (io.vertx.core.http.HttpServer server : this.httpServers) {
8099
server
81100
.webSocketHandler(socket -> {
82-
socket.exceptionHandler(err -> {
83-
log.error(err.getMessage(), err);
84-
});
85-
86-
String url = socket.path();
87-
if (url.endsWith("/")) {
88-
url = url.substring(0, url.length() - 1);
101+
long pending = PENDING.incrementAndGet(this);
102+
if (maxConcurrency > 0 && pending >= maxConcurrency) {
103+
PENDING.decrementAndGet(this);
104+
socket.close((short) WebSocketCloseStatus.TRY_AGAIN_LATER.code());
105+
return;
89106
}
90-
VertxWebSocketExchange exchange = new VertxWebSocketExchange(socket);
91-
92-
websocketRoute
93-
.findTopic("/ws" + url)
94-
.flatMapIterable(Topic::getSubscribers)
95-
.doOnNext(sink -> sink.next(exchange))
96-
.switchIfEmpty(Mono.fromRunnable(() -> {
97107

98-
log.warn("http server no handler for:[{}://{}{}]", socket.scheme(), socket.host(), socket.path());
99-
socket.reject(404);
100-
101-
}))
102-
.subscribe();
108+
socket.exceptionHandler(err -> {
109+
if(err instanceof HttpClosedException){
110+
return;
111+
}
112+
log.warn(err.getMessage(), err);
113+
});
114+
VertxWebSocketExchange exchange = new VertxWebSocketExchange(
115+
socket
116+
);
117+
exchange.closeHandler(() -> PENDING.decrementAndGet(this));
118+
Topic.find(
119+
WEBSOCKET_PREFIX.append(parsePath(exchange.getPath())),
120+
websocketRoute,
121+
exchange,
122+
socket,
123+
null,
124+
null,
125+
(_exchange, _req, nil, nil2, topic) -> {
126+
for (FluxSink<WebSocketExchange> sink : topic.getSubscribers()) {
127+
_exchange.mark();
128+
sink.next(_exchange);
129+
}
130+
}, (_exchange, _socket, nil2, nil) -> {
131+
if (_exchange.handleCount() == 0) {
132+
if (log.isInfoEnabled()) {
133+
log.info("http server no handler for:[{}://{}{}] remote: {}",
134+
_socket.scheme(),
135+
_socket.authority(),
136+
_socket.path(),
137+
_socket.remoteAddress());
138+
}
139+
_socket.close((short) WebSocketCloseStatus.ENDPOINT_UNAVAILABLE.code());
140+
}
141+
});
103142

104143
})
105144
.requestHandler(request -> {
106-
request.exceptionHandler(err -> {
107-
log.error(err.getMessage(), err);
108-
});
109145

110-
VertxHttpExchange exchange = new VertxHttpExchange(request, config);
146+
request.endHandler(ignore -> PENDING.decrementAndGet(this));
111147

112-
String url = exchange.getUrl();
113-
if (url.endsWith("/")) {
114-
url = url.substring(0, url.length() - 1);
148+
long pending = PENDING.incrementAndGet(this);
149+
if (maxConcurrency > 0 && pending >= maxConcurrency) {
150+
request
151+
.response()
152+
.setStatusCode(HttpStatus.SERVICE_UNAVAILABLE.value())
153+
.end();
154+
return;
155+
}
156+
request.exceptionHandler(err -> {
157+
if(err instanceof HttpClosedException){
158+
return;
159+
}
160+
log.warn(err.getMessage(), err);
161+
});
162+
try {
163+
VertxHttpExchange exchange = new VertxHttpExchange(request, config);
164+
Topic.find(
165+
getHttpPrefix(exchange.request().getMethod())
166+
.append(parsePath(exchange.getPath())),
167+
route,
168+
exchange,
169+
request,
170+
null,
171+
null,
172+
//查找到订阅者
173+
(_exchange,
174+
_req,
175+
nil, nil2,
176+
topic) -> {
177+
for (FluxSink<HttpExchange> sink : topic.getSubscribers()) {
178+
_exchange.mark();
179+
sink.next(_exchange);
180+
}
181+
},
182+
//全部查找结束
183+
(_exchange,
184+
_req,
185+
nil2, nil) -> {
186+
if (_exchange.handleCount() == 0) {
187+
if (log.isInfoEnabled()) {
188+
log.info("http server no handler for:[{} {}://{}:{}] remote: {}",
189+
_req.method(),
190+
_req.scheme(),
191+
_req.authority(),
192+
_req.path(),
193+
_req.remoteAddress());
194+
}
195+
_req
196+
.response()
197+
.setStatusCode(HttpStatus.NOT_FOUND.value())
198+
.end();
199+
}
200+
});
201+
} catch (Throwable e) {
202+
request
203+
.response()
204+
.setStatusCode(HttpStatus.BAD_GATEWAY.value())
205+
.end();
115206
}
116-
117-
route.findTopic("/" + exchange.request().getMethod().name().toLowerCase() + url)
118-
.flatMapIterable(Topic::getSubscribers)
119-
.doOnNext(sink -> sink.next(exchange))
120-
.switchIfEmpty(Mono.fromRunnable(() -> {
121-
122-
log.warn("http server no handler for:[{} {}://{}{}]", request.method(), request.scheme(), request.host(), request.path());
123-
request.response()
124-
.setStatusCode(HttpStatus.NOT_FOUND.value())
125-
.end();
126-
127-
}))
128-
.subscribe();
129207

130208
});
131-
server.exceptionHandler(err -> log.error(err.getMessage(), err));
209+
server.exceptionHandler(err -> {
210+
// 忽略
211+
if (err instanceof HttpClosedException) {
212+
return;
213+
}
214+
log.warn("http server [{}] error", bindAddress, err);
215+
});
132216
}
133217
}
134218

219+
private SeparatedCharSequence parsePath(String url) {
220+
while (url.charAt(url.length() - 1) == '/') {
221+
url = url.substring(0, url.length() - 1);
222+
}
223+
String[] split = TopicUtils.split(
224+
url,
225+
'/',
226+
(i, p) -> {
227+
//不支持通配符
228+
if ("*".equals(p) || "**".equals(p)) {
229+
throw new I18nSupportException.NoStackTrace("error.illegal_http_path");
230+
}
231+
return p;
232+
});
233+
return SharedPathString.of(split);
234+
}
235+
236+
private SeparatedCharSequence getHttpPrefix(HttpMethod method) {
237+
return HTTP_PREFIX_CACHE.computeIfAbsent(
238+
method,
239+
m -> {
240+
String prefix = m.name().toLowerCase();
241+
return SharedPathString.of("/" + prefix);
242+
});
243+
}
244+
245+
135246
@Override
136247
public Flux<HttpExchange> handleRequest() {
137248
return handleRequest("*", "/**");
@@ -144,7 +255,7 @@ public Flux<WebSocketExchange> handleWebsocket(String urlPattern) {
144255

145256
@Override
146257
public Flux<HttpExchange> handleRequest(String method, String... urlPatterns) {
147-
return createRoute(route, method, urlPatterns);
258+
return createRoute(route, method.toLowerCase(), urlPatterns);
148259
}
149260

150261
private <T> Flux<T> createRoute(Topic<FluxSink<T>> root, String prefix, String... urlPatterns) {

‎jetlinks-components/network-component/http-component/src/main/java/org/jetlinks/community/network/http/server/vertx/VertxWebSocketExchange.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,7 @@ public class VertxWebSocketExchange implements WebSocketExchange {
7676

7777
@Getter
7878
private final String id;
79+
private int handleCount;
7980

8081
public VertxWebSocketExchange(ServerWebSocket serverWebSocket) {
8182
this.serverWebSocket = serverWebSocket;
@@ -85,6 +86,16 @@ public VertxWebSocketExchange(ServerWebSocket serverWebSocket) {
8586
this.id = IDGenerator.RANDOM.generate();
8687
}
8788

89+
90+
void mark() {
91+
handleCount++;
92+
}
93+
94+
int handleCount() {
95+
return handleCount;
96+
}
97+
98+
8899
@Override
89100
public Optional<InetSocketAddress> getRemoteAddress() {
90101
return Optional.of(address);

0 commit comments

Comments
 (0)