Merge pull request #5012 from 317787106/hotfix/zmq_lock

feat(db): add timeout sending of zmq

Author: 317787106

framework/src/main/java/org/tron/common/logsfilter/nativequeue/NativeMessageQueue.java

@@ -11,6 +11,7 @@ public class NativeMessageQueue {
 
   private static final int DEFAULT_BIND_PORT = 5555;
   private static final int DEFAULT_QUEUE_LENGTH = 1000;
+  private static final int ZMQ_SEND_TIME_OUT = 1_000;
   private static NativeMessageQueue instance;
   private ZContext context = null;
   private ZMQ.Socket publisher = null;
@@ -33,7 +34,7 @@ public boolean start(int bindPort, int sendQueueLength) {
     if (Objects.isNull(publisher)) {
       return false;
     }
-
+    publisher.setSendTimeOut(ZMQ_SEND_TIME_OUT);
     if (bindPort == 0 || bindPort < 0) {
       bindPort = DEFAULT_BIND_PORT;
     }

framework/src/main/java/org/tron/core/net/P2pEventHandlerImpl.java

@@ -222,10 +222,10 @@ private void processException(PeerConnection peer, TronMessage msg, Exception ex
           break;
       }
       if (type.equals(P2pException.TypeEnum.BAD_MESSAGE)) {
-        logger.error("Message from {} process failed, {} \n type: {}",
+        logger.error("Message from {} process failed, {} \n type: ({})",
             peer.getInetSocketAddress(), msg, type, ex);
       } else {
-        logger.warn("Message from {} process failed, {} \n type: {}, detail: {}",
+        logger.warn("Message from {} process failed, {} \n type: ({}), detail: {}",
             peer.getInetSocketAddress(), msg, type, ex.getMessage());
       }
     } else {