Shikong/wvp-gb28181-project
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
648540858 2021-11-17 17:03:48 +08:00
parent d591ef3248
commit 0ffacc3053
4 changed files with 51 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java
View File

@ -36,9 +36,9 @@ public class SIPProcessorObserver implements ISIPProcessorObserver {
@Autowired @Autowired
private SipSubscribe sipSubscribe; private SipSubscribe sipSubscribe;
@Autowired // @Autowired
@Qualifier(value = "taskExecutor") // @Qualifier(value = "taskExecutor")
private ThreadPoolTaskExecutor poolTaskExecutor; // private ThreadPoolTaskExecutor poolTaskExecutor;
/** /**
* 添加 request订阅 * 添加 request订阅
@ -71,9 +71,8 @@ public class SIPProcessorObserver implements ISIPProcessorObserver {
* @param requestEvent RequestEvent事件 * @param requestEvent RequestEvent事件
*/ */
@Override @Override
@Async
public void processRequest(RequestEvent requestEvent) { public void processRequest(RequestEvent requestEvent) {
poolTaskExecutor.execute(() -> {
String method = requestEvent.getRequest().getMethod(); String method = requestEvent.getRequest().getMethod();
ISIPRequestProcessor sipRequestProcessor = requestProcessorMap.get(method); ISIPRequestProcessor sipRequestProcessor = requestProcessorMap.get(method);
if (sipRequestProcessor == null) { if (sipRequestProcessor == null) {
@ -81,7 +80,6 @@ public class SIPProcessorObserver implements ISIPProcessorObserver {
return; return;
} }
requestProcessorMap.get(method).process(requestEvent); requestProcessorMap.get(method).process(requestEvent);
});
} }
@ -90,18 +88,9 @@ public class SIPProcessorObserver implements ISIPProcessorObserver {
* @param responseEvent responseEvent事件 * @param responseEvent responseEvent事件
*/ */
@Override @Override
@Async
public void processResponse(ResponseEvent responseEvent) { public void processResponse(ResponseEvent responseEvent) {
logger.debug(responseEvent.getResponse().toString()); 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(); Response response = responseEvent.getResponse();
logger.debug(responseEvent.getResponse().toString()); logger.debug(responseEvent.getResponse().toString());
int status = response.getStatusCode(); int status = response.getStatusCode();
@ -138,7 +127,6 @@ public class SIPProcessorObserver implements ISIPProcessorObserver {
} }
} }
} }
});
} }

5
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java
View File

@ -36,7 +36,8 @@ import java.util.Map;
@Component @Component
public class ByeRequestProcessor extends SIPRequestProcessorParent implements InitializingBean, ISIPRequestProcessor { 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 @Autowired
private ISIPCommander cmder; private ISIPCommander cmder;
@ -53,8 +54,6 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
@Autowired @Autowired
private IMediaServerService mediaServerService; private IMediaServerService mediaServerService;
private String method = "BYE";
@Autowired @Autowired
private SIPProcessorObserver sipProcessorObserver; private SIPProcessorObserver sipProcessorObserver;

3
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/MessageHandlerAbstract.java
View File

@ -16,9 +16,6 @@ public abstract class MessageHandlerAbstract extends SIPRequestProcessorParent i
public static Map<String, IMessageHandler> messageHandlerMap = new ConcurrentHashMap<>(); public static Map<String, IMessageHandler> messageHandlerMap = new ConcurrentHashMap<>();
@Autowired
public MessageRequestProcessor messageRequestProcessor;
public void addHandler(String cmdType, IMessageHandler messageHandler) { public void addHandler(String cmdType, IMessageHandler messageHandler) {
messageHandlerMap.put(cmdType, messageHandler); messageHandlerMap.put(cmdType, messageHandler);
} }

5
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/NotifyMessageHandler.java
View File

@ -1,7 +1,9 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.notify; 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.MessageHandlerAbstract;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.MessageRequestProcessor;
import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
@Component @Component
@ -9,6 +11,9 @@ public class NotifyMessageHandler extends MessageHandlerAbstract implements Init
private final String messageType = "Notify"; private final String messageType = "Notify";
@Autowired
private MessageRequestProcessor messageRequestProcessor;
@Override @Override
public void afterPropertiesSet() throws Exception { public void afterPropertiesSet() throws Exception {
messageRequestProcessor.addHandler(messageType, this); messageRequestProcessor.addHandler(messageType, this);