@Override
    public void execute(Tuple tuple) {
        speed++;
        if (speed == 1){
            ptime = System.currentTimeMillis();
        }
        if (logger.isInfoEnabled()){
            logger.info(format("updateTimeBolt received:%s,%s", tuple.getString(0), tuple.getLong(1)));
        }
        long time = tuple.getLong(1);
        if (now < time) {
            if (logger.isInfoEnabled()) {
                logger.info(format("[%s]:%s", UPDATE_TIME, tuple.toString()));
            }
            this.outputCollector.emit(UPDATE_TIME, new Values(time));
            now = time;
            if (logger.isInfoEnabled()){
                logger.info(format("updateTimeBolt sent:%s,%s", tuple.getString(0), tuple.getLong(1)));
            }
        } else {
            if (logger.isInfoEnabled()){
                logger.info(format("now >= time: %s >= %s", now, time));
            }
        }

        this.outputCollector.ack(tuple);

        if (speed % 1000 == 0){
            if (logger.isInfoEnabled()){
                logger.info(format("1000 cost time:%s", System.currentTimeMillis()-ptime));
            }
            speed = 0;
        }
    }