基于springboot 長輪詢的實現(xiàn)操作

更新時間：2021年01月20日 09:16:27 作者：食得落

這篇文章主要介紹了基于springboot 長輪詢的實現(xiàn)操作，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧

springboot 長輪詢實現(xiàn)

基于 @EnableAsync , @Sync

@SpringBootApplication
@EnableAsync
public class DemoApplication {
 public static void main(String[] args) {
 SpringApplication.run(DemoApplication.class, args);
 }
}

@RequestMapping("/async")
@RestController
public class AsyncRequestDemo {
 @Autowired
 private AsyncRequestService asyncRequestService;
 @GetMapping("/value")
 public String getValue() {
 String msg = null;
 Future<String> result = null;
 try{
  result = asyncRequestService.getValue();
  msg = result.get(10, TimeUnit.SECONDS);
 }catch (Exception e){
  e.printStackTrace();
 }finally {
  if (result != null){
  result.cancel(true);
  }
 }
 return msg;
 }
 @PostMapping("/value")
 public void postValue(String msg) {
 asyncRequestService.postValue(msg);
 }
}

@Service
public class AsyncRequestService {
 private String msg = null;
 @Async
 public Future<String> getValue() throws InterruptedException {
 while (true){
  synchronized (this){
  if (msg != null){
   String resultMsg = msg;
   msg = null;
   return new AsyncResult(resultMsg);
  }
  }
  Thread.sleep(100);
 }
 }
 public synchronized void postValue(String msg) {
 this.msg = msg;
 }
}

備注

@EnableAsync 開啟異步

@Sync 標記異步方法

Future 用于接收異步返回值

result.get(10, TimeUnit.SECONDS); 阻塞，超時獲取結果

Future.cancel() 中斷線程

補充：通過spring提供的DeferredResult實現(xiàn)長輪詢服務端推送消息

DeferredResult字面意思就是推遲結果，是在servlet3.0以后引入了異步請求之后，spring封裝了一下提供了相應的支持，也是一個很老的特性了。DeferredResult可以允許容器線程快速釋放以便可以接受更多的請求提升吞吐量，讓真正的業(yè)務邏輯在其他的工作線程中去完成。

最近再看apollo配置中心的實現(xiàn)原理，apollo的發(fā)布配置推送變更消息就是用DeferredResult實現(xiàn)的，apollo客戶端會像服務端發(fā)送長輪訓http請求，超時時間60秒，當超時后返回客戶端一個304 httpstatus,表明配置沒有變更，客戶端繼續(xù)這個步驟重復發(fā)起請求，當有發(fā)布配置的時候，服務端會調用DeferredResult.setResult返回200狀態(tài)碼，然后輪訓請求會立即返回（不會超時），客戶端收到響應結果后，會發(fā)起請求獲取變更后的配置信息。

下面我們自己寫一個簡單的demo來演示這個過程

springboot啟動類：

@SpringBootApplication
public class DemoApplication implements WebMvcConfigurer { 
 public static void main(String[] args) {
 SpringApplication.run(DemoApplication.class, args);
 } 
 
 @Bean
 public ThreadPoolTaskExecutor mvcTaskExecutor() {
 ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
 executor.setCorePoolSize(10);
 executor.setQueueCapacity(100);
 executor.setMaxPoolSize(25);
 return executor;
 
 }
 
 //配置異步支持，設置了一個用來異步執(zhí)行業(yè)務邏輯的工作線程池，設置了默認的超時時間是60秒
 @Override
 public void configureAsyncSupport(AsyncSupportConfigurer configurer) {
 configurer.setTaskExecutor(mvcTaskExecutor());
 configurer.setDefaultTimeout(60000L);
 }
}

import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
import com.google.common.collect.Multimaps;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.context.request.async.DeferredResult; 
import java.util.Collection;
 
@RestController
public class ApolloController {
 private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
 //guava中的Multimap，多值map,對map的增強，一個key可以保持多個value
 private Multimap<String, DeferredResult<String>> watchRequests = Multimaps.synchronizedSetMultimap(HashMultimap.create());
 
 
 //模擬長輪詢
 @RequestMapping(value = "/watch/{namespace}", method = RequestMethod.GET, produces = "text/html")
 public DeferredResult<String> watch(@PathVariable("namespace") String namespace) {
 logger.info("Request received");
 DeferredResult<String> deferredResult = new DeferredResult<>();
 //當deferredResult完成時（不論是超時還是異常還是正常完成），移除watchRequests中相應的watch key
 deferredResult.onCompletion(new Runnable() {
  @Override
  public void run() {
  System.out.println("remove key:" + namespace);
  watchRequests.remove(namespace, deferredResult);
  }
 });
 watchRequests.put(namespace, deferredResult);
 logger.info("Servlet thread released");
 return deferredResult;
 
 
 }
 
 //模擬發(fā)布namespace配置
 @RequestMapping(value = "/publish/{namespace}", method = RequestMethod.GET, produces = "text/html")
 public Object publishConfig(@PathVariable("namespace") String namespace) {
 if (watchRequests.containsKey(namespace)) {
  Collection<DeferredResult<String>> deferredResults = watchRequests.get(namespace);
  Long time = System.currentTimeMillis();
  //通知所有watch這個namespace變更的長輪訓配置變更結果
  for (DeferredResult<String> deferredResult : deferredResults) {
  deferredResult.setResult(namespace + " changed:" + time);
  }
 }
 return "success";
 
 }
}

當請求超時的時候會產生AsyncRequestTimeoutException,我們定義一個全局異常捕獲類:

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.ControllerAdvice;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.context.request.async.AsyncRequestTimeoutException;
 
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
 
@ControllerAdvice
class GlobalControllerExceptionHandler {
 
 protected static final Logger logger = LoggerFactory.getLogger(GlobalControllerExceptionHandler.class);
 
 @ResponseStatus(HttpStatus.NOT_MODIFIED)//返回304狀態(tài)碼
 @ResponseBody
 @ExceptionHandler(AsyncRequestTimeoutException.class) //捕獲特定異常
 public void handleAsyncRequestTimeoutException(AsyncRequestTimeoutException e, HttpServletRequest request) {
 System.out.println("handleAsyncRequestTimeoutException");
 }
}

然后我們通過postman工具發(fā)送請求http://localhost:8080/watch/mynamespace，請求會掛起，60秒后，DeferredResult超時，客戶端正常收到了304狀態(tài)碼，表明在這個期間配置沒有變更過。

然后我們在模擬配置變更的情況，再次發(fā)起請求http://localhost:8080/watch/mynamespace，等待個10秒鐘（不要超過60秒），然后調用http://localhost:8080/publish/mynamespace,發(fā)布配置變更。這時postman會立刻收到response響應結果: