This is an automated email from the ASF dual-hosted git repository.
weihu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new f22e30ff0 Make field final. Make Random performance better.
new bd1c980a1 Merge pull request #3802 from pandaapo/master-issue3461
f22e30ff0 is described below
commit f22e30ff0d7fabeb0bc119c7f35a0d6877727162
Author: pandaapo <[email protected]>
AuthorDate: Sun Apr 23 03:02:23 2023 +0800
Make field final. Make Random performance better.
---
.../http/push/AbstractHTTPPushRequest.java | 22 +++++++++++-----------
.../protocol/http/push/AsyncHTTPPushRequest.java | 17 +++++++----------
.../core/protocol/http/push/HTTPClientPool.java | 6 +++---
.../protocol/http/push/HTTPMessageHandler.java | 12 ++++++------
4 files changed, 27 insertions(+), 30 deletions(-)
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AbstractHTTPPushRequest.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AbstractHTTPPushRequest.java
index 42ad2cedd..3fd83268d 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AbstractHTTPPushRequest.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AbstractHTTPPushRequest.java
@@ -26,38 +26,38 @@ import
org.apache.eventmesh.runtime.core.protocol.http.retry.RetryContext;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
-import org.apache.commons.lang3.RandomUtils;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;
import com.google.common.collect.Lists;
public abstract class AbstractHTTPPushRequest extends RetryContext {
- public EventMeshHTTPServer eventMeshHTTPServer;
+ public final EventMeshHTTPServer eventMeshHTTPServer;
- public long createTime = System.currentTimeMillis();
+ public final long createTime = System.currentTimeMillis();
public long lastPushTime = System.currentTimeMillis();
- public Map<String /** IDC*/, List<String>> urls;
+ public final Map<String /** IDC*/, List<String>> urls;
- public List<String> totalUrls;
+ public final List<String> totalUrls;
public volatile int startIdx;
- public EventMeshHTTPConfiguration eventMeshHttpConfiguration;
+ public final EventMeshHTTPConfiguration eventMeshHttpConfiguration;
- public HttpRetryer retryer;
+ public final HttpRetryer retryer;
- public int ttl;
+ public final int ttl;
- public HandleMsgContext handleMsgContext;
+ public final HandleMsgContext handleMsgContext;
- private AtomicBoolean complete = new AtomicBoolean(Boolean.FALSE);
+ private final AtomicBoolean complete = new AtomicBoolean(Boolean.FALSE);
public AbstractHTTPPushRequest(HandleMsgContext handleMsgContext) {
this.eventMeshHTTPServer = handleMsgContext.getEventMeshHTTPServer();
@@ -67,7 +67,7 @@ public abstract class AbstractHTTPPushRequest extends
RetryContext {
this.eventMeshHttpConfiguration =
handleMsgContext.getEventMeshHTTPServer().getEventMeshHttpConfiguration();
this.retryer =
handleMsgContext.getEventMeshHTTPServer().getHttpRetryer();
this.ttl = handleMsgContext.getTtl();
- this.startIdx = RandomUtils.nextInt(0, totalUrls.size());
+ this.startIdx = ThreadLocalRandom.current().nextInt(0,
totalUrls.size());
}
public void tryHTTPRequest() {
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java
index 891a66ee1..02732e6c3 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java
@@ -51,7 +51,6 @@ import org.apache.http.util.EntityUtils;
import java.io.IOException;
import java.nio.charset.Charset;
-import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@@ -78,7 +77,7 @@ public class AsyncHTTPPushRequest extends
AbstractHTTPPushRequest {
public static final Logger LOGGER =
LoggerFactory.getLogger("AsyncHTTPPushRequest");
public String currPushUrl;
- private Map<String, Set<AbstractHTTPPushRequest>> waitingRequests;
+ private final Map<String, Set<AbstractHTTPPushRequest>> waitingRequests;
public AsyncHTTPPushRequest(HandleMsgContext handleMsgContext,
Map<String, Set<AbstractHTTPPushRequest>> waitingRequests) {
@@ -136,14 +135,12 @@ public class AsyncHTTPPushRequest extends
AbstractHTTPPushRequest {
content = ((HttpCommand)
protocolTransportObject).getBody().toMap().get("content").toString();
} else {
HttpEventWrapper httpEventWrapper = (HttpEventWrapper)
protocolTransportObject;
- Map<String, Object> sysHeaderMap =
httpEventWrapper.getSysHeaderMap();
- Set<Map.Entry<String, Object>> sysHeaderMapEntry =
sysHeaderMap.entrySet();
- content = new String(httpEventWrapper.getBody(),
StandardCharsets.UTF_8);
- for (Map.Entry<String, Object> header : sysHeaderMapEntry) {
- if (!builder.containsHeader(header.getKey())) {
- builder.addHeader(header.getKey(),
header.getValue().toString());
+ content = new String(httpEventWrapper.getBody(),
Constants.DEFAULT_CHARSET);
+ httpEventWrapper.getSysHeaderMap().forEach((k, v) -> {
+ if (!builder.containsHeader(k)) {
+ builder.addHeader(k, v.toString());
}
- }
+ });
}
} catch (Exception ex) {
@@ -175,7 +172,7 @@ public class AsyncHTTPPushRequest extends
AbstractHTTPPushRequest {
body.add(new BasicNameValuePair(PushMessageRequestBody.EXTFIELDS,
JsonUtils.toJSONString(EventMeshUtil.getEventProp(handleMsgContext.getEvent()))));
- HttpEntity httpEntity = new UrlEncodedFormEntity(body,
StandardCharsets.UTF_8);
+ HttpEntity httpEntity = new UrlEncodedFormEntity(body,
Constants.DEFAULT_CHARSET);
builder.setEntity(httpEntity);
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/HTTPClientPool.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/HTTPClientPool.java
index 0b819758e..fde09ae80 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/HTTPClientPool.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/HTTPClientPool.java
@@ -18,7 +18,6 @@
package org.apache.eventmesh.runtime.core.protocol.http.push;
import org.apache.commons.collections4.CollectionUtils;
-import org.apache.commons.lang3.RandomUtils;
import org.apache.http.config.Registry;
import org.apache.http.config.RegistryBuilder;
import org.apache.http.conn.socket.ConnectionSocketFactory;
@@ -43,6 +42,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import javax.net.ssl.SSLContext;
@@ -55,7 +55,7 @@ public class HTTPClientPool {
private final transient List<CloseableHttpClient> clients =
Collections.synchronizedList(new ArrayList<>());
- private int core;
+ private final int core;
private static final int DEFAULT_MAX_TOTAL = 200;
private static final int DEFAULT_IDLETIME_SECONDS = 30;
@@ -73,7 +73,7 @@ public class HTTPClientPool {
return client;
}
- return clients.get(RandomUtils.nextInt(core, 2 * core) % core);
+ return clients.get(ThreadLocalRandom.current().nextInt(core, 2 * core)
% core);
}
public void shutdown() throws IOException {
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/HTTPMessageHandler.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/HTTPMessageHandler.java
index cad2c394f..0c33337cf 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/HTTPMessageHandler.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/HTTPMessageHandler.java
@@ -44,24 +44,24 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j
public class HTTPMessageHandler implements MessageHandler {
- private transient EventMeshConsumer eventMeshConsumer;
+ private final transient EventMeshConsumer eventMeshConsumer;
private static final transient ScheduledExecutorService SCHEDULER =
ThreadPoolFactory.createSingleScheduledExecutor("eventMesh-pushMsgTimeout");
private static final Integer CONSUMER_GROUP_WAITING_REQUEST_THRESHOLD =
10000;
- public static final transient Map<String, Set<AbstractHTTPPushRequest>>
waitingRequests = Maps.newConcurrentMap();
+ protected static final transient Map<String, Set<AbstractHTTPPushRequest>>
waitingRequests = Maps.newConcurrentMap();
- private transient ThreadPoolExecutor pushExecutor;
+ private final transient ThreadPoolExecutor pushExecutor;
private void checkTimeout() {
- waitingRequests.forEach((key, value) -> {
+ waitingRequests.forEach((key, value) ->
value.forEach(r -> {
r.timeout();
waitingRequests.get(r.handleMsgContext.getConsumerGroup()).remove(r);
- });
- });
+ })
+ );
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
