autumn/JChargePointProtocol
1
0
Fork 0
mirror of https://gitee.com/san-bing/JChargePointProtocol synced 2026-05-04 18:09:54 +08:00
Code Issues Packages Projects Releases Wiki Activity

protocolSession的缓存改为同步caffeine，因为没有IO操作

Browse Source
This commit is contained in:
三丙
2024-10-09 10:08:32 +08:00
parent a1db728be5
commit c3295ce01c
11 changed files with 75 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
jcpp-app-bootstrap/src/main/resources/app-service.yml
View File

@@ -210,5 +210,5 @@ service:
thread-pool:
sharding:
hash_function_name: "${THREAD_POOL_SHARDING_HASH_FUNCTION_NAME:murmur3_128}" # murmur3_32, murmur3_128 or sha256
parallelism: "${THREAD_POOL_SHARDING_PARALLELISM:128}"
parallelism: "${THREAD_POOL_SHARDING_PARALLELISM:8}"
stats-print-interval-ms: "${THREAD_POOL_SHARDING_STATS_PRINT_INTERVAL_MS:10000}"

23
jcpp-infrastructure-util/src/main/java/sanbing/jcpp/infrastructure/util/config/ScheduledTaskConfig.java Normal file
View File

@@ -0,0 +1,23 @@
/**
* 抖音关注：程序员三丙
* 知识星球：https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.infrastructure.util.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@Configuration
public class ScheduledTaskConfig {
@Bean
public TaskScheduler taskScheduler() {
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.setPoolSize(Runtime.getRuntime().availableProcessors());
scheduler.setThreadNamePrefix("scheduled-task-");
scheduler.initialize();
return scheduler;
}
}

21
jcpp-infrastructure-util/src/main/java/sanbing/jcpp/infrastructure/util/config/ShardingThreadPool.java
View File

@@ -33,12 +33,12 @@ public class ShardingThreadPool {
@Value("${thread-pool.sharding.hash_function_name:murmur3_128}")
private String hashFunctionName;
@Value("${thread-pool.sharding.parallelism:128}")
@Value("${thread-pool.sharding.parallelism:8}")
private int parallelism;
private HashFunction hashFunction;
private final Map<Integer, ExecutorService> EXECUTOR_SERVICE_MAP = new ConcurrentHashMap<>(128);
private final Map<Integer, ExecutorService> EXECUTOR_SERVICE_MAP = new ConcurrentHashMap<>(8);
@PostConstruct
public void init() {
@@ -59,13 +59,16 @@ public class ShardingThreadPool {
ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) v;
log.info("分区 {}/{} 的线程池中剩余 {} 条待执行任务,当前正在执行的线程数 {}, 已完成任务 {} / {}",
k,
EXECUTOR_SERVICE_MAP.size(),
threadPoolExecutor.getQueue().size(),
threadPoolExecutor.getActiveCount(),
threadPoolExecutor.getCompletedTaskCount(),
threadPoolExecutor.getTaskCount());
int size = threadPoolExecutor.getQueue().size();
if (size > 1) {
log.info("分区 {} 的线程池中剩余 {} 条待执行任务,当前正在执行的线程数 {}, 已完成任务 {} / {}",
k,
size,
threadPoolExecutor.getActiveCount(),
threadPoolExecutor.getCompletedTaskCount(),
threadPoolExecutor.getTaskCount());
}
});
}

4
jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/ProtocolMessageProcessor.java
View File

@@ -47,7 +47,7 @@ public abstract class ProtocolMessageProcessor {
}));
}
protected abstract void uplinkHandle(ListenerToHandlerMsg listenerToHandlerMsg) throws Exception;
protected abstract void uplinkHandle(ListenerToHandlerMsg listenerToHandlerMsg);
public void downlinkHandle(SessionToHandlerMsg sessionToHandlerMsg, MessagesStats downlinkMsgStats) throws DownlinkException {
try {
@@ -62,5 +62,5 @@ public abstract class ProtocolMessageProcessor {
}
}
protected abstract void downlinkHandle(SessionToHandlerMsg sessionToHandlerMsg) throws Exception;
protected abstract void downlinkHandle(SessionToHandlerMsg sessionToHandlerMsg);
}

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

@@ -20,7 +20,6 @@ import sanbing.jcpp.protocol.domain.ProtocolSession;
import sanbing.jcpp.protocol.provider.ProtocolSessionRegistryProvider;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
/**
* @author baigod
@@ -44,14 +43,14 @@ public class DownlinkController {
final DeferredResult<ResponseEntity<String>> response = new DeferredResult<>(onDownlinkTimeout,
ResponseEntity.status(HttpStatus.REQUEST_TIMEOUT).build());
UUID protocolSessionId = new UUID(downlinkMsg.getSessionIdMSB(),downlinkMsg.getSessionIdLSB()) ;
UUID protocolSessionId = new UUID(downlinkMsg.getSessionIdMSB(), downlinkMsg.getSessionIdLSB());
CompletableFuture<ProtocolSession> protocolSessionCompletableFuture = protocolSessionRegistryProvider.get(protocolSessionId);
ProtocolSession protocolSession = protocolSessionRegistryProvider.get(protocolSessionId);
protocolSessionCompletableFuture.thenAccept(session -> {
if (session != null) {
try {
if (protocolSession != null) {
session.onDownlink(downlinkMsg);
protocolSession.onDownlink(downlinkMsg);
response.setResult(ResponseEntity.status(HttpStatus.OK).build());
} else {
@@ -60,17 +59,15 @@ public class DownlinkController {
response.setResult(ResponseEntity.status(HttpStatus.NOT_FOUND).body("Protocol Session not found for ID:" + protocolSessionId));
}
}).whenComplete((unused, throwable) -> {
if (throwable != null) {
} catch (Exception e) {
log.warn("下发报文时处理失败 sessionId: {}", protocolSessionId, throwable);
log.warn("下发报文时处理失败 sessionId: {}", protocolSessionId, e);
if (!response.isSetOrExpired()) {
if (!response.isSetOrExpired()) {
response.setResult(ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(throwable.getMessage()));
}
response.setResult(ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(e.getMessage()));
}
});
}
return response;
}

3
jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/provider/ProtocolSessionRegistryProvider.java
View File

@@ -7,7 +7,6 @@ package sanbing.jcpp.protocol.provider;
import sanbing.jcpp.protocol.domain.ProtocolSession;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
/**
* @author baigod
@@ -21,7 +20,7 @@ public interface ProtocolSessionRegistryProvider {
void unregister(UUID sessionId);
CompletableFuture<ProtocolSession> get(UUID sessionId);
ProtocolSession get(UUID sessionId);
/**
* 活跃会话

40
jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/provider/impl/DefaultProtocolSessionRegistryProvider.java
View File

@@ -4,7 +4,7 @@
*/
package sanbing.jcpp.protocol.provider.impl;
import com.github.benmanes.caffeine.cache.AsyncCache;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
@@ -20,7 +20,6 @@ import sanbing.jcpp.protocol.provider.ProtocolSessionRegistryProvider;
import java.time.LocalDateTime;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -33,6 +32,7 @@ import java.util.concurrent.TimeUnit;
@Slf4j
public class DefaultProtocolSessionRegistryProvider implements ProtocolSessionRegistryProvider {
private static final int INIT_CACHE_LIMIT = 100_000;
private static final int MAXIMUM_SIZE = 1_000_000;
@Value("${service.protocols.sessions.default-inactivity-timeout-in-sec}")
private int defaultInactivityTimeoutInSec;
@@ -41,23 +41,18 @@ public class DefaultProtocolSessionRegistryProvider implements ProtocolSessionRe
private int defaultStateCheckIntervalInSec;
@Getter
private final AsyncCache<UUID, ProtocolSession> SESSION_CACHE = buildCache();
private final Cache<UUID, ProtocolSession> sessionCache = buildCache();
private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(JCPPThreadFactory.forName("session-state-checker"));
@PostConstruct
public void init() {
scheduledExecutorService.scheduleAtFixedRate(() ->
SESSION_CACHE.asMap().forEach((id, sessionCompletableFuture) ->
sessionCompletableFuture.whenComplete((protocolSession, throwable) -> {
if (throwable == null && protocolSession != null) {
if (protocolSession.getLastActivityTime().isBefore(LocalDateTime.now().minusSeconds(defaultInactivityTimeoutInSec))) {
protocolSession.close(SessionCloseReason.INACTIVE);
unregister(protocolSession.getId());
}
}
})
), defaultStateCheckIntervalInSec, defaultStateCheckIntervalInSec, TimeUnit.SECONDS);
scheduledExecutorService.scheduleAtFixedRate(() -> sessionCache.asMap().forEach((id, session) -> {
if (session.getLastActivityTime().isBefore(LocalDateTime.now().minusSeconds(defaultInactivityTimeoutInSec))) {
session.close(SessionCloseReason.INACTIVE);
unregister(session.getId());
}
}), defaultStateCheckIntervalInSec, defaultStateCheckIntervalInSec, TimeUnit.SECONDS);
}
@PreDestroy
@@ -72,7 +67,7 @@ public class DefaultProtocolSessionRegistryProvider implements ProtocolSessionRe
log.debug("Registering session {}", protocolSession);
}
SESSION_CACHE.put(protocolSession.getId(), CompletableFuture.supplyAsync(() -> protocolSession, ThreadPoolConfiguration.JCPP_COMMON_THREAD_POOL));
sessionCache.put(protocolSession.getId(), protocolSession);
}
@@ -81,16 +76,15 @@ public class DefaultProtocolSessionRegistryProvider implements ProtocolSessionRe
log.info("Unregistering session {}", sessionId);
SESSION_CACHE.synchronous().invalidate(sessionId);
sessionCache.invalidate(sessionId);
}
@Override
public CompletableFuture<ProtocolSession> get(UUID sessionId) {
public ProtocolSession get(UUID sessionId) {
log.debug("Get session {}", sessionId);
return SESSION_CACHE.get(sessionId, uuid -> null);
return sessionCache.get(sessionId, uuid -> null);
}
@Override
@@ -102,14 +96,14 @@ public class DefaultProtocolSessionRegistryProvider implements ProtocolSessionRe
protocolSession.setLastActivityTime(LocalDateTime.now());
SESSION_CACHE.put(protocolSession.getId(), CompletableFuture.supplyAsync(() -> protocolSession, ThreadPoolConfiguration.JCPP_COMMON_THREAD_POOL));
sessionCache.put(protocolSession.getId(), protocolSession);
}
private AsyncCache<UUID, ProtocolSession> buildCache() {
private Cache<UUID, ProtocolSession> buildCache() {
return Caffeine.newBuilder()
.initialCapacity(INIT_CACHE_LIMIT)
.maximumSize(INIT_CACHE_LIMIT * 10)
.maximumSize(MAXIMUM_SIZE)
.executor(ThreadPoolConfiguration.JCPP_COMMON_THREAD_POOL)
.buildAsync();
.build();
}
}

2
jcpp-protocol-bootstrap/src/main/resources/protocol-service.yml
View File

@@ -148,5 +148,5 @@ queue:
thread-pool:
sharding:
hash_function_name: "${THREAD_POOL_SHARDING_HASH_FUNCTION_NAME:murmur3_128}" # murmur3_32, murmur3_128 or sha256
parallelism: "${THREAD_POOL_SHARDING_PARALLELISM:128}"
parallelism: "${THREAD_POOL_SHARDING_PARALLELISM:8}"
stats-print-interval-ms: "${THREAD_POOL_SHARDING_STATS_PRINT_INTERVAL_MS:10000}"

2
jcpp-protocol-bootstrap/src/test/java/sanbing/jcpp/protocol/adapter/DownlinkControllerIT.java
View File

@@ -109,7 +109,7 @@ class DownlinkControllerIT extends AbstractProtocolTestBase {
Thread.sleep(1000);
UUID messageId = UUID.randomUUID();
ProtocolSession protocolSession = sessionRegistryProvider.getSESSION_CACHE().asMap().values().stream().findFirst().get().get();
ProtocolSession protocolSession = sessionRegistryProvider.getSessionCache().asMap().values().stream().findFirst().get();
UUID sessionId = protocolSession.getId();
UUID requestId = UUID.randomUUID();

12
jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/AbstractYunKuaiChongCmdExe.java
View File

@@ -121,10 +121,10 @@ public class AbstractYunKuaiChongCmdExe {
}
protected void encodeAndWriteFlush(YunKuaiChongV150DownlinkCmdEnum downlinkCmd,
int seqNo,
int encryptionFlag,
ByteBuf msgBody,
TcpSession tcpSession) {
int seqNo,
int encryptionFlag,
ByteBuf msgBody,
TcpSession tcpSession) {
byte[] encode = encode(downlinkCmd, seqNo, encryptionFlag, msgBody);
@@ -132,8 +132,8 @@ public class AbstractYunKuaiChongCmdExe {
}
protected void encodeAndWriteFlush(YunKuaiChongV150DownlinkCmdEnum downlinkCmd,
ByteBuf msgBody,
TcpSession tcpSession) {
ByteBuf msgBody,
TcpSession tcpSession) {
byte[] encode = encode(downlinkCmd,
tcpSession.nextSeqNo(SequenceNumberLength.SHORT),

2
jcpp-protocol-yunkuaichong/src/main/java/sanbing/jcpp/protocol/yunkuaichong/v150/YunKuaiChongV15ProtocolMessageProcessor.java
View File

@@ -160,7 +160,7 @@ public class YunKuaiChongV15ProtocolMessageProcessor extends ProtocolMessageProc
}
@Override
public void downlinkHandle(SessionToHandlerMsg sessionToHandlerMsg) throws Exception {
public void downlinkHandle(SessionToHandlerMsg sessionToHandlerMsg) {
TcpSession session = (TcpSession) sessionToHandlerMsg.session();
DownlinkRestMessage protocolDownlinkMsg = sessionToHandlerMsg.downlinkMsg();