DeferredResult的使用场景及用法

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

这里说是推送，其实仍是基于请求-响应机制，只不过发起的请求会在服务端挂起，直到请求超时或服务端有数据推送时才会作出响应，响应的时机彻底由服务端控制。因此，总体效果看起来就像是服务端真的在“实时推送”同样。

能够利用SpringMVC的DeferredResult来实现异步长链接的服务端实时推送。

核心代码

Controller

@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;
}

web.xml

<servlet>
    <servlet-name>springServlet</servlet-name>
    <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
    <init-param>
        <param-name>contextConfigLocation</param-name>
        <param-value>/WEB-INF/spring-mvc.xml</param-value>
    </init-param>
    <load-on-startup>1</load-on-startup>
    <async-supported>true</async-supported> <!-- 添加异步支持 -->
</servlet>
<servlet-mapping>
    <servlet-name>springServlet</servlet-name>
    <url-pattern>/</url-pattern>
</servlet-mapping>

确保你使用的是servlet 3+

js请求

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);
    }

}

---

代码就是这些了，好好理一下思路，实现你本身的功能吧！