调整kafka参数

docker/start.sh

@@ -12,7 +12,7 @@ export JAVA_APP_OPTS="-XX:+UseContainerSupport -XX:InitialRAMPercentage=10 -XX:M
                                  -XX:HeapDumpPath=/var/log/sanbing/heapdump/ \
                                  -XX:+UseTLAB -XX:+ResizeTLAB -XX:+PerfDisableSharedMem -XX:+UseCondCardMark \
                                  -XX:+UseG1GC -XX:MaxGCPauseMillis=500 -XX:+UseStringDeduplication -XX:+ParallelRefProcEnabled -XX:MaxTenuringThreshold=10 \
-                                 -Xss512k -XX:MaxDirectMemorySize=128M -XX:G1ReservePercent=20 \
+                                 -Xss512k -XX:MaxDirectMemorySize=256M -XX:G1ReservePercent=20 \
                                  -XX:-OmitStackTraceInFastThrow \
                                  -Dlogging.config=/app/config/log4j2.xml"
 

jcpp-app/src/main/java/sanbing/jcpp/app/service/grpc/DownlinkGrpcClient.java

@@ -261,7 +261,10 @@ public class DownlinkGrpcClient {
                 }
 
                 if (connectErrTimesMap.computeIfAbsent(hostAndPort, k -> new AtomicInteger()).incrementAndGet() >= maxReconnectTimes) {
-                    queueMap.remove(hostAndPort);
+                    LinkedBlockingQueue<RequestMsg> queue = queueMap.remove(hostAndPort);
+                    if (queue != null) {
+                        queue.clear();
+                    }
                     connectErrTimesMap.remove(hostAndPort);
                     log.info("[{}] Grpc 客户端重连异常超过{}次，不再重连", hostAndPort, maxReconnectTimes);
                 }