Springboot中怎么实现高吞吐量异步处理
创始人
2024-12-23 19:42:52
0次

1. 新建Maven项目 async

2. pom.xml

 4.0.0  com.java  async  1.0.0      org.springframework.boot    spring-boot-starter-parent    2.0.5.RELEASE                  org.springframework.boot      spring-boot-starter-web                  org.springframework      springloaded      1.2.8.RELEASE      provided              org.springframework.boot      spring-boot-devtools      provided            ${project.artifactId}                  org.apache.maven.plugins        maven-compiler-plugin                  1.8          1.8          UTF-8                            org.springframework.boot        spring-boot-maven-plugin                                            repackage                                          

3. AsyncStarter.java

package com.java;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplicationpublic class AsyncStarter {  public static void main(String[] args) {    SpringApplication.run(AsyncStarter.class, args);  }}

4. AsyncVo.java

package com.java.vo;import org.springframework.web.context.request.async.DeferredResult;/** * 存储异步处理信息 *  * @author Logen * * @param 接口输入参数 * @param 接口返回参数 */public class AsyncVo {  /**   * 请求参数   */  private I params;  /**   * 响应结果   */  private DeferredResult result;  public I getParams() {    return params;  }  public void setParams(I params) {    this.params = params;  }  public DeferredResult getResult() {    return result;  }  public void setResult(DeferredResult result) {    this.result = result;  }}

5. RequestQueue.java

package com.java.queue;import java.util.concurrent.BlockingQueue;import java.util.concurrent.LinkedBlockingQueue;import org.springframework.stereotype.Component;import com.java.vo.AsyncVo;/** * 存放所有异步处理接口请求队列的对象,一个接口对应一个队列 *  * @author Logen * */@Componentpublic class RequestQueue {  /**   * 处理下订单接口的队列,设置缓冲容量为50   */  private BlockingQueue> orderQueue = new LinkedBlockingQueue<>(50);  public BlockingQueue> getOrderQueue() {    return orderQueue;  }}

6. OrderTask.java

package com.java.task;import java.util.HashMap;import java.util.Map;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;import com.java.queue.RequestQueue;import com.java.vo.AsyncVo;/** * 处理订单接口的任务,每个任务类处理一种接口 *  * @author Logen * */@Componentpublic class OrderTask extends Thread {  @Autowired  private RequestQueue queue;  private boolean running = true;  @Override  public void run() {    while (running) {      try {        AsyncVo vo = queue.getOrderQueue().take();        System.out.println("[ OrderTask ]开始处理订单");        String params = vo.getParams();        Thread.sleep(3000);        Map map = new HashMap<>();        map.put("params", params);        map.put("time", System.currentTimeMillis());        vo.getResult().setResult(map);        System.out.println("[ OrderTask ]订单处理完成");      } catch (InterruptedException e) {        e.printStackTrace();        running = false;      }    }  }  public void setRunning(boolean running) {    this.running = running;  }}

7. QueueListener.java

package com.java.listener;import javax.annotation.PostConstruct;import javax.annotation.PreDestroy;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;import com.java.task.OrderTask;/** * 队列监听器,初始化启动所有监听任务 *  * @author Logen * */@Componentpublic class QueueListener {  @Autowired  private OrderTask orderTask;  /**   * 初始化时启动监听请求队列   */  @PostConstruct  public void init() {    orderTask.start();  }  /**   * 销毁容器时停止监听任务   */  @PreDestroy  public void destory() {    orderTask.setRunning(false);  }}

8. OrderController.java

package com.java.controller;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.RestController;import org.springframework.web.context.request.async.DeferredResult;import com.java.queue.RequestQueue;import com.java.vo.AsyncVo;/** *

*  *
 *  * 模拟下单处理,实现高吞吐量异步处理请求 *  * 1、 Controller层接口只接收请求,不进行处理,而是把请求信息放入到对应该接口的请求队列中 * 2、 该接口对应的任务类监听对应接口的请求队列,从队列中顺序取出请求信息并进行处理 *  * 优点:接口几乎在收到请求的同时就已经返回,处理程序在后台异步进行处理,大大提高吞吐量 *  *  * 
*  *
*  * @author Logen * */@RestControllerpublic class OrderController {  @Autowired  private RequestQueue queue;  @GetMapping("/order")  public DeferredResult order(String number) throws InterruptedException {    System.out.println("[ OrderController ] 接到下单请求");    System.out.println("当前待处理订单数: " + queue.getOrderQueue().size());    AsyncVo vo = new AsyncVo<>();    DeferredResult result = new DeferredResult<>();    vo.setParams(number);    vo.setResult(result);    queue.getOrderQueue().put(vo);    System.out.println("[ OrderController ] 返回下单结果");    return result;  }}

9. 运行 AsyncStarter.java ,启动测试

浏览器输入 http://localhost:8080/order?number=10001

正常情况处理3秒返回,返回结果如下

{"time":1548241500718,"params":"10001"}

观察控制台打印日志,如下所示:

[ OrderController ] 接到下单请求当前待处理订单数: 0[ OrderController ] 返回下单结果[ OrderTask ]开始处理订单[ OrderTask ]订单处理完成

结论:Controller层几乎在接收到请求的同时就已经返回,处理程序在后台异步处理任务。

快速多次刷新浏览器,目的为了高并发测试,观察控制台打印信息

现象:Controller层快速返回,待处理请求在队列中开始增加,异步处理程序在按顺序处理请求。

优点:对客户端响应时间不变,但提高了服务端的吞吐量。大大提升高并发处理性能!

相关内容

热门资讯

空调连续工作多久需要停机休息 空调连续工作的时间取决于多个因素,包括空调的型号、功率、使用环境和维护情况等。一般来说,空调可以连续...
空调哪个符号是除湿的 空调除湿功能通常用一个水滴形状的符号表示。不同品牌和型号的空调遥控器可能会有所差异,但一般来说,除湿...
格力空调多长时间加一次氟利昂 1、一般情况下,空调对氟利昂的使用量是非常小的,所以在空调使用了四年左右的时候再考虑加也没问题。2、...
空调用多长时间需要加氟一次 一般情况下是五年。当出现以下这几种情况时,空调才需要加氟。1、空调的室内机和室外机各有两个接口,当接...
格调生活·C标思享荟:卡萨帝在... 9月12日的郑州龙湖金融岛,夜风轻拂,灯光璀璨。“C标时刻共传奇——卡萨帝思享荟暨20周年品牌盛典郑...
品牌价值破千亿、高端份额持续领... 9月12日,“C标时刻共传奇——卡萨帝思享荟暨20周年品牌盛典郑州站”在郑州龙湖金融岛举办。2026...
空调主板坏了修800还是换新的 空调主板坏了是修还是换新,需要考虑多个因素,以下是一些建议:1. 维修成本:如果维修费用较低,比如 ...
空调开多久需要休息多长时间 问题:空调开多久需要休息多长时间回答:1、在使用空调的过程中,一般空调开7~8个小时左右需要休息2~...
中央空调多长时间清洗一次 1、一般空调开机前清洗一次即可,空调开机中间时段再清洗一次,空调关机时清洗一次,这样比较合理。2、空...
中央空调氟利昂多久加一次 看空调制热和制冷程度,很多人的观点认为氟利昂必须一年放一次,其实这个观点是错误的,理解上面有点偏差,...