这种流式异步处理本身就很难界定“完成”,而且又是多task,本身connect是没法做的吧。
如果非要做可以写个旁支程序,设置一个静默时间slienceTime(可以使redis可以是取的到的单例变量),比如task有数据时就把slienceTime赋为0,另一个定时任务每一秒把这个slienceTime+1,等到slienceTime==60时就知道connect已经60s没处理数据了