Merge remote-tracking branch 'refs/remotes/public/develop' into feature/YunKuaiChongV160

# Conflicts:
#	jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/YunKuaiChongProtocolMessageProcessor.java

docker/Dockerfile-App

@@ -35,6 +35,9 @@ RUN chmod a+x *.sh && mv start.sh /usr/bin
 
 EXPOSE 8080 8080
 
+ENV APP_LOG_LEVEL=INFO
+ENV PROTOCOLS_LOG_LEVEL=INFO
+
 CMD ["start.sh"]
 
 

docker/Dockerfile-Protocol

@@ -35,6 +35,8 @@ RUN chmod a+x *.sh && mv start.sh /usr/bin
 
 EXPOSE 8081 8081
 
+ENV PROTOCOLS_LOG_LEVEL=INFO
+
 CMD ["start.sh"]
 
 

jcpp-app-bootstrap/src/main/resources/app-service.yml

@@ -27,7 +27,7 @@ spring:
     password: "${SPRING_DATASOURCE_PASSWORD:postgres}"
     hikari:
       leak-detection-threshold: "${SPRING_DATASOURCE_HIKARI_LEAK_DETECTION_THRESHOLD:0}"
-      maximum-pool-size: "${SPRING_DATASOURCE_MAXIMUM_POOL_SIZE:16}"
+      maximum-pool-size: "${SPRING_DATASOURCE_MAXIMUM_POOL_SIZE:64}"
       register-mbeans: "${SPRING_DATASOURCE_HIKARI_REGISTER_MBEANS:false}"
 
 mybatis-plus:
@@ -63,9 +63,10 @@ queue:
   type: "${QUEUE_TYPE:memory}"
   partitions:
     hash_function_name: "${QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" # murmur3_32, murmur3_128 or sha256
-  in_memory:
+  in-memory:
+    queue-capacity: "${QUEUE_IN_MEMORY_QUEUE_CAPACITY:100000}"
     stats:
-      print-interval-ms: "${QUEUE_IN_MEMORY_STATS_PRINT_INTERVAL_MS:60000}"
+      print-interval-ms: "${QUEUE_IN_MEMORY_STATS_PRINT_INTERVAL_MS:10000}"
   kafka:
     bootstrap-servers: "${KAFKA_SERVERS:kafka:9092}"
     ssl:
@@ -93,7 +94,7 @@ queue:
     auto-offset-reset: "${QUEUE_KAFKA_AUTO_OFFSET_RESET:earliest}"
     other-inline: "${QUEUE_KAFKA_OTHER_PROPERTIES:}"
     topic-properties:
-      app: "${QUEUE_KAFKA_APP_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}"
+      app: "${QUEUE_KAFKA_APP_TOPIC_PROPERTIES:retention.ms:86400000;segment.bytes:52428800;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}"
     consumer-stats:
       enabled: "${QUEUE_KAFKA_CONSUMER_STATS_ENABLED:true}"
       print-interval-ms: "${QUEUE_KAFKA_CONSUMER_STATS_MIN_PRINT_INTERVAL_MS:60000}"
@@ -115,8 +116,8 @@ cache:
   type: "${CACHE_TYPE:caffeine}" # caffeine or redis
   specs:
     piles:
-      timeToLiveInMinutes: "${CACHE_SPECS_PILES_TTL:15}"
+      timeToLiveInMinutes: "${CACHE_SPECS_PILES_TTL:1440}"
-      maxSize: "${CACHE_SPECS_PILES_MAX_SIZE:1000}"
+      maxSize: "${CACHE_SPECS_PILES_MAX_SIZE:100000}"
     pileSessions:
       timeToLiveInMinutes: "${CACHE_SPECS_PILE_SESSIONS_TTL:1440}"
       maxSize: "${CACHE_SPECS_PILE_SESSIONS_MAX_SIZE:100000}"
@@ -132,7 +133,7 @@ redis:
     commandTimeout: "${REDIS_CLIENT_COMMAND_TIMEOUT:30000}"
     shutdownTimeout: "${REDIS_CLIENT_SHUTDOWN_TIMEOUT:1000}"
     readTimeout: "${REDIS_CLIENT_READ_TIMEOUT:60000}"
-    usePoolConfig: "${REDIS_CLIENT_USE_POOL_CONFIG:false}"
+    usePoolConfig: "${REDIS_CLIENT_USE_POOL_CONFIG:true}"
   cluster:
     nodes: "${REDIS_NODES:redis-node-0:6379,redis-node-1:6379,redis-node-2:6379,redis-node-3:6379,redis-node-4:6379,redis-node-5:6379}"
     max-redirects: "${REDIS_MAX_REDIRECTS:12}"
@@ -146,7 +147,7 @@ redis:
   password: "${REDIS_PASSWORD:sanbing}"
   pool_config:
     maxTotal: "${REDIS_POOL_CONFIG_MAX_TOTAL:128}"
-    maxIdle: "${REDIS_POOL_CONFIG_MAX_IDLE:128}"
+    maxIdle: "${REDIS_POOL_CONFIG_MAX_IDLE:64}"
     minIdle: "${REDIS_POOL_CONFIG_MIN_IDLE:16}"
     testOnBorrow: "${REDIS_POOL_CONFIG_TEST_ON_BORROW:false}"
     testOnReturn: "${REDIS_POOL_CONFIG_TEST_ON_RETURN:false}"

jcpp-app-bootstrap/src/main/resources/log4j2.xml

@@ -18,7 +18,8 @@
             <PatternLayout pattern="${LOG_PATTERN}"/>
 
             <Policies>
-                <TimeBasedTriggeringPolicy modulate="true" interval="1"/>
+                <SizeBasedTriggeringPolicy size="1G"/>
+                <TimeBasedTriggeringPolicy interval="1" modulate="true"/>
             </Policies>
 
             <DefaultRolloverStrategy>
@@ -45,6 +46,12 @@
             <AppenderRef ref="ROLLING_FILE"/>
         </AsyncLogger>
 
+        <AsyncLogger name="sanbing.jcpp.app" level="${env:APP_LOG_LEVEL:-TRACE}"
+                     additivity="false" includeLocation="false">
+            <AppenderRef ref="CONSOLE"/>
+            <AppenderRef ref="ROLLING_FILE"/>
+        </AsyncLogger>
+
         <AsyncLogger name="sanbing.jcpp.protocol" level="${env:PROTOCOLS_LOG_LEVEL:-TRACE}"
                      additivity="false" includeLocation="false">
             <AppenderRef ref="CONSOLE"/>

jcpp-app-bootstrap/src/test/java/sanbing/jcpp/app/dal/mapper/GunMapperIT.java

@@ -12,9 +12,9 @@ import sanbing.jcpp.app.dal.config.ibatis.enums.GunOptStatusEnum;
 import sanbing.jcpp.app.dal.config.ibatis.enums.GunRunStatusEnum;
 import sanbing.jcpp.app.dal.config.ibatis.enums.OwnerTypeEnum;
 import sanbing.jcpp.app.dal.entity.Gun;
-import sanbing.jcpp.app.dal.entity.Pile;
 import sanbing.jcpp.infrastructure.util.jackson.JacksonUtil;
 
+import java.text.DecimalFormat;
 import java.time.LocalDateTime;
 import java.util.UUID;
 
@@ -42,9 +42,6 @@ class GunMapperIT extends AbstractTestBase {
     @Resource
     GunMapper gunMapper;
 
-    @Resource
-    PileMapper pileMapper;
-
     @Test
     void curdTest() {
         gunMapper.delete(Wrappers.lambdaQuery());
@@ -53,15 +50,13 @@ class GunMapperIT extends AbstractTestBase {
             UUID pileId = NORMAL_PILE_ID[i];
             UUID gunId = NORMAL_GUN_ID[i];
 
-            Pile pile = pileMapper.selectById(pileId);
-
             Gun gun = Gun.builder()
                     .id(gunId)
                     .createdTime(LocalDateTime.now())
                     .additionalInfo(JacksonUtil.newObjectNode())
                     .gunNo("02")
-                    .gunName(pile.getPileName() + "的2号枪")
+                    .gunName(String.format("三丙家的%d号充电桩", i + 1) + "的2号枪")
-                    .gunCode(pile.getPileCode() + "-02")
+                    .gunCode("202312120000" + new DecimalFormat("00").format(i + 1) + "-02")
                     .stationId(NORMAL_STATION_ID)
                     .pileId(pileId)
                     .ownerId(NORMAL_USER_ID)

jcpp-app-bootstrap/src/test/java/sanbing/jcpp/app/dal/mapper/PileMapperIT.java

@@ -17,7 +17,10 @@ import sanbing.jcpp.infrastructure.util.jackson.JacksonUtil;
 
 import java.text.DecimalFormat;
 import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static sanbing.jcpp.app.dal.mapper.StationMapperIT.NORMAL_STATION_ID;
 import static sanbing.jcpp.app.dal.mapper.UserMapperIT.NORMAL_USER_ID;
@@ -68,7 +71,36 @@ class PileMapperIT extends AbstractTestBase {
             pileMapper.insertOrUpdate(pile);
 
             log.info("{}", pileMapper.selectById(pileId));
-
         }
     }
+
+    @Test
+    void generate100KPiles() {
+        AtomicInteger number = new AtomicInteger(1);
+        for (int i = 0; i < 100; i++) {
+            List<Pile> piles = new ArrayList<>();
+            for (int j = 0; j < 1000; j++, number.incrementAndGet()) {
+                Pile pile = Pile.builder()
+                        .id(UUID.randomUUID())
+                        .createdTime(LocalDateTime.now())
+                        .additionalInfo(JacksonUtil.newObjectNode())
+                        .pileName(String.format("三丙家的%d号充电桩", number.get()))
+                        .pileCode("20241015" + new DecimalFormat("000000").format(number.get()))
+                        .protocol("yunkuaichongV150")
+                        .stationId(NORMAL_STATION_ID)
+                        .ownerId(NORMAL_USER_ID)
+                        .ownerType(OwnerTypeEnum.C)
+                        .brand("星星")
+                        .model("10A")
+                        .manufacturer("星星")
+                        .status(PileStatusEnum.IDLE)
+                        .type(PileTypeEnum.AC)
+                        .build();
+                piles.add(pile);
+            }
+            pileMapper.insert(piles);
+        }
+
+        log.info("{}", pileMapper.selectCount(Wrappers.lambdaQuery()));
+    }
 }

jcpp-app/src/main/java/sanbing/jcpp/app/service/cache/session/PileSessionCacheKey.java

@@ -7,35 +7,26 @@ package sanbing.jcpp.app.service.cache.session;
 import lombok.Builder;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
-import lombok.RequiredArgsConstructor;
 
 import java.io.Serializable;
-import java.util.Optional;
-import java.util.UUID;
 
 /**
  * @author baigod
  */
 @Getter
 @EqualsAndHashCode
-@RequiredArgsConstructor
 @Builder
 public class PileSessionCacheKey implements Serializable {
 
-    private final UUID pileId;
     private final String pileCode;
 
-    public PileSessionCacheKey(UUID pileId) {
-        this(pileId, null);
-    }
-
     public PileSessionCacheKey(String pileCode) {
-        this(null, pileCode);
+        this.pileCode = pileCode;
     }
 
     @Override
     public String toString() {
-        return Optional.ofNullable(pileId).map(UUID::toString).orElse(pileCode);
+        return pileCode;
     }
 
 }

jcpp-app/src/main/java/sanbing/jcpp/app/service/impl/DefaultDownlinkCallService.java

@@ -6,6 +6,7 @@ package sanbing.jcpp.app.service.impl;
 
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.http.HttpEntity;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
@@ -44,12 +45,15 @@ public class DefaultDownlinkCallService implements DownlinkCallService {
     @Resource
     TransactionalCache<PileSessionCacheKey, PileSession> pileSessionCache;
 
+    @Value("${cache.type}")
+    private String cacheType;
+
     @Override
     public void sendDownlinkMessage(DownlinkRestMessage.Builder downlinkMessageBuilder, String pileCode) {
-        if (serviceInfoProvider.isMonolith()) {
+        if (serviceInfoProvider.isMonolith() && "caffeine".equalsIgnoreCase(cacheType)) {
 
             downlinkController.onDownlink(downlinkMessageBuilder.build())
-                    .setResultHandler(result -> log.info("下行消息发送完成"));
+                    .setResultHandler(result -> log.debug("下行消息发送完成"));
 
         } else {
             try {
@@ -83,7 +87,7 @@ public class DefaultDownlinkCallService implements DownlinkCallService {
         try {
             ResponseEntity<?> response = downlinkRestTemplate.postForEntity("http://" + nodeWebapiIpPort + "/api/onDownlink",
                     entity, ResponseEntity.class);
-            log.info("下行消息发送成功 {}", response);
+            log.debug("下行消息发送成功 {}", response);
         } catch (RestClientException e) {
             log.error("下行消息发送失败 {}", downlinkRestMessage, e);
             throw new RuntimeException(e);

jcpp-app/src/main/java/sanbing/jcpp/app/service/impl/DefaultPileProtocolService.java

@@ -48,7 +48,7 @@ public class DefaultPileProtocolService implements PileProtocolService {
 
     @Override
     public void pileLogin(UplinkQueueMessage uplinkQueueMessage, Callback callback) {
-        log.info("接收到桩登录事件 {}", uplinkQueueMessage);
+        log.debug("接收到桩登录事件 {}", uplinkQueueMessage);
 
         LoginRequest loginRequest = uplinkQueueMessage.getLoginRequest();
 
@@ -56,7 +56,7 @@ public class DefaultPileProtocolService implements PileProtocolService {
 
         String pileCode = loginRequest.getPileCode();
 
-        log.info("查询到充电桩信息 {}", pile);
+        log.debug("查询到充电桩信息 {}", pile);
 
         // 构造下行回复
         DownlinkRestMessage.Builder downlinkMessageBuilder = createDownlinkMessageBuilder(uplinkQueueMessage, loginRequest.getPileCode());
@@ -108,7 +108,6 @@ public class DefaultPileProtocolService implements PileProtocolService {
         pileSession.setRemoteAddress(remoteAddress);
         pileSession.setNodeId(nodeId);
         pileSession.setNodeWebapiIpPort(nodeWebapiIpPort);
-        pileSessionCache.put(new PileSessionCacheKey(pile.getId()), pileSession);
         pileSessionCache.put(new PileSessionCacheKey(pile.getPileCode()), pileSession);
     }
 

jcpp-app/src/main/java/sanbing/jcpp/app/service/queue/AbstractConsumerService.java

@@ -31,7 +31,7 @@ public abstract class AbstractConsumerService extends JCPPApplicationEventListen
     protected ScheduledExecutorService scheduler;
 
     public void init(String prefix) {
-        this.consumersExecutor = Executors.newCachedThreadPool(JCPPThreadFactory.forName(prefix + "-consumer"));
+        this.consumersExecutor = JCPPExecutors.newVirtualThreadPool(prefix + "-consumer-virtual");
         this.mgmtExecutor = JCPPExecutors.newWorkStealingPool(getMgmtThreadPoolSize(), prefix + "-mgmt");
         this.scheduler = Executors.newSingleThreadScheduledExecutor(JCPPThreadFactory.forName(prefix + "-consumer-scheduler"));
     }

jcpp-app/src/main/java/sanbing/jcpp/app/service/queue/consumer/ProtocolUplinkConsumerService.java

@@ -35,16 +35,12 @@ import sanbing.jcpp.infrastructure.util.trace.TracerRunnable;
 import sanbing.jcpp.proto.gen.ProtocolProto.UplinkQueueMessage;
 
 import java.util.List;
-import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.*;
 import java.util.stream.Collectors;
 
-import static sanbing.jcpp.infrastructure.queue.common.QueueConstants.MSG_MD_PREFIX;
+import static sanbing.jcpp.infrastructure.queue.common.QueueConstants.*;
-import static sanbing.jcpp.infrastructure.queue.common.QueueConstants.MSG_MD_TS;
-import static sanbing.jcpp.infrastructure.util.trace.TracerContextUtil.JCPP_TRACER_ID;
-import static sanbing.jcpp.infrastructure.util.trace.TracerContextUtil.JCPP_TRACER_ORIGIN;
 
 
 /**
@@ -101,6 +97,7 @@ public class ProtocolUplinkConsumerService extends AbstractConsumerService imple
     }
 
 
+    @Override
     @PreDestroy
     public void destroy() {
         super.destroy();
@@ -132,80 +129,109 @@ public class ProtocolUplinkConsumerService extends AbstractConsumerService imple
                 processingTimeoutLatch, pendingMap, new ConcurrentHashMap<>());
         PendingMsgHolder pendingMsgHolder = new PendingMsgHolder();
         Future<?> packSubmitFuture = consumersExecutor.submit(new TracerRunnable(() ->
-                orderedMsgList.forEach((element) -> {
+                orderedMsgList.forEach(element -> {
+
                     UUID id = element.getUuid();
+
                     ProtoQueueMsg<UplinkQueueMessage> msg = element.getMsg();
+
                     tracer(msg);
-                    log.trace("[{}] Creating main callback for message: {}", id, msg.getValue());
+
+                    log.trace("[{}] Creating PackCallback for message: {}", id, msg.getValue());
+
                     Callback callback = new PackCallback<>(id, ctx);
+
                     try {
                         UplinkQueueMessage uplinkQueueMsg = msg.getValue();
-                        pendingMsgHolder.setUplinkQueueMessage(uplinkQueueMsg);
-                        if (uplinkQueueMsg.hasLoginRequest()) {
-                            pileProtocolService.pileLogin(uplinkQueueMsg, callback);
-                        } else if (uplinkQueueMsg.hasHeartBeatRequest()) {
-                            pileProtocolService.heartBeat(uplinkQueueMsg, callback);
-                        } else if (uplinkQueueMsg.hasVerifyPricingRequest()) {
-                            pileProtocolService.verifyPricing(uplinkQueueMsg, callback);
-                        } else if (uplinkQueueMsg.hasQueryPricingRequest()) {
-                            pileProtocolService.queryPricing(uplinkQueueMsg, callback);
-                        } else if (uplinkQueueMsg.hasGunRunStatusProto()) {
-                            pileProtocolService.postGunRunStatus(uplinkQueueMsg, callback);
-                        } else if (uplinkQueueMsg.hasChargingProgressProto()) {
-                            pileProtocolService.postChargingProgress(uplinkQueueMsg, callback);
-                        } else if (uplinkQueueMsg.hasSetPricingResponse()) {
-                            pileProtocolService.onSetPricingResponse(uplinkQueueMsg, callback);
-                        } else if (uplinkQueueMsg.hasRemoteStartChargingResponse()) {
-                            pileProtocolService.onRemoteStartChargingResponse(uplinkQueueMsg, callback);
-                        }  else if (uplinkQueueMsg.hasRemoteStopChargingResponse()) {
-                            pileProtocolService.onRemoteStopChargingResponse(uplinkQueueMsg, callback);
-                        } else if(uplinkQueueMsg.hasTransactionRecord()){
-                            pileProtocolService.onTransactionRecord(uplinkQueueMsg, callback);
-                        }else {
-                            callback.onSuccess();
-                        }
 
                         if (statsEnabled) {
                             stats.log(uplinkQueueMsg);
                         }
+
+                        pendingMsgHolder.setUplinkQueueMessage(uplinkQueueMsg);
+
+                        if (uplinkQueueMsg.hasLoginRequest()) {
+
+                            pileProtocolService.pileLogin(uplinkQueueMsg, callback);
+
+                        } else if (uplinkQueueMsg.hasHeartBeatRequest()) {
+
+                            pileProtocolService.heartBeat(uplinkQueueMsg, callback);
+
+                        } else if (uplinkQueueMsg.hasVerifyPricingRequest()) {
+
+                            pileProtocolService.verifyPricing(uplinkQueueMsg, callback);
+
+                        } else if (uplinkQueueMsg.hasQueryPricingRequest()) {
+
+                            pileProtocolService.queryPricing(uplinkQueueMsg, callback);
+
+                        } else if (uplinkQueueMsg.hasGunRunStatusProto()) {
+
+                            pileProtocolService.postGunRunStatus(uplinkQueueMsg, callback);
+
+                        } else if (uplinkQueueMsg.hasChargingProgressProto()) {
+
+                            pileProtocolService.postChargingProgress(uplinkQueueMsg, callback);
+
+                        } else if (uplinkQueueMsg.hasSetPricingResponse()) {
+
+                            pileProtocolService.onSetPricingResponse(uplinkQueueMsg, callback);
+
+                        } else if (uplinkQueueMsg.hasRemoteStartChargingResponse()) {
+
+                            pileProtocolService.onRemoteStartChargingResponse(uplinkQueueMsg, callback);
+
+                        } else if (uplinkQueueMsg.hasRemoteStopChargingResponse()) {
+
+                            pileProtocolService.onRemoteStopChargingResponse(uplinkQueueMsg, callback);
+
+                        } else if (uplinkQueueMsg.hasTransactionRecord()) {
+
+                            pileProtocolService.onTransactionRecord(uplinkQueueMsg, callback);
+
+                        } else {
+
+                            callback.onSuccess();
+                        }
+
                     } catch (Throwable e) {
+
                         log.warn("[{}] Failed to process message: {}", id, msg, e);
+
                         callback.onFailure(e);
                     }
                 }))
         );
-        if (!processingTimeoutLatch.await(packProcessingTimeout, TimeUnit.MILLISECONDS)) {
+
+        if (!ctx.await(packProcessingTimeout, TimeUnit.MILLISECONDS)) {
+
             if (!packSubmitFuture.isDone()) {
+
                 packSubmitFuture.cancel(true);
+
                 UplinkQueueMessage lastSubmitMsg = pendingMsgHolder.getUplinkQueueMessage();
+
                 log.warn("Timeout to process message: {}", lastSubmitMsg);
             }
+
             if (log.isDebugEnabled()) {
-                ctx.getAckMap().forEach((id, msg) -> log.debug("[{}] Timeout to process message: {}", id, msg.getValue()));
+
+                ctx.getAckMap().forEach((id, msg) -> log.info("[{}] Timeout to process message: {}", id, msg.getValue()));
+
             }
+
             ctx.getFailedMap().forEach((id, msg) -> log.warn("[{}] Failed to process message: {}", id, msg.getValue()));
         }
+
         consumer.commit();
     }
 
     private void tracer(ProtoQueueMsg<UplinkQueueMessage> msg) {
-        Optional.ofNullable(msg.getHeaders().get(MSG_MD_PREFIX + JCPP_TRACER_ID))
-                .map(tracerId -> {
-                    String origin = null;
-                    byte[] tracerOrigin = msg.getHeaders().get(MSG_MD_PREFIX + JCPP_TRACER_ORIGIN);
-                    if (tracerOrigin != null) {
-                        origin = ByteUtil.bytesToString(tracerOrigin);
-                    }
 
-                    long ts = System.currentTimeMillis();
+        TracerContextUtil.newTracer(ByteUtil.bytesToString(msg.getHeaders().get(MSG_MD_TRACER_ID)),
-                    byte[] tracerTs = msg.getHeaders().get(MSG_MD_PREFIX + MSG_MD_TS);
+                ByteUtil.bytesToString(msg.getHeaders().get(MSG_MD_TRACER_ORIGIN)),
-                    if (tracerTs != null) {
+                ByteUtil.bytesToLong(msg.getHeaders().get(MSG_MD_TRACER_TS)));
-                        ts = ByteUtil.bytesToLong(tracerTs);
-                    }
-
-                    return TracerContextUtil.newTracer(ByteUtil.bytesToString(tracerId), origin, ts);
-                })
-                .orElseGet(TracerContextUtil::newTracer);
 
         MDCUtils.recordTracer();
     }
@@ -231,6 +257,6 @@ public class ProtocolUplinkConsumerService extends AbstractConsumerService imple
     @Setter
     @Getter
     private static class PendingMsgHolder {
-        private volatile UplinkQueueMessage uplinkQueueMessage;
+        private UplinkQueueMessage uplinkQueueMessage;
     }
 }

jcpp-infrastructure-queue/src/main/java/sanbing/jcpp/infrastructure/queue/common/QueueConstants.java

@@ -4,6 +4,8 @@
  */
 package sanbing.jcpp.infrastructure.queue.common;
 
+import static sanbing.jcpp.infrastructure.util.trace.TracerContextUtil.*;
+
 /**
  * @author baigod
  */
@@ -11,5 +13,10 @@ public final class QueueConstants {
 
     public static final String MSG_MD_PREFIX = "jcpp_";
 
-    public static final String MSG_MD_TS = "ts";
+    public static final String MSG_MD_TRACER_ID = MSG_MD_PREFIX + JCPP_TRACER_ID;
+
+    public static final String MSG_MD_TRACER_ORIGIN = MSG_MD_PREFIX + JCPP_TRACER_ORIGIN;
+
+    public static final String MSG_MD_TRACER_TS = MSG_MD_PREFIX + JCPP_TRACER_TS;
+
 }

jcpp-infrastructure-queue/src/main/java/sanbing/jcpp/infrastructure/queue/memory/DefaultInMemoryStorage.java

@@ -5,6 +5,7 @@
 package sanbing.jcpp.infrastructure.queue.memory;
 
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
 import sanbing.jcpp.infrastructure.queue.QueueMsg;
 
@@ -18,15 +19,16 @@ import java.util.concurrent.LinkedBlockingQueue;
 public final class DefaultInMemoryStorage implements InMemoryStorage {
     private final ConcurrentHashMap<String, BlockingQueue<QueueMsg>> storage = new ConcurrentHashMap<>();
 
+    @Value("${queue.in-memory.queue-capacity:100000}")
+    private int queueCapacity;
+
     @Override
     public void printStats() {
-        if (log.isDebugEnabled()) {
+        storage.forEach((topic, queue) -> {
-            storage.forEach((topic, queue) -> {
+            if (!queue.isEmpty()) {
-                if (!queue.isEmpty()) {
+                log.info("[{}] Queue Size [{}]", topic, queue.size());
-                    log.debug("[{}] Queue Size [{}]", topic, queue.size());
+            }
-                }
+        });
-            });
-        }
     }
 
     @Override
@@ -41,11 +43,11 @@ public final class DefaultInMemoryStorage implements InMemoryStorage {
 
     @Override
     public boolean put(String topic, QueueMsg msg) {
-        return storage.computeIfAbsent(topic, (t) -> new LinkedBlockingQueue<>()).add(msg);
+        return storage.computeIfAbsent(topic, t -> new LinkedBlockingQueue<>(queueCapacity)).add(msg);
     }
 
     @Override
-    public  List<QueueMsg> get(String topic) throws InterruptedException {
+    public List<QueueMsg> get(String topic) throws InterruptedException {
         final BlockingQueue<QueueMsg> queue = storage.get(topic);
         if (queue != null) {
             final QueueMsg firstMsg = queue.poll();

jcpp-infrastructure-queue/src/main/java/sanbing/jcpp/infrastructure/queue/provider/InMemoryAppQueueFactory.java

@@ -40,7 +40,7 @@ public class InMemoryAppQueueFactory implements AppQueueFactory {
         return new InMemoryQueueProducer<>(storage, topic);
     }
 
-    @Scheduled(fixedRateString = "${queue.in_memory.stats.print-interval-ms:60000}")
+    @Scheduled(fixedRateString = "${queue.in-memory.stats.print-interval-ms:60000}")
     private void printInMemoryStats() {
         storage.printStats();
     }

jcpp-infrastructure-queue/src/main/java/sanbing/jcpp/infrastructure/queue/provider/KafkaAppQueueFactory.java

@@ -41,7 +41,6 @@ public class KafkaAppQueueFactory implements AppQueueFactory {
         this.appAdmin = new KafkaAdmin(kafkaSettings, kafkaTopicConfigs.getAppConfigs());
     }
 
-
     @Override
     public QueueConsumer<ProtoQueueMsg<UplinkQueueMessage>> createProtocolUplinkMsgConsumer() {
         KafkaConsumerTemplate.KafkaConsumerTemplateBuilder<ProtoQueueMsg<UplinkQueueMessage>> consumerBuilder = KafkaConsumerTemplate.builder();

jcpp-infrastructure-util/src/main/java/sanbing/jcpp/infrastructure/util/async/JCPPVirtualThreadFactory.java

@@ -4,8 +4,6 @@
  */
 package sanbing.jcpp.infrastructure.util.async;
 
-import sanbing.jcpp.infrastructure.util.trace.TracerRunnable;
-
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -19,6 +17,6 @@ public class JCPPVirtualThreadFactory implements ThreadFactory {
 
     @Override
     public Thread newThread(Runnable r) {
-        return Thread.ofVirtual().name(namePrefix + "-" + threadNumber.getAndIncrement()).unstarted(new TracerRunnable(r));
+        return Thread.ofVirtual().name(namePrefix + "-" + threadNumber.getAndIncrement()).unstarted(r);
     }
 }

jcpp-infrastructure-util/src/main/java/sanbing/jcpp/infrastructure/util/mdc/MDCUtils.java

@@ -34,7 +34,6 @@ public class MDCUtils {
         }
 
         return tracer.getTraceId();
-
     }
 
     public static void cleanTracer() {

jcpp-infrastructure-util/src/main/java/sanbing/jcpp/infrastructure/util/trace/TracerContextUtil.java

@@ -40,7 +40,6 @@ public class TracerContextUtil {
             tracer = new Tracer(traceId, origin, ts);
         }
 
-
         TRACE_ID_CONTAINER.set(tracer);
 
         return tracer;

jcpp-infrastructure-util/src/test/java/sanbing/jcpp/infrastructure/util/async/JCPPExecutorsTest.java

@@ -0,0 +1,30 @@
+/**
+ * 抖音关注：程序员三丙
+ * 知识星球：https://t.zsxq.com/j9b21
+ */
+package sanbing.jcpp.infrastructure.util.async;
+
+import org.junit.jupiter.api.Test;
+import org.slf4j.MDC;
+import sanbing.jcpp.infrastructure.util.mdc.MDCUtils;
+import sanbing.jcpp.infrastructure.util.trace.TracerContextUtil;
+import sanbing.jcpp.infrastructure.util.trace.TracerRunnable;
+
+import java.util.concurrent.ExecutorService;
+
+class JCPPExecutorsTest {
+
+    @Test
+    void newVirtualThreadPool() {
+        ExecutorService executorService = JCPPExecutors.newVirtualThreadPool("test-consumer-virtual");
+
+        TracerContextUtil.newTracer();
+        MDCUtils.recordTracer();
+
+        System.out.println(MDC.get("TRACE_ID"));
+
+        executorService.submit(new TracerRunnable(() -> {
+            System.out.println(MDC.get("TRACE_ID"));
+        }));
+    }
+}

jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/adapter/DownlinkController.java

@@ -38,7 +38,7 @@ public class DownlinkController {
 
     @PostMapping(value = "/onDownlink", consumes = "application/x-protobuf", produces = "application/x-protobuf")
     public DeferredResult<ResponseEntity<String>> onDownlink(@RequestBody DownlinkRestMessage downlinkMsg) {
-        log.info("收到REST下行请求 {}", downlinkMsg);
+        log.debug("收到REST下行请求 {}", downlinkMsg);
 
         final DeferredResult<ResponseEntity<String>> response = new DeferredResult<>(onDownlinkTimeout,
                 ResponseEntity.status(HttpStatus.REQUEST_TIMEOUT).build());
@@ -55,7 +55,7 @@ public class DownlinkController {
                 response.setResult(ResponseEntity.status(HttpStatus.OK).build());
             } else {
 
-                log.warn("下发报文时Session未找到 sessionId: {}", protocolSessionId);
+                log.info("下发报文时Session未找到 sessionId: {}", protocolSessionId);
 
                 response.setResult(ResponseEntity.status(HttpStatus.NOT_FOUND).body("Protocol Session not found for ID:" + protocolSessionId));
             }

jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/domain/ProtocolUplinkMsg.java

@@ -13,8 +13,8 @@ public record ProtocolUplinkMsg<T>(SocketAddress address, UUID id, T data, int s
 
     @Override
     public String toString() {
-        if (data instanceof byte[]) {
+        if (data instanceof byte[] bytes) {
-            return ByteBufUtil.hexDump((byte[]) data);
+            return ByteBufUtil.hexDump(bytes);
         } else {
             return data.toString();
         }

jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/forwarder/Forwarder.java

@@ -25,10 +25,7 @@ import sanbing.jcpp.proto.gen.ProtocolProto.UplinkQueueMessage;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiConsumer;
 
-import static sanbing.jcpp.infrastructure.queue.common.QueueConstants.MSG_MD_PREFIX;
+import static sanbing.jcpp.infrastructure.queue.common.QueueConstants.*;
-import static sanbing.jcpp.infrastructure.queue.common.QueueConstants.MSG_MD_TS;
-import static sanbing.jcpp.infrastructure.util.trace.TracerContextUtil.JCPP_TRACER_ID;
-import static sanbing.jcpp.infrastructure.util.trace.TracerContextUtil.JCPP_TRACER_ORIGIN;
 
 /**
  * @author baigod
@@ -51,7 +48,7 @@ public abstract class Forwarder {
     protected final boolean isMonolith;
     protected QueueProducer<ProtoQueueMsg<UplinkQueueMessage>> producer;
 
-    public Forwarder(String protocolName, StatsFactory statsFactory, PartitionProvider partitionProvider, ServiceInfoProvider serviceInfoProvider) {
+    protected Forwarder(String protocolName, StatsFactory statsFactory, PartitionProvider partitionProvider, ServiceInfoProvider serviceInfoProvider) {
         this.protocolName = protocolName;
         this.partitionProvider = partitionProvider;
         this.serviceInfoProvider = serviceInfoProvider;
@@ -66,12 +63,13 @@ public abstract class Forwarder {
     public abstract void destroy();
 
     protected void jcppForward(String topic, String key, UplinkQueueMessage msg, BiConsumer<Boolean, ObjectNode> consumer) {
+        forwarderMessagesStats.incrementTotal();
         QueueMsgHeaders headers = new DefaultQueueMsgHeaders();
 
         Tracer currentTracer = TracerContextUtil.getCurrentTracer();
-        headers.put(MSG_MD_PREFIX + JCPP_TRACER_ID, ByteUtil.stringToBytes(currentTracer.getTraceId()));
+        headers.put(MSG_MD_TRACER_ID, ByteUtil.stringToBytes(currentTracer.getTraceId()));
-        headers.put(MSG_MD_PREFIX + JCPP_TRACER_ORIGIN, ByteUtil.stringToBytes(currentTracer.getOrigin()));
+        headers.put(MSG_MD_TRACER_ORIGIN, ByteUtil.stringToBytes(currentTracer.getOrigin()));
-        headers.put(MSG_MD_PREFIX + MSG_MD_TS, ByteUtil.longToBytes(currentTracer.getTracerTs()));
+        headers.put(MSG_MD_TRACER_TS, ByteUtil.longToBytes(currentTracer.getTracerTs()));
 
         TopicPartitionInfo tpi = partitionProvider.resolve(ServiceType.APP, topic, key);
         producer.send(tpi, new ProtoQueueMsg<>(key, msg, headers), new QueueCallback() {
@@ -80,11 +78,13 @@ public abstract class Forwarder {
 
                 TracerContextUtil.newTracer(currentTracer.getTraceId(), currentTracer.getOrigin(), currentTracer.getTracerTs());
                 MDCUtils.recordTracer();
+
                 log.trace("单体消息转发成功 key:{}", key);
 
                 if (consumer != null) {
                     consumer.accept(true, JacksonUtil.newObjectNode());
                 }
+                forwarderMessagesStats.incrementSuccessful();
             }
 
             @Override
@@ -92,6 +92,7 @@ public abstract class Forwarder {
 
                 TracerContextUtil.newTracer(currentTracer.getTraceId(), currentTracer.getOrigin(), currentTracer.getTracerTs());
                 MDCUtils.recordTracer();
+
                 log.warn("单体消息转发异常", t);
 
                 if (consumer != null) {
@@ -99,6 +100,7 @@ public abstract class Forwarder {
                     objectNode.put(ERROR, t.getClass() + ": " + t.getMessage());
                     consumer.accept(true, objectNode);
                 }
+                forwarderMessagesStats.incrementFailed();
             }
         });
     }

jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/forwarder/KafkaForwarder.java

@@ -36,10 +36,7 @@ import java.util.Properties;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiConsumer;
 
-import static sanbing.jcpp.infrastructure.queue.common.QueueConstants.MSG_MD_PREFIX;
+import static sanbing.jcpp.infrastructure.queue.common.QueueConstants.*;
-import static sanbing.jcpp.infrastructure.queue.common.QueueConstants.MSG_MD_TS;
-import static sanbing.jcpp.infrastructure.util.trace.TracerContextUtil.JCPP_TRACER_ID;
-import static sanbing.jcpp.infrastructure.util.trace.TracerContextUtil.JCPP_TRACER_ORIGIN;
 
 /**
  * @author baigod
@@ -150,12 +147,13 @@ public class KafkaForwarder extends Forwarder {
     }
 
     private void kafkaForward(String topic, String key, UplinkQueueMessage msg, BiConsumer<Boolean, ObjectNode> consumer) throws InvalidProtocolBufferException {
+        forwarderMessagesStats.incrementTotal();
         Headers headers = new RecordHeaders();
 
         Tracer currentTracer = TracerContextUtil.getCurrentTracer();
-        headers.add(new RecordHeader(MSG_MD_PREFIX + JCPP_TRACER_ID, ByteUtil.stringToBytes(currentTracer.getTraceId())));
+        headers.add(new RecordHeader(MSG_MD_TRACER_ID, ByteUtil.stringToBytes(currentTracer.getTraceId())));
-        headers.add(new RecordHeader(MSG_MD_PREFIX + JCPP_TRACER_ORIGIN, ByteUtil.stringToBytes(currentTracer.getOrigin())));
+        headers.add(new RecordHeader(MSG_MD_TRACER_ORIGIN, ByteUtil.stringToBytes(currentTracer.getOrigin())));
-        headers.add(new RecordHeader(MSG_MD_PREFIX + MSG_MD_TS, ByteUtil.longToBytes(currentTracer.getTracerTs())));
+        headers.add(new RecordHeader(MSG_MD_TRACER_TS, ByteUtil.longToBytes(currentTracer.getTracerTs())));
 
         if (kafkaCfg.getEncoder() == KafkaCfg.EncoderType.json) {
 
@@ -177,6 +175,7 @@ public class KafkaForwarder extends Forwarder {
     private void logAndDoConsumer(BiConsumer<Boolean, ObjectNode> consumer, RecordMetadata metadata, Exception e, Tracer currentTracer) {
         TracerContextUtil.newTracer(currentTracer.getTraceId(), currentTracer.getOrigin(), currentTracer.getTracerTs());
         MDCUtils.recordTracer();
+
         log.debug("Kafka 消息转发完成, success:{}", e == null);
 
         if (consumer != null) {
@@ -196,6 +195,9 @@ public class KafkaForwarder extends Forwarder {
 
         if (e != null) {
             objectNode.put(ERROR, e.getClass() + ": " + e.getMessage());
+            forwarderMessagesStats.incrementFailed();
+        } else {
+            forwarderMessagesStats.incrementSuccessful();
         }
 
         consumer.accept(e == null, objectNode);

jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/listener/tcp/TcpListener.java

@@ -41,8 +41,8 @@ public class TcpListener extends Listener {
     }
 
     private void tcpServerBootstrap(TcpCfg tcpCfg) throws InterruptedException {
-        bossGroup = new NioEventLoopGroup(tcpCfg.getBossGroupThreadCount(), JCPPThreadFactory.forName("tcp-boss"));
+        bossGroup = new NioEventLoopGroup(tcpCfg.getBossGroupThreadCount(), JCPPThreadFactory.forName("tcp-boss-%d"));
-        workerGroup = new NioEventLoopGroup(tcpCfg.getWorkerGroupThreadCount(), JCPPThreadFactory.forName("tcp-worker"));
+        workerGroup = new NioEventLoopGroup(tcpCfg.getWorkerGroupThreadCount(), JCPPThreadFactory.forName("tcp-worker-%d"));
 
         ChannelHandlerInitializer<SocketChannel> channelHandler = ChannelHandlerInitializer.createTcpChannelHandler(tcpCfg, parameter);
 

jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/listener/tcp/handler/ConnectionLimitHandler.java

@@ -42,14 +42,14 @@ public class ConnectionLimitHandler extends ChannelInboundHandlerAdapter {
     public void channelActive(ChannelHandlerContext ctx) throws Exception {
         super.channelActive(ctx);
         channelGroup.add(ctx.channel());
-        log.info("[{}]{} channelActive 当前连接数管道数 {} / {}",protocolName, ctx.channel(), channelGroup.size(), maxConnections);
+        log.info("[{}]{} channelActive 当前连接数 {} / {}",protocolName, ctx.channel(), channelGroup.size(), maxConnections);
     }
 
     @Override
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
         super.channelInactive(ctx);
         channelGroup.remove(ctx.channel());
-        log.info("[{}]{} channelInactive 当前连接数管道数 {} / {}",protocolName, ctx.channel(), channelGroup.size(), maxConnections);
+        log.info("[{}]{} channelInactive 当前连接数 {} / {}",protocolName, ctx.channel(), channelGroup.size(), maxConnections);
     }
 
     @Override

jcpp-protocol-bootstrap/src/main/resources/log4j2.xml

@@ -18,7 +18,8 @@
             <PatternLayout pattern="${LOG_PATTERN}"/>
 
             <Policies>
-                <TimeBasedTriggeringPolicy modulate="true" interval="1"/>
+                <SizeBasedTriggeringPolicy size="1G"/>
+                <TimeBasedTriggeringPolicy interval="1" modulate="true"/>
             </Policies>
 
             <DefaultRolloverStrategy>

jcpp-protocol-bootstrap/src/main/resources/protocol-service.yml

@@ -135,9 +135,6 @@ queue:
   type: "${QUEUE_TYPE:kafka}"
   partitions:
     hash_function_name: "${QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" # murmur3_32, murmur3_128 or sha256
-  in_memory:
-    stats:
-      print-interval-ms: "${QUEUE_IN_MEMORY_STATS_PRINT_INTERVAL_MS:60000}"
   kafka:
     bootstrap-servers: "${KAFKA_SERVERS:kafka:9092}"
     ssl:
@@ -165,7 +162,7 @@ queue:
     auto-offset-reset: "${QUEUE_KAFKA_AUTO_OFFSET_RESET:earliest}"
     other-inline: "${QUEUE_KAFKA_OTHER_PROPERTIES:}"
     topic-properties:
-      app: "${QUEUE_KAFKA_APP_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:52428800;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}"
+      app: "${QUEUE_KAFKA_APP_TOPIC_PROPERTIES:retention.ms:86400000;segment.bytes:52428800;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}"
     consumer-stats:
       enabled: "${QUEUE_KAFKA_CONSUMER_STATS_ENABLED:true}"
       print-interval-ms: "${QUEUE_KAFKA_CONSUMER_STATS_MIN_PRINT_INTERVAL_MS:60000}"

jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/YunKuaiChongProtocolMessageProcessor.java

@@ -138,7 +138,7 @@ public class YunKuaiChongProtocolMessageProcessor extends ProtocolMessageProcess
             csTemp.writeBytes(byCheckSum);
             checkSum = csTemp.readUnsignedShortLE();
             checkResult = checkCrcSum(checkData, checkSum);
-            log.info("云快充检验和 第二次检查: checkResult:{}, checkSum:{}", checkResult, checkSum);
+            log.debug("云快充检验和 第二次检查: checkResult:{}, checkSum:{}", checkResult, checkSum);
         }
 
         if (Boolean.FALSE.equals(checkResult.getFirst())) {

jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/v150/YunKuaiChongV150LoginAckDLCmd.java

@@ -4,6 +4,7 @@
  */
 package sanbing.jcpp.protocol.yunkuaichong.v150;
 
+import cn.hutool.core.util.RandomUtil;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import lombok.extern.slf4j.Slf4j;
@@ -42,7 +43,7 @@ public class YunKuaiChongV150LoginAckDLCmd extends YunKuaiChongDownlinkCmdExe {
 
     @Override
     public void execute(TcpSession tcpSession, YunKuaiChongDwonlinkMessage yunKuaiChongDwonlinkMessage, ProtocolContext ctx) {
-        log.info("{} 云快充1.5.0登录认证应答", tcpSession);
+        log.debug("{} 云快充1.5.0登录认证应答", tcpSession);
 
         if (!yunKuaiChongDwonlinkMessage.getMsg().hasLoginResponse()) {
             return;
@@ -95,7 +96,7 @@ public class YunKuaiChongV150LoginAckDLCmd extends YunKuaiChongDownlinkCmdExe {
                     log.info("{} 云快充1.5.0开始注册定时对时任务", tcpSession);
                     return PROTOCOL_SESSION_SCHEDULED.scheduleAtFixedRate(() ->
                                     syncTime(tcpSession, pileCodeBytes, requestData),
-                            0, 8, TimeUnit.HOURS);
+                            0, RandomUtil.randomInt(420, 480), TimeUnit.MINUTES);
                 }
         );
     }

jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/v150/YunKuaiChongV150LoginULCmd.java

@@ -29,7 +29,7 @@ public class YunKuaiChongV150LoginULCmd extends YunKuaiChongUplinkCmdExe {
 
     @Override
     public void execute(TcpSession tcpSession, YunKuaiChongUplinkMessage yunKuaiChongUplinkMessage, ProtocolContext ctx) {
-        log.info("{} 云快充1.5.0登录认证请求", tcpSession);
+        log.debug("{} 云快充1.5.0登录认证请求", tcpSession);
         ByteBuf byteBuf = Unpooled.copiedBuffer(yunKuaiChongUplinkMessage.getMsgBody());
 
         ObjectNode additionalInfo = JacksonUtil.newObjectNode();