使用@Async多线程处理sip消息

648540858
1 parent f559e6a3
Showing 4 changed files with 49 additions and 60 deletions
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/MessageHandlerAbstract.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/NotifyMessageHandler.java
@@ -36,9 +36,9 @@ public class SIPProcessorObserver implements ISIPProcessorObserver {
     @Autowired
     private SipSubscribe sipSubscribe;
-    @Autowired
-    @Qualifier(value = "taskExecutor")
-    private ThreadPoolTaskExecutor poolTaskExecutor;
+//    @Autowired
+//    @Qualifier(value = "taskExecutor")
+//    private ThreadPoolTaskExecutor poolTaskExecutor;
     /**
      * 添加 request订阅
@@ -71,17 +71,15 @@ public class SIPProcessorObserver implements ISIPProcessorObserver {
      * @param requestEvent RequestEvent事件
      */
     @Override
+    @Async
     public void processRequest(RequestEvent requestEvent) {
-
-        poolTaskExecutor.execute(() -> {
-            String method = requestEvent.getRequest().getMethod();
-            ISIPRequestProcessor sipRequestProcessor = requestProcessorMap.get(method);
-            if (sipRequestProcessor == null) {
-                logger.warn("不支持方法{}的request", method);
-                return;
-            }
-            requestProcessorMap.get(method).process(requestEvent);
-        });
+        String method = requestEvent.getRequest().getMethod();
+        ISIPRequestProcessor sipRequestProcessor = requestProcessorMap.get(method);
+        if (sipRequestProcessor == null) {
+            logger.warn("不支持方法{}的request", method);
+            return;
+        }
+        requestProcessorMap.get(method).process(requestEvent);
     }
@@ -90,55 +88,45 @@ public class SIPProcessorObserver implements ISIPProcessorObserver {
      * @param responseEvent responseEvent事件
      */
     @Override
+    @Async
     public void processResponse(ResponseEvent responseEvent) {
         logger.debug(responseEvent.getResponse().toString());
-//        CSeqHeader cseqHeader = (CSeqHeader) responseEvent.getResponse().getHeader(CSeqHeader.NAME);
-//        String method = cseqHeader.getMethod();
-//        ISIPResponseProcessor sipRequestProcessor = responseProcessorMap.get(method);
-//        if (sipRequestProcessor == null) {
-//            logger.warn("不支持方法{}的response", method);
-//            return;
-//        }
-//        sipRequestProcessor.process(responseEvent);
-
-        poolTaskExecutor.execute(() -> {
-            Response response = responseEvent.getResponse();
-            logger.debug(responseEvent.getResponse().toString());
-            int status = response.getStatusCode();
-            if (((status >= 200) && (status < 300)) || status == 401) { // Success!
+        Response response = responseEvent.getResponse();
+        logger.debug(responseEvent.getResponse().toString());
+        int status = response.getStatusCode();
+        if (((status >= 200) && (status < 300)) || status == 401) { // Success!
 //            ISIPResponseProcessor processor = processorFactory.createResponseProcessor(evt);
-                CSeqHeader cseqHeader = (CSeqHeader) responseEvent.getResponse().getHeader(CSeqHeader.NAME);
-                String method = cseqHeader.getMethod();
-                ISIPResponseProcessor sipRequestProcessor = responseProcessorMap.get(method);
-                if (sipRequestProcessor != null) {
-                    sipRequestProcessor.process(responseEvent);
-                }
-                if (responseEvent.getResponse() != null && sipSubscribe.getOkSubscribesSize() > 0 ) {
-                    CallIdHeader callIdHeader = (CallIdHeader)responseEvent.getResponse().getHeader(CallIdHeader.NAME);
-                    if (callIdHeader != null) {
-                        SipSubscribe.Event subscribe = sipSubscribe.getOkSubscribe(callIdHeader.getCallId());
-                        if (subscribe != null) {
-                            SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(responseEvent);
-                            subscribe.response(eventResult);
-                        }
+            CSeqHeader cseqHeader = (CSeqHeader) responseEvent.getResponse().getHeader(CSeqHeader.NAME);
+            String method = cseqHeader.getMethod();
+            ISIPResponseProcessor sipRequestProcessor = responseProcessorMap.get(method);
+            if (sipRequestProcessor != null) {
+                sipRequestProcessor.process(responseEvent);
+            }
+            if (responseEvent.getResponse() != null && sipSubscribe.getOkSubscribesSize() > 0 ) {
+                CallIdHeader callIdHeader = (CallIdHeader)responseEvent.getResponse().getHeader(CallIdHeader.NAME);
+                if (callIdHeader != null) {
+                    SipSubscribe.Event subscribe = sipSubscribe.getOkSubscribe(callIdHeader.getCallId());
+                    if (subscribe != null) {
+                        SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(responseEvent);
+                        subscribe.response(eventResult);
                     }
                 }
-            } else if ((status >= 100) && (status < 200)) {
-                // 增加其它无需回复的响应，如101、180等
-            } else {
-                logger.warn("接收到失败的response响应！status：" + status + ",message:" + response.getReasonPhrase()/* .getContent().toString()*/);
-                if (responseEvent.getResponse() != null && sipSubscribe.getErrorSubscribesSize() > 0 ) {
-                    CallIdHeader callIdHeader = (CallIdHeader)responseEvent.getResponse().getHeader(CallIdHeader.NAME);
-                    if (callIdHeader != null) {
-                        SipSubscribe.Event subscribe = sipSubscribe.getErrorSubscribe(callIdHeader.getCallId());
-                        if (subscribe != null) {
-                            SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(responseEvent);
-                            subscribe.response(eventResult);
-                        }
+            }
+        } else if ((status >= 100) && (status < 200)) {
+            // 增加其它无需回复的响应，如101、180等
+        } else {
+            logger.warn("接收到失败的response响应！status：" + status + ",message:" + response.getReasonPhrase()/* .getContent().toString()*/);
+            if (responseEvent.getResponse() != null && sipSubscribe.getErrorSubscribesSize() > 0 ) {
+                CallIdHeader callIdHeader = (CallIdHeader)responseEvent.getResponse().getHeader(CallIdHeader.NAME);
+                if (callIdHeader != null) {
+                    SipSubscribe.Event subscribe = sipSubscribe.getErrorSubscribe(callIdHeader.getCallId());
+                    if (subscribe != null) {
+                        SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(responseEvent);
+                        subscribe.response(eventResult);
                     }
                 }
             }
-        });
+        }
     }
@@ -36,7 +36,8 @@ import java.util.Map;
 @Component
 public class ByeRequestProcessor extends SIPRequestProcessorParent implements InitializingBean, ISIPRequestProcessor {
-	private Logger logger = LoggerFactory.getLogger(ByeRequestProcessor.class);
+	private final Logger logger = LoggerFactory.getLogger(ByeRequestProcessor.class);
+	private final String method = "BYE";
 	@Autowired
 	private ISIPCommander cmder;
@@ -53,8 +54,6 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
 	@Autowired
 	private IMediaServerService mediaServerService;
-	private String method = "BYE";
-
 	@Autowired
 	private SIPProcessorObserver sipProcessorObserver;
@@ -16,9 +16,6 @@ public abstract class MessageHandlerAbstract extends SIPRequestProcessorParent i
     public static Map<String, IMessageHandler> messageHandlerMap = new ConcurrentHashMap<>();
-    @Autowired
-    public MessageRequestProcessor messageRequestProcessor;
-
     public void addHandler(String cmdType, IMessageHandler messageHandler) {
         messageHandlerMap.put(cmdType, messageHandler);
     }
 package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.notify;
 import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.MessageHandlerAbstract;
+import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.MessageRequestProcessor;
 import org.springframework.beans.factory.InitializingBean;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 @Component
@@ -9,6 +11,9 @@ public class NotifyMessageHandler extends MessageHandlerAbstract implements Init
     private final String messageType = "Notify";
+    @Autowired
+    private MessageRequestProcessor messageRequestProcessor;
+
     @Override
     public void afterPropertiesSet() throws Exception {
         messageRequestProcessor.addHandler(messageType, this);