Java学习

DeferredResult的使用场景及用法

2021-02-18  本文已影响0人  jackcooper

场景

假设我们现在要实现这样一个功能:浏览器要实时展示服务端计算出来的数据。
一种可能的实现是:浏览器频繁(例如定时1秒)向服务端发起请求以获得服务端数据。但定时请求并不能“实时”反应服务端的数据变化情况。
若定时周期为S,则数据延迟周期最大即为S。若想缩短数据延迟周期,则应使S尽量小,而S越小,浏览器向服务端发起请求的频率越高,又造成网络握手次数越多,影响了效率。因此,此场景应使用服务端实时推送技术。

这里说是推送,其实还是基于请求-响应机制,只不过发起的请求会在服务端挂起,直到请求超时或服务端有数据推送时才会做出响应,响应的时机完全由服务端控制。所以,整体效果看起来就像是服务端真的在“实时推送”一样。

可以利用SpringMVC的DeferredResult来实现异步长连接的服务端实时推送。

入门Demo

后端代码

@RequestMapping("/call")
@ResponseBody
public DeferredResult<Object> call() { // 泛型Object表示返回结果的类型
    DeferredResult<Object> response = new DeferredResult<Object>( 
        10000, // 请求的超时时间 
        null); // 超时后响应的结果
    response.onCompletion(new Runnable() {
        
        @Override
        public void run() {
            // 请求处理完成后所做的一些工作
        }
    });
    // 设置响应结果
    // 调用此方法时立即向浏览器发出响应;未调用时请求被挂起
    response.setResult(new Object());
    return response;
}

前端代码

var loopCall = function() {
    $.get("${yourContext}/call", function(r) {
        loopCall();
        console.log("call: ");
        console.log(r);
    });
};
loopCall(); // 循环发起异步请求

执行逻辑

  1. 浏览器发起异步请求
  2. 请求到达服务端被挂起(使用浏览器查看请求状态,此时为pending)
  3. 向浏览器进行响应,分为两种情况:
    3.1 调用DeferredResult.setResult(),请求被唤醒,返回结果
    3.2 超时,返回一个你设定的结果
  4. 浏览得到响应,再次重复1,处理此次响应结果

真实场景

示例:浏览器向A系统发起异步长连接请求,等到B系统给A推送数据时,A会立刻向浏览器响应结果。

数据格式

定义AB之间传输数据的格式

public interface DeferredData {
    String getId(); // 唯一标识
}

DeferredResult的持有者

public interface DeferredResultHolder<DeferredData> {
    DeferredResult<DeferredData> newDeferredResult(String key, long timeout, Object timeoutResult);
    void add(String key, DeferredResult<DeferredData> deferredResult);
    DeferredResult<DeferredData> get(String key);
    void remove(String key);
    void handleDeferredData(DeferredData deferredData);
}

DeferredResult的持有者实现

public class SimpleDeferredResultHolder implements DeferredResultHolder<DeferredData> {

    private Map<String, DeferredResult<DeferredData>> deferredResults = new ConcurrentHashMap<String, DeferredResult<DeferredData>>();
    
    public DeferredResult<DeferredData> newDeferredResult(String key) {
        return newDeferredResult(key, 30 * 1000L, null);
    }
    
    public DeferredResult<DeferredData> newDeferredResult(String key, long timeout) {
        return newDeferredResult(key, timeout, null);
    }
    
    public DeferredResult<DeferredData> newDeferredResult(String key, Object timeoutResult) {
        return newDeferredResult(key, 30 * 1000L, timeoutResult);
    }
    
    @Override
    public DeferredResult<DeferredData> newDeferredResult(String key, long timeout, Object timeoutResult) {
        DeferredResult<DeferredData> deferredResult = new DeferredResult<DeferredData>(timeout, timeoutResult);
        add(key, deferredResult);
        deferredResult.onCompletion(new Runnable() {
            
            @Override
            public void run() {
                remove(key);
            }
        });
        return deferredResult;
    }

    @Override
    public void add(String key, DeferredResult<DeferredData> deferredResult) {
        deferredResults.put(key, deferredResult);
    }

    @Override
    public DeferredResult<DeferredData> get(String key) {
        return deferredResults.get(key);
    }

    @Override
    public void remove(String key) {
        deferredResults.remove(key);
    }
    
    @Override
    public void handleDeferredData(DeferredData deferredData) {
        String key = deferredData.getId();
        DeferredResult<DeferredData> deferredResult = get(key);
        if (deferredResult != null) {
            deferredResult.setResult(deferredData);
        }
    }

}

消息发送和消费

用mq或dubbo等技术发送都可以,这里用rabbitmq做示例。
如果消费的消费者做了集群部署,则只能使用mq的topic机制分发,推送消息到A的所有部署节点,若使用dubbo则只能调用其中一个节点。因此,这里最好还是使用mq

消费的发送者

public interface DeferredDataProducer {
    void sendDeferredData(DeferredData deferredData);
}

消息的消费者

public interface DeferredDataConsumer {
   void consume(DeferredData deferredData) throws Exception;
}

消费的发送者的实现

public class DeferredDataMqProducer implements DeferredDataProducer {

    @Autowired
    private AmqpTemplate amqpTemplate;

    private String exchange;
    private String routingKey = "";

    public void setExchange(String exchange) {
        this.exchange = exchange;
    }

    public void setRoutingKey(String routingKey) {
        this.routingKey = routingKey;
    }

    @Override
    public void sendDeferredData(DeferredData deferredData) {
        amqpTemplate.convertAndSend(exchange, routingKey, deferredData);
    }

}

消费的消费者的实现

public class DeferredDataMqConsumer implements DeferredDataConsumer {

    private DeferredResultHolder<DeferredNotification> deferredResultHolder;
    
    public void setDeferredResultHolder(DeferredResultHolder<DeferredNotification> deferredResultHolder) {
        this.deferredResultHolder = deferredResultHolder;
    }
    
    @Override
    public void consume(DeferredData deferredData) throws Exception;
        deferredResultHolder.handleDeferredData(deferredData);
    }
}

具体的数据格式

通知对象

public class Notification {
    private String to; // 接收者
    private String content; // 内容
    
    // 省略getter和setter方法
}

Notification的DeferredData适配器

public DeferredNotification extends Notification implements DeferredResponse {
    
    @Override
    public String getId() {
        return getTo();
    }
    
}

B的逻辑

消息的发送者的实现

public class NotificationProducer implements DeferredResponseProducer<DeferredNotification> {

    @Autowired
    private AmqpTemplate amqpTemplate;
    
    @Override
    public void sendMessage(DeferredNotification notification, String exchange) {
        amqpTemplate.convertAndSend(exchange, "", notification);
    }
}

B的Service

@Autowired
private DeferredDataProducer<DeferredNotification> deferredDataProducer;

public void test() {
    DeferredNotification n = new DeferredNotification();
    n.setTo("abc"); // 会员
    n.setContent("哈哈,我是从admin推送过来的");
    deferredDataProducer.sendDeferredData(n);
}

A的逻辑

js请求

var loopCall = function() {
    $.get("${yourContext}/call", function(r) {
        loopCall();
        console.log("call: ");
        console.log(r);
    });
};
loopCall(); // 循环发起异步请求

Controller

@RequestMapping
@Controller
public class CallController {
    
    @Autowired
    private DeferredResultHolder deferredResultHolder;
    
    @RequestMapping("/call")
    @ResponseBody
    public DeferredResult<DeferredData> call() {
        String id = "abc";
        return deferredResultHolder.newDeferredResult(id, 10 * 1000L, null);
    }

}

代码就是这些了,好好理一下思路,实现你自己的功能吧!



摘自: https://my.oschina.net/ojeta/blog/806087

上一篇下一篇

猜你喜欢

热点阅读