guzijian / wvp-GB28181-pro · Commits

Commit 9561e952a382f1214e386c9009e5f7eea9e78802

Authored by 648540858
1 parent 874b37ef

添加使用多线程消息处理sip消息

Inline Side-by-side
Showing 7 changed files with 119 additions and 58 deletions
src/main/java/com/genersoft/iot/vmp/VManageBootstrap.java
  View file @9561e95
... ... @@ -6,6 +6,7 @@ import org.springframework.boot.SpringApplication;
6 6 import org.springframework.boot.autoconfigure.SpringBootApplication;
7 7 import org.springframework.boot.web.servlet.ServletComponentScan;
8 8 import org.springframework.context.ConfigurableApplicationContext;
  9 +import org.springframework.scheduling.annotation.EnableAsync;
9 10 import org.springframework.scheduling.annotation.EnableScheduling;
10 11 import springfox.documentation.oas.annotations.EnableOpenApi;
11 12  
... ...
src/main/java/com/genersoft/iot/vmp/conf/ThreadPoolTaskConfig.java 0 → 100644
  View file @9561e95
  1 +package com.genersoft.iot.vmp.conf;
  2 +
  3 +import org.springframework.context.annotation.Bean;
  4 +import org.springframework.context.annotation.Configuration;
  5 +import org.springframework.scheduling.annotation.EnableAsync;
  6 +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  7 +
  8 +import java.util.concurrent.ThreadPoolExecutor;
  9 +
  10 +@Configuration
  11 +@EnableAsync
  12 +public class ThreadPoolTaskConfig {
  13 +
  14 + /**
  15 + * 默认情况下，在创建了线程池后，线程池中的线程数为0，当有任务来之后，就会创建一个线程去执行任务，
  16 + * 当线程池中的线程数目达到corePoolSize后，就会把到达的任务放到缓存队列当中；
  17 + * 当队列满了，就继续创建线程，当线程数量大于等于maxPoolSize后，开始使用拒绝策略拒绝
  18 + */
  19 +
  20 + /**
  21 + * 核心线程数（默认线程数）
  22 + */
  23 + private static final int corePoolSize = 5;
  24 + /**
  25 + * 最大线程数
  26 + */
  27 + private static final int maxPoolSize = 30;
  28 + /**
  29 + * 允许线程空闲时间（单位：默认为秒）
  30 + */
  31 + private static final int keepAliveTime = 30;
  32 + /**
  33 + * 缓冲队列大小
  34 + */
  35 + private static final int queueCapacity = 10000;
  36 + /**
  37 + * 线程池名前缀
  38 + */
  39 + private static final String threadNamePrefix = "hdl-uhi-service-";
  40 +
  41 + @Bean("taskExecutor") // bean的名称，默认为首字母小写的方法名
  42 + public ThreadPoolTaskExecutor taskExecutor() {
  43 + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
  44 + executor.setCorePoolSize(corePoolSize);
  45 + executor.setMaxPoolSize(maxPoolSize);
  46 + executor.setQueueCapacity(queueCapacity);
  47 + executor.setKeepAliveSeconds(keepAliveTime);
  48 + executor.setThreadNamePrefix(threadNamePrefix);
  49 +
  50 + // 线程池对拒绝任务的处理策略
  51 + // CallerRunsPolicy：由调用线程（提交任务的线程）处理该任务
  52 + executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
  53 + // 初始化
  54 + executor.initialize();
  55 + return executor;
  56 + }
  57 +}
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java
  View file @9561e95
... ... @@ -2,6 +2,7 @@ package com.genersoft.iot.vmp.gb28181;
2 2  
3 3 import com.genersoft.iot.vmp.conf.SipConfig;
4 4 import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
  5 +import com.genersoft.iot.vmp.gb28181.transmit.ISIPProcessorObserver;
5 6 import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
6 7 import gov.nist.javax.sip.SipProviderImpl;
7 8 import gov.nist.javax.sip.SipStackImpl;
... ... @@ -28,28 +29,12 @@ public class SipLayer{
28 29 private SipConfig sipConfig;
29 30  
30 31 @Autowired
31   - private SIPProcessorObserver sipProcessorObserver;
32   -
33   - @Autowired
34   - private SipSubscribe sipSubscribe;
  32 + private ISIPProcessorObserver sipProcessorObserver;
35 33  
36 34 private SipStackImpl sipStack;
37 35  
38 36 private SipFactory sipFactory;
39 37  
40   - /**
41   - * 消息处理器线程池
42   - */
43   - private ThreadPoolExecutor processThreadPool;
44   -
45   - public SipLayer() {
46   - int processThreadNum = Runtime.getRuntime().availableProcessors() * 10;
47   - LinkedBlockingQueue<Runnable> processQueue = new LinkedBlockingQueue<>(10000);
48   - processThreadPool = new ThreadPoolExecutor(processThreadNum,processThreadNum,
49   - 0L,TimeUnit.MILLISECONDS,processQueue,
50   - new ThreadPoolExecutor.CallerRunsPolicy());
51   - }
52   -
53 38  
54 39 @Bean("sipFactory")
55 40 private SipFactory createSipFactory() {
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/ISIPProcessorObserver.java 0 → 100644
  View file @9561e95
  1 +package com.genersoft.iot.vmp.gb28181.transmit;
  2 +
  3 +import javax.sip.SipListener;
  4 +
  5 +public interface ISIPProcessorObserver extends SipListener {
  6 +}
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java
  View file @9561e95
... ... @@ -7,6 +7,9 @@ import com.genersoft.iot.vmp.gb28181.transmit.event.timeout.ITimeoutProcessor;
7 7 import org.slf4j.Logger;
8 8 import org.slf4j.LoggerFactory;
9 9 import org.springframework.beans.factory.annotation.Autowired;
  10 +import org.springframework.beans.factory.annotation.Qualifier;
  11 +import org.springframework.scheduling.annotation.Async;
  12 +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
10 13 import org.springframework.stereotype.Component;
11 14  
12 15 import javax.sip.*;
... ... @@ -22,7 +25,7 @@ import java.util.concurrent.ConcurrentHashMap;
22 25 * @date: 2021年11月5日 下午15：32
23 26 */
24 27 @Component
25   -public class SIPProcessorObserver implements SipListener {
  28 +public class SIPProcessorObserver implements ISIPProcessorObserver {
26 29  
27 30 private final static Logger logger = LoggerFactory.getLogger(SIPProcessorObserver.class);
28 31  
... ... @@ -33,6 +36,10 @@ public class SIPProcessorObserver implements SipListener {
33 36 @Autowired
34 37 private SipSubscribe sipSubscribe;
35 38  
  39 + @Autowired
  40 + @Qualifier(value = "taskExecutor")
  41 + private ThreadPoolTaskExecutor poolTaskExecutor;
  42 +
36 43 /**
37 44 * 添加 request订阅
38 45 * @param method 方法名
... ... @@ -65,13 +72,17 @@ public class SIPProcessorObserver implements SipListener {
65 72 */
66 73 @Override
67 74 public void processRequest(RequestEvent requestEvent) {
68   - String method = requestEvent.getRequest().getMethod();
69   - ISIPRequestProcessor sipRequestProcessor = requestProcessorMap.get(method);
70   - if (sipRequestProcessor == null) {
71   - logger.warn("不支持方法{}的request", method);
72   - return;
73   - }
74   - requestProcessorMap.get(method).process(requestEvent);
  75 +
  76 + poolTaskExecutor.execute(() -> {
  77 + String method = requestEvent.getRequest().getMethod();
  78 + ISIPRequestProcessor sipRequestProcessor = requestProcessorMap.get(method);
  79 + if (sipRequestProcessor == null) {
  80 + logger.warn("不支持方法{}的request", method);
  81 + return;
  82 + }
  83 + requestProcessorMap.get(method).process(requestEvent);
  84 + });
  85 +
75 86 }
76 87  
77 88 /**
... ... @@ -90,43 +101,45 @@ public class SIPProcessorObserver implements SipListener {
90 101 // }
91 102 // sipRequestProcessor.process(responseEvent);
92 103  
93   -
94   - Response response = responseEvent.getResponse();
95   - logger.debug(responseEvent.getResponse().toString());
96   - int status = response.getStatusCode();
97   - if (((status >= 200) && (status < 300)) || status == 401) { // Success!
  104 + poolTaskExecutor.execute(() -> {
  105 + Response response = responseEvent.getResponse();
  106 + logger.debug(responseEvent.getResponse().toString());
  107 + int status = response.getStatusCode();
  108 + if (((status >= 200) && (status < 300)) || status == 401) { // Success!
98 109 // ISIPResponseProcessor processor = processorFactory.createResponseProcessor(evt);
99   - CSeqHeader cseqHeader = (CSeqHeader) responseEvent.getResponse().getHeader(CSeqHeader.NAME);
100   - String method = cseqHeader.getMethod();
101   - ISIPResponseProcessor sipRequestProcessor = responseProcessorMap.get(method);
102   - if (sipRequestProcessor != null) {
103   - sipRequestProcessor.process(responseEvent);
104   - }
105   - if (responseEvent.getResponse() != null && sipSubscribe.getOkSubscribesSize() > 0 ) {
106   - CallIdHeader callIdHeader = (CallIdHeader)responseEvent.getResponse().getHeader(CallIdHeader.NAME);
107   - if (callIdHeader != null) {
108   - SipSubscribe.Event subscribe = sipSubscribe.getOkSubscribe(callIdHeader.getCallId());
109   - if (subscribe != null) {
110   - SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(responseEvent);
111   - subscribe.response(eventResult);
  110 + CSeqHeader cseqHeader = (CSeqHeader) responseEvent.getResponse().getHeader(CSeqHeader.NAME);
  111 + String method = cseqHeader.getMethod();
  112 + ISIPResponseProcessor sipRequestProcessor = responseProcessorMap.get(method);
  113 + if (sipRequestProcessor != null) {
  114 + sipRequestProcessor.process(responseEvent);
  115 + }
  116 + if (responseEvent.getResponse() != null && sipSubscribe.getOkSubscribesSize() > 0 ) {
  117 + CallIdHeader callIdHeader = (CallIdHeader)responseEvent.getResponse().getHeader(CallIdHeader.NAME);
  118 + if (callIdHeader != null) {
  119 + SipSubscribe.Event subscribe = sipSubscribe.getOkSubscribe(callIdHeader.getCallId());
  120 + if (subscribe != null) {
  121 + SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(responseEvent);
  122 + subscribe.response(eventResult);
  123 + }
112 124 }
113 125 }
114   - }
115   - } else if ((status >= 100) && (status < 200)) {
116   - // 增加其它无需回复的响应，如101、180等
117   - } else {
118   - logger.warn("接收到失败的response响应！status：" + status + ",message:" + response.getReasonPhrase()/* .getContent().toString()*/);
119   - if (responseEvent.getResponse() != null && sipSubscribe.getErrorSubscribesSize() > 0 ) {
120   - CallIdHeader callIdHeader = (CallIdHeader)responseEvent.getResponse().getHeader(CallIdHeader.NAME);
121   - if (callIdHeader != null) {
122   - SipSubscribe.Event subscribe = sipSubscribe.getErrorSubscribe(callIdHeader.getCallId());
123   - if (subscribe != null) {
124   - SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(responseEvent);
125   - subscribe.response(eventResult);
  126 + } else if ((status >= 100) && (status < 200)) {
  127 + // 增加其它无需回复的响应，如101、180等
  128 + } else {
  129 + logger.warn("接收到失败的response响应！status：" + status + ",message:" + response.getReasonPhrase()/* .getContent().toString()*/);
  130 + if (responseEvent.getResponse() != null && sipSubscribe.getErrorSubscribesSize() > 0 ) {
  131 + CallIdHeader callIdHeader = (CallIdHeader)responseEvent.getResponse().getHeader(CallIdHeader.NAME);
  132 + if (callIdHeader != null) {
  133 + SipSubscribe.Event subscribe = sipSubscribe.getErrorSubscribe(callIdHeader.getCallId());
  134 + if (subscribe != null) {
  135 + SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(responseEvent);
  136 + subscribe.response(eventResult);
  137 + }
126 138 }
127 139 }
128 140 }
129   - }
  141 + });
  142 +
130 143  
131 144 }
132 145  
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderProvider.java
  View file @9561e95
... ... @@ -204,7 +204,6 @@ public class SIPRequestHeaderProvider {
204 204  
205 205 // Event
206 206 EventHeader eventHeader = sipFactory.createHeaderFactory().createEventHeader(event);
207   - eventHeader.setEventType("Catalog");
208 207 request.addHeader(eventHeader);
209 208  
210 209 ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("APPLICATION", "MANSCDP+xml");
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
  View file @9561e95
... ... @@ -1496,7 +1496,7 @@ public class SIPCommander implements ISIPCommander {
1496 1496 CallIdHeader callIdHeader = device.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
1497 1497 : udpSipProvider.getNewCallId();
1498 1498  
1499   - Request request = headerProvider.createSubscribeRequest(device, cmdXml.toString(), "z9hG4bK-viaPos-" + tm, "fromTagPos" + tm, null, device.getSubscribeCycleForCatalog(), "presence" , callIdHeader);
  1499 + Request request = headerProvider.createSubscribeRequest(device, cmdXml.toString(), "z9hG4bK-viaPos-" + tm, "fromTagPos" + tm, null, device.getSubscribeCycleForCatalog(), "Catalog" , callIdHeader);
1500 1500 transmitRequest(device, request, errorEvent, okEvent);
1501 1501  
1502 1502 return true;
... ...