更改项目需求以及项目之前阻塞模式问题的叙述已经在上一篇说过了，详情可参考：https://www.cnblogs.com/darope/p/10276213.html ，https://yq.aliyun.com/articles/679654两篇文章的介绍。 一，关于brpop为什么要更改，这里简单分析一下原版本的阻塞代码。
1 @Override 2 public void readyForControl(Service.ControlRequest request, StreamObserver<Service.ControlResponse> responseObserver) { 3 byte[] uuidByte = request.getH().getId().toByteArray(); 4 JUUID juuid = new JUUID(uuidByte); 5 String uuid = juuid.toString(); 6 logger.info("readyForControl uuid: " + uuid); 7 // agent上线 8 Long onlineTime = System.currentTimeMillis(); 9 redisService.set(ONLINE_PREFIX + uuid, String.valueOf(System.currentTimeMillis())); 10 onlineAgent(uuid); 11 12 while (true) { 13 try { 14 //暂时没有更好的办法处理，降低两个while同时守护任务redis的可能性 15 if (needBreak(uuid, onlineTime)) { 16 break; 17 } 18 List<Task> tasks = taskRedisMap.brpop(uuid); 19 if (Objects.isNull(tasks)) { 20 continue; 21 } 22 for(Task task : tasks) { 23 //agent 重启后丢失一个任务；老的rpc通道收到任务放回机制 24 if (needBreak(uuid, onlineTime)) { 25 taskRedisMap.pushTask(task); 26 continue; 27 } 28 logger.info("task get uuid: " + uuid + " nodeId: " + task.getNodeId()); 29 30 Service.ControlResponse.ControlCmd controlCmd = Service.ControlResponse.ControlCmd.forNumber(task.getTaskType()); 31 Service.ControlResponse response = null; 32 assert controlCmd != null; 33 // 根据任务类型分配任务 34 response = getControlResponseOption(task, controlCmd, null); 35 logger.info("cmd: " + controlCmd + " nodeId " + task.getNodeId()); 36 if (Objects.isNull(response)) { 37 logger.info("empty response. nodeId " + task.getNodeId()); 38 return; 39 } 40 41 // 通知业务调用方 42 readyForControlEvent(task); 43 logger.info("readyForControlEvent"); 44 task.setTaskStatus(TaskStatusEnum.BUSY); 45 task.setStartExecTimeout(System.currentTimeMillis()); 46 task.setReceiveEvent(true); 47 taskRedisMap.update(task); 48 logger.info("onNext ..."); 49 responseObserver.onNext(response); 50 logger.info("onNext OK..."); 51 } 52 } catch (Throwable e) { 53 logger.error("readyForControl异常, uuid={}", e, uuid); 54 } 55 } 56 }