From 9561e952a382f1214e386c9009e5f7eea9e78802 Mon Sep 17 00:00:00 2001 From: 648540858 <456PANlinlin> Date: 星期三, 17 十一月 2021 00:07:34 +0800 Subject: [PATCH] 添加使用多线程消息处理sip消息 --- src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java | 95 +++++++++++++++++++++++++++-------------------- 1 files changed, 54 insertions(+), 41 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java index cac1a01..9149be1 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java @@ -7,6 +7,9 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.scheduling.annotation.Async; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Component; import javax.sip.*; @@ -22,7 +25,7 @@ * @date: 2021骞�11鏈�5鏃� 涓嬪崍15锛�32 */ @Component -public class SIPProcessorObserver implements SipListener { +public class SIPProcessorObserver implements ISIPProcessorObserver { private final static Logger logger = LoggerFactory.getLogger(SIPProcessorObserver.class); @@ -32,6 +35,10 @@ @Autowired private SipSubscribe sipSubscribe; + + @Autowired + @Qualifier(value = "taskExecutor") + private ThreadPoolTaskExecutor poolTaskExecutor; /** * 娣诲姞 request璁㈤槄 @@ -65,13 +72,17 @@ */ @Override public void processRequest(RequestEvent requestEvent) { - String method = requestEvent.getRequest().getMethod(); - ISIPRequestProcessor sipRequestProcessor = requestProcessorMap.get(method); - if (sipRequestProcessor == null) { - logger.warn("涓嶆敮鎸佹柟娉晎}鐨剅equest", method); - return; - } - requestProcessorMap.get(method).process(requestEvent); + + poolTaskExecutor.execute(() -> { + String method = requestEvent.getRequest().getMethod(); + ISIPRequestProcessor sipRequestProcessor = requestProcessorMap.get(method); + if (sipRequestProcessor == null) { + logger.warn("涓嶆敮鎸佹柟娉晎}鐨剅equest", method); + return; + } + requestProcessorMap.get(method).process(requestEvent); + }); + } /** @@ -90,43 +101,45 @@ // } // sipRequestProcessor.process(responseEvent); - - Response response = responseEvent.getResponse(); - logger.debug(responseEvent.getResponse().toString()); - int status = response.getStatusCode(); - if (((status >= 200) && (status < 300)) || status == 401) { // Success! + poolTaskExecutor.execute(() -> { + 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鍝嶅簲锛乻tatus锛�" + 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鍝嶅簲锛乻tatus锛�" + 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); - } - } - } - } + }); + } -- Gitblit v1.8.0