[ISSUE #3880]Optimize ClientSessionGroupMapping

apache · mxsm · May 7, 2023 · May 7, 2023 · 0364e2dacf1a3e440f295e9310e9205e4b8c21c5 · Pil0tXia
commit 0364e2dacf1a3e440f295e9310e9205e4b8c21c5
diff --git a/...c/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientByIpPortHandler.java b/...c/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientByIpPortHandler.java
@@ -31,7 +31,6 @@
 
 import java.io.IOException;
 import java.io.OutputStream;
-import java.net.InetSocketAddress;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -73,7 +72,7 @@ public void handle(HttpExchange httpExchange) throws IOException {
  log.info("redirectClientByIpPort in admin,ip:{},port:{},destIp:{},destPort:{}====================", ip,
  port, destEventMeshIp, destEventMeshPort);
  ClientSessionGroupMapping clientSessionGroupMapping = eventMeshTCPServer.getClientSessionGroupMapping();
- ConcurrentHashMap<InetSocketAddress, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
+ ConcurrentHashMap<String, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
  StringBuilder redirectResult = new StringBuilder();
  try {
  if (!sessionMap.isEmpty()) {

diff --git a/...src/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientByPathHandler.java b/...src/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientByPathHandler.java
@@ -31,7 +31,6 @@
 
 import java.io.IOException;
 import java.io.OutputStream;
-import java.net.InetSocketAddress;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -75,7 +74,7 @@ public void handle(HttpExchange httpExchange) throws IOException {
  log.info("redirectClientByPath in admin,path:{},destIp:{},destPort:{}====================", path,
  destEventMeshIp, destEventMeshPort);
  ClientSessionGroupMapping clientSessionGroupMapping = eventMeshTCPServer.getClientSessionGroupMapping();
- ConcurrentHashMap<InetSocketAddress, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
+ ConcurrentHashMap<String, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
  StringBuilder redirectResult = new StringBuilder();
  try {
  if (!sessionMap.isEmpty()) {

diff --git a/...ain/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientBySubSystemHandler.java b/...ain/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientBySubSystemHandler.java
@@ -31,7 +31,6 @@
 
 import java.io.IOException;
 import java.io.OutputStream;
-import java.net.InetSocketAddress;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
@@ -79,7 +78,7 @@ public void handle(final HttpExchange httpExchange) throws IOException {
  }
 
  final ClientSessionGroupMapping clientSessionGroupMapping = eventMeshTCPServer.getClientSessionGroupMapping();
- final ConcurrentHashMap<InetSocketAddress, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
+ final ConcurrentHashMap<String, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
  final StringBuilder redirectResult = new StringBuilder();
  try {
  if (!sessionMap.isEmpty()) {

diff --git a/...time/src/main/java/org/apache/eventmesh/runtime/admin/handler/RejectAllClientHandler.java b/...time/src/main/java/org/apache/eventmesh/runtime/admin/handler/RejectAllClientHandler.java
@@ -61,14 +61,14 @@ public RejectAllClientHandler(final EventMeshTCPServer eventMeshTCPServer,
  public void handle(final HttpExchange httpExchange) throws IOException {
  try (OutputStream out = httpExchange.getResponseBody()) {
  final ClientSessionGroupMapping clientSessionGroupMapping = eventMeshTCPServer.getClientSessionGroupMapping();
- final ConcurrentHashMap<InetSocketAddress, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
+ final ConcurrentHashMap<String, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
  final List<InetSocketAddress> successRemoteAddrs = new ArrayList<>();
  try {
  if (log.isInfoEnabled()) {
  log.info("rejectAllClient in admin====================");
  }
  if (!sessionMap.isEmpty()) {
- for (final Map.Entry<InetSocketAddress, Session> entry : sessionMap.entrySet()) {
+ for (final Map.Entry<String, Session> entry : sessionMap.entrySet()) {
  final InetSocketAddress addr = EventMeshTcp2Client.serverGoodby2Client(
  eventMeshTCPServer, entry.getValue(), clientSessionGroupMapping);
  if (addr != null) {

diff --git a/...src/main/java/org/apache/eventmesh/runtime/admin/handler/RejectClientByIpPortHandler.java b/...src/main/java/org/apache/eventmesh/runtime/admin/handler/RejectClientByIpPortHandler.java
@@ -68,16 +68,17 @@ public void handle(HttpExchange httpExchange) throws IOException {
  out.write(result.getBytes(Constants.DEFAULT_CHARSET));
  return;
  }
+ String address = ip + ":" + port;
  log.info("rejectClientByIpPort in admin,ip:{},port:{}====================", ip, port);
  ClientSessionGroupMapping clientSessionGroupMapping = eventMeshTCPServer.getClientSessionGroupMapping();
- ConcurrentHashMap<InetSocketAddress, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
+ ConcurrentHashMap<String, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
  final List<InetSocketAddress> successRemoteAddrs = new ArrayList<InetSocketAddress>();
  try {
  if (!sessionMap.isEmpty()) {
- for (Map.Entry<InetSocketAddress, Session> entry : sessionMap.entrySet()) {
- if (entry.getKey().getHostString().equals(ip) && String.valueOf(entry.getKey().getPort()).equals(port)) {
- InetSocketAddress addr = EventMeshTcp2Client.serverGoodby2Client(eventMeshTCPServer,
- entry.getValue(), clientSessionGroupMapping);
+ for (Map.Entry<String, Session> entry : sessionMap.entrySet()) {
+ if (StringUtils.equals(entry.getKey(), address)) {
+ InetSocketAddress addr = EventMeshTcp2Client.serverGoodby2Client(eventMeshTCPServer, entry.getValue(),
+ clientSessionGroupMapping);
  if (addr != null) {
  successRemoteAddrs.add(addr);
  }

diff --git a/.../main/java/org/apache/eventmesh/runtime/admin/handler/RejectClientBySubSystemHandler.java b/.../main/java/org/apache/eventmesh/runtime/admin/handler/RejectClientBySubSystemHandler.java
@@ -86,7 +86,7 @@ public void handle(HttpExchange httpExchange) throws IOException {
 
  log.info("rejectClientBySubSystem in admin,subsys:{}====================", subSystem);
  ClientSessionGroupMapping clientSessionGroupMapping = eventMeshTCPServer.getClientSessionGroupMapping();
- ConcurrentHashMap<InetSocketAddress, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
+ ConcurrentHashMap<String, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
  final List<InetSocketAddress> successRemoteAddrs = new ArrayList<>();
  try {
  if (!sessionMap.isEmpty()) {

diff --git a/...e/src/main/java/org/apache/eventmesh/runtime/admin/handler/ShowClientBySystemHandler.java b/...e/src/main/java/org/apache/eventmesh/runtime/admin/handler/ShowClientBySystemHandler.java
@@ -29,7 +29,6 @@
 
 import java.io.IOException;
 import java.io.OutputStream;
-import java.net.InetSocketAddress;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -68,7 +67,7 @@ public void handle(HttpExchange httpExchange) throws IOException {
  log.info("showClientBySubsys,subsys:{}", subSystem);
  }
  ClientSessionGroupMapping clientSessionGroupMapping = eventMeshTCPServer.getClientSessionGroupMapping();
- ConcurrentHashMap<InetSocketAddress, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
+ ConcurrentHashMap<String, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
  if (sessionMap != null && !sessionMap.isEmpty()) {
  for (Session session : sessionMap.values()) {
  if (session.getClient().getSubsystem().equals(subSystem)) {

diff --git a/...h-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/ShowClientHandler.java b/...h-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/ShowClientHandler.java
@@ -27,7 +27,6 @@
 
 import java.io.IOException;
 import java.io.OutputStream;
-import java.net.InetSocketAddress;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -61,7 +60,7 @@ public void handle(HttpExchange httpExchange) throws IOException {
 
  HashMap<String, AtomicInteger> statMap = new HashMap<String, AtomicInteger>();
 
- Map<InetSocketAddress, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
+ Map<String, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
  if (!sessionMap.isEmpty()) {
  for (Session session : sessionMap.values()) {
  String key = session.getClient().getSubsystem();

diff --git a/...sh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/TCPClientHandler.java b/...sh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/TCPClientHandler.java
@@ -31,11 +31,12 @@
 import org.apache.eventmesh.runtime.core.protocol.tcp.client.group.ClientSessionGroupMapping;
 import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
 
+import org.apache.commons.lang3.StringUtils;
+
 import java.io.IOException;
 import java.io.OutputStream;
 import java.io.PrintWriter;
 import java.io.StringWriter;
-import java.net.InetSocketAddress;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -84,14 +85,12 @@ void delete(HttpExchange httpExchange) throws IOException {
  try {
  String request = HttpExchangeUtils.streamToString(httpExchange.getRequestBody());
  DeleteTCPClientRequest deleteTCPClientRequest = JsonUtils.parseObject(request, DeleteTCPClientRequest.class);
- String host = deleteTCPClientRequest.getHost();
- int port = deleteTCPClientRequest.getPort();
 
  ClientSessionGroupMapping clientSessionGroupMapping = eventMeshTCPServer.getClientSessionGroupMapping();
- ConcurrentHashMap<InetSocketAddress, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
+ ConcurrentHashMap<String, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
  if (!sessionMap.isEmpty()) {
- for (Map.Entry<InetSocketAddress, Session> entry : sessionMap.entrySet()) {
- if (entry.getKey().getHostString().equals(host) && entry.getKey().getPort() == port) {
+ for (Map.Entry<String, Session> entry : sessionMap.entrySet()) {
+ if (StringUtils.equals(entry.getKey(), deleteTCPClientRequest.toString())) {
  EventMeshTcp2Client.serverGoodby2Client(
  eventMeshTCPServer,
  entry.getValue(),
@@ -100,7 +99,6 @@ void delete(HttpExchange httpExchange) throws IOException {
  }
  }
  }
-
  httpExchange.getResponseHeaders().add("Access-Control-Allow-Origin", "*");
  httpExchange.sendResponseHeaders(200, 0);
  } catch (Exception e) {
@@ -136,7 +134,7 @@ void list(HttpExchange httpExchange) throws IOException {
  try {
  // Get the list of TCP clients
  ClientSessionGroupMapping clientSessionGroupMapping = eventMeshTCPServer.getClientSessionGroupMapping();
- Map<InetSocketAddress, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
+ Map<String, Session> sessionMap = clientSessionGroupMapping.getSessionMap();
  List<GetClientResponse> getClientResponseList = new ArrayList<>();
  for (Session session : sessionMap.values()) {
  UserAgent userAgent = session.getClient();

diff --git a/...time/src/main/java/org/apache/eventmesh/runtime/admin/request/DeleteTCPClientRequest.java b/...time/src/main/java/org/apache/eventmesh/runtime/admin/request/DeleteTCPClientRequest.java
@@ -22,9 +22,11 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
-import lombok.Data;
+import lombok.Getter;
+import lombok.Setter;
 
-@Data
+@Getter
+@Setter
 public class DeleteTCPClientRequest {
 
  private String host;
@@ -40,4 +42,9 @@ public DeleteTCPClientRequest(
  this.host = host;
  this.port = port;
  }
+
+ @Override
+ public String toString() {
+ return host + ":" + port;
+ }
 }
diff --git a/...rg/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java b/...rg/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
@@ -36,7 +36,6 @@
 import org.apache.commons.collections4.MapUtils;
 
 import java.lang.ref.WeakReference;
-import java.net.InetSocketAddress;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
@@ -56,13 +55,11 @@ public class ClientSessionGroupMapping {
 
  private static final Logger SESSION_LOGGER = LoggerFactory.getLogger("sessionLogger");
 
- private final ConcurrentHashMap<InetSocketAddress, Session> sessionTable = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String/*ip:port*/, Session> sessionTable = new ConcurrentHashMap<>(64);
 
- private final ConcurrentHashMap<String /** subsystem eg . 5109 or 5109-1A0 */, ClientGroupWrapper> clientGroupMap =
- new ConcurrentHashMap<String, ClientGroupWrapper>();
+ private final ConcurrentHashMap<String /** subsystem eg . 5109 or 5109-1A0 */, ClientGroupWrapper> clientGroupMap = new ConcurrentHashMap<>(64);
 
- private final ConcurrentHashMap<String /** subsystem eg . 5109 or 5109-1A0 */, Object> lockMap =
- new ConcurrentHashMap<String, Object>();
+ private final ConcurrentHashMap<String /** subsystem eg . 5109 or 5109-1A0 */, Object> lockMap = new ConcurrentHashMap<>(64);
 
  private EventMeshTCPServer eventMeshTCPServer;
 
@@ -83,27 +80,31 @@ public ClientGroupWrapper getClientGroupWrapper(String sysId) {
  }
 
  public Session getSession(ChannelHandlerContext ctx) {
- return getSession((InetSocketAddress) ctx.channel().remoteAddress());
+ return getSession(RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
  }
 
- public Session getSession(InetSocketAddress address) {
+ public Session getSession(String address) {
  return sessionTable.get(address);
  }
 
- public Session createSession(UserAgent user, ChannelHandlerContext ctx) throws Exception {
- InetSocketAddress addr = (InetSocketAddress) ctx.channel().remoteAddress();
- user.setHost(addr.getHostString());
- user.setPort(addr.getPort());
+ public Session getOrCreateSession(UserAgent user, ChannelHandlerContext ctx) throws Exception {
+ String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
+ String[] split = remoteAddress.split(":");
+ if (Objects.isNull(split) || split.length != 2) {
+ throw new Exception("Parse channel remote address ettor");
+ }
+ user.setHost(split[0]);
+ user.setPort(Integer.parseInt(split[1]));
  Session session;
- if (!sessionTable.containsKey(addr)) {
- log.info("createSession client[{}]", RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
+ if (!sessionTable.containsKey(remoteAddress)) {
+ log.info("createSession client[{}]", remoteAddress);
  session = new Session(user, ctx, eventMeshTCPServer.getEventMeshTCPConfiguration());
  initClientGroupWrapper(user, session);
- sessionTable.put(addr, session);
- SESSION_LOGGER.info("session|open|succeed|user={}", user);
+ sessionTable.put(remoteAddress, session);
+ SESSION_LOGGER.info("session|open|succeed|user={},client={}", user, remoteAddress);
  } else {
- session = sessionTable.get(addr);
- SESSION_LOGGER.error("session|open|failed|user={}|msg={}", user, "session has been created!");
+ session = sessionTable.get(remoteAddress);
+ SESSION_LOGGER.warn("session|open|failed|user={}|msg=session has been created!", user);
  }
  return session;
  }
@@ -117,38 +118,36 @@ public void readySession(Session session) throws Exception {
 
  public synchronized void closeSession(ChannelHandlerContext ctx) throws Exception {
 
- InetSocketAddress addr = (InetSocketAddress) ctx.channel().remoteAddress();
- Session session = MapUtils.getObject(sessionTable, addr, null);
+ String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
+ Session session = MapUtils.getObject(sessionTable, remoteAddress, null);
  if (session == null) {
- final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
  log.info("begin to close channel to remote address[{}]", remoteAddress);
  ctx.channel().close().addListener(
  (ChannelFutureListener) future -> log.info("close the connection to remote address[{}] result: {}", remoteAddress,
  future.isSuccess()));
- SESSION_LOGGER.info("session|close|succeed|address={}|msg={}", addr, "no session was found");
+ SESSION_LOGGER.warn("session|close|succeed|address={}|msg=no session was found", remoteAddress);
  return;
  }
-
  closeSession(session);
 
  //remove session from sessionTable
- sessionTable.remove(addr);
+ sessionTable.remove(remoteAddress);
 
  SESSION_LOGGER.info("session|close|succeed|user={}", session.getClient());
  }
 
  private void closeSession(Session session) throws Exception {
  final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(session.getContext().channel());
  if (SessionState.CLOSED == session.getSessionState()) {
- log.info("session has been closed, addr:{}", remoteAddress);
+ log.warn("session has been closed, addr:{}", remoteAddress);
  return;
  }
 
  //session must be synchronized to avoid SessionState be confound, for example adding subscribe when session closing
  synchronized (session) {
 
  if (SessionState.CLOSED == session.getSessionState()) {
- log.info("session has been closed in sync, addr:{}", remoteAddress);
+ log.warn("session has been closed in sync, addr:{}", remoteAddress);
  return;
  }
 
@@ -304,15 +303,12 @@ private void handleUnackMsgsInSession(Session session) {
  continue;
  }
  Session reChooseSession = clientGroupWrapper.getDownstreamDispatchStrategy()
- .select(clientGroupWrapper.getGroup(),
- downStreamMsgContext.event.getSubject(),
- clientGroupWrapper.groupConsumerSessions);
+ .select(clientGroupWrapper.getGroup(), downStreamMsgContext.event.getSubject(), clientGroupWrapper.groupConsumerSessions);
  if (reChooseSession != null) {
  downStreamMsgContext.setSession(reChooseSession);
  reChooseSession.getPusher().unAckMsg(downStreamMsgContext.seq, downStreamMsgContext);
  reChooseSession.downstreamMsg(downStreamMsgContext);
- log.info("rePush msg form unAckMsgs,seq:{},rePushClient:{}", entry.getKey(),
- downStreamMsgContext.getSession().getClient());
+ log.info("rePush msg form unAckMsgs,seq:{},rePushClient:{}", entry.getKey(), downStreamMsgContext.getSession().getClient());
  } else {
  log.warn("select session fail in handleUnackMsgsInSession,seq:{},topic:{}", entry.getKey(),
  downStreamMsgContext.event.getSubject());
@@ -378,7 +374,6 @@ private void initSessionCleaner() {
  private void initDownStreamMsgContextCleaner() {
  eventMeshTCPServer.getScheduler().scheduleAtFixedRate(
  () -> {
-
  //scan non-broadcast msg
  for (Session tmp : sessionTable.values()) {
  for (Map.Entry<String, DownStreamMsgContext> entry : tmp.getPusher().getUnAckMsg().entrySet()) {
@@ -443,30 +438,14 @@ public void shutdown() throws Exception {
  log.info("ClientSessionGroupMapping shutdown......");
  }
 
- public ConcurrentHashMap<InetSocketAddress, Session> getSessionMap() {
+ public ConcurrentHashMap<String, Session> getSessionMap() {
  return sessionTable;
  }
 
  public ConcurrentHashMap<String, ClientGroupWrapper> getClientGroupMap() {
  return clientGroupMap;
  }
 
- public Map<String, Map<String, Integer>> prepareEventMeshClientDistributionData() {
- Map<String, Map<String, Integer>> result = null;
-
- if (!clientGroupMap.isEmpty()) {
- result = new HashMap<>();
- for (Map.Entry<String, ClientGroupWrapper> entry : clientGroupMap.entrySet()) {
- Map<String, Integer> map = new HashMap<>();
- map.put(EventMeshConstants.PURPOSE_SUB, entry.getValue().getGroupConsumerSessions().size());
- map.put(EventMeshConstants.PURPOSE_PUB, entry.getValue().getGroupProducerSessions().size());
- result.put(entry.getKey(), map);
- }
- }
-
- return result;
- }
-
  public Map<String, Map<String, Integer>> prepareProxyClientDistributionData() {
  Map<String, Map<String, Integer>> result = null;