java.util.concurrent.FutureTask#cancel ( )源码实例Demo

/** Disabled when on low memory condition, or when num_preloader_threads is smaller than 1. */
static public void preload(final Collection<Patch> patches, final double mag, final boolean repaint) {
	if (low_memory_conditions || num_preloader_threads < 1) return;
	if (null == preloader) setupPreloader(null);
	else return;
	synchronized (preloads) {
		for (final FutureTask< MipMapImage > fu : preloads) fu.cancel(false);
	}
	preloads.clear();
	try {
		preloader.submit(new Runnable() { @Override
           public void run() {
			for (final Patch p : patches) preload(p, mag, repaint);
		}});
	} catch (final Throwable t) { Utils.log2("Ignoring error with preloading"); }
}
 

/**
 * Blocks until the given condition evaluates to true. The condition is evaluated every @code{frequency}
 * milliseconds, so, the given condition should be an idempotent operation.
 * If the condition is not met within the given timeout, an exception is thrown.
 *
 * @param condition the condition to wait for
 * @param timeout the timeout value
 * @param timeUnit the unit for the timeout
 * @param frequency the frequency of the condition's evaluation in milliseconds
 */
public static void until(Callable<Boolean> condition, long timeout, TimeUnit timeUnit, long frequency) {
    FutureTask<Void> futureTask = new FutureTask<Void>(() -> {
        while (!condition.call()) {
            Thread.sleep(frequency);
        }
        return null;
    });

    ExecutorService executor = Executors.newFixedThreadPool(1);
    executor.submit(futureTask);
    try {
        futureTask.get(timeout, timeUnit);
    } catch (InterruptedException | ExecutionException | TimeoutException e) {
        futureTask.cancel(true);
        e.printStackTrace();
        Assert.fail(e.getMessage());
    }
}
 

/**
 * This method maintains a lock using a file. If the creation fails we return null
 *
 * @return FSDataOutputStream object corresponding to the newly opened lock file
 * @throws IOException if IO failure occurs
 */
public static Pair<Path, FSDataOutputStream> checkAndMarkRunningHbck(Configuration conf,
    RetryCounter retryCounter) throws IOException {
  FileLockCallable callable = new FileLockCallable(conf, retryCounter);
  ExecutorService executor = Executors.newFixedThreadPool(1);
  FutureTask<FSDataOutputStream> futureTask = new FutureTask<>(callable);
  executor.execute(futureTask);
  final int timeoutInSeconds = conf.getInt(
    "hbase.hbck.lockfile.maxwaittime", DEFAULT_WAIT_FOR_LOCK_TIMEOUT);
  FSDataOutputStream stream = null;
  try {
    stream = futureTask.get(timeoutInSeconds, TimeUnit.SECONDS);
  } catch (ExecutionException ee) {
    LOG.warn("Encountered exception when opening lock file", ee);
  } catch (InterruptedException ie) {
    LOG.warn("Interrupted when opening lock file", ie);
    Thread.currentThread().interrupt();
  } catch (TimeoutException exception) {
    // took too long to obtain lock
    LOG.warn("Took more than " + timeoutInSeconds + " seconds in obtaining lock");
    futureTask.cancel(true);
  } finally {
    executor.shutdownNow();
  }
  return new Pair<Path, FSDataOutputStream>(callable.getHbckLockPath(), stream);
}
 

@Test
public void testFuture() throws Exception {
    JexlScript e = JEXL.createScript("while(true);");
    FutureTask<Object> future = new FutureTask<Object>(e.callable(null));

    ExecutorService executor = Executors.newFixedThreadPool(1);
    executor.submit(future);
    Object t = 42;
    try {
        t = future.get(100, TimeUnit.MILLISECONDS);
        Assert.fail("should have timed out");
    } catch (TimeoutException xtimeout) {
        // ok, ignore
        future.cancel(true);
    } finally {
        executor.shutdown();
    }

    Assert.assertTrue(future.isCancelled());
    Assert.assertEquals(42, t);
}
 

public void connect() throws BinlogException {
    FutureTask<Void> future = new FutureTask<Void>(() -> {
        handshake();
        return null;
    });

    MySQLExecuteService.connExecutor.execute(future);
    try {
        future.get(MySQLExecutor.EXECUTE_TIMEOUT, TimeUnit.MILLISECONDS);
    } catch (Throwable e) {
        future.cancel(true);
        throw new BinlogException(ErrorCode.WARN_MySQL_HANDSHAKE, e, username + "/****");
    }
}
 

public ResultSetPacket execute(final String sql) throws IOException {
    FutureTask<ResultSetPacket> future = new FutureTask<ResultSetPacket>(new Callable<ResultSetPacket>() {
        public ResultSetPacket call() throws Exception {
            return query(sql);
        }
    });
    MySQLExecuteService.connExecutor.execute(future);

    try {
        return future.get(EXECUTE_TIMEOUT, TimeUnit.MILLISECONDS);
    } catch (Throwable e) {
        future.cancel(true);
        throw new IOException("sql: [" + sql + "] execute timeout");
    }
}
 

@Override
public Future<V> apply(InvocationContext<Future<V>> ctx) {
    FutureTask<Future<V>> task = new NamedFutureTask<>("FutureExecution", () -> delegate.apply(ctx));
    executor.execute(task);
    return new Future<V>() {
        @Override
        public boolean cancel(boolean mayInterruptIfRunning) {
            ctx.fireEvent(CancellationEvent.INSTANCE);
            return task.cancel(mayInterruptIfRunning);
        }

        @Override
        public boolean isCancelled() {
            return task.isCancelled();
        }

        @Override
        public boolean isDone() {
            return task.isDone();
        }

        @Override
        public V get() throws InterruptedException, ExecutionException {
            return task.get().get();
        }

        @Override
        public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
            // at worst, the timeout here could possibly be 2x the requested value
            return task.get(timeout, unit).get(timeout, unit);
        }
    };
}
 

@Override
public void actionPerformed(ActionEvent e) {
    setEnabled(false); // discourage repeated clicking

    FutureTask<DockerImage> actionTask;
    synchronized (this) {
        actionTask = task;
    }

    if (actionTask != null) {
        actionTask.cancel(true);
    }
}
 

/**
 * This is not intended to be Overridden.
 * <p>
 * This method creates a {@link FutureTask} using the {@link Runnable} provided in the concrete implementations.
 * If the job is cancelled, the executor cancels the task (this uses thread interrupt).
 *
 * @param monitor Monitor for reporting progress
 * @throws Exception on error
 */
@Override
public void run(JobMonitor monitor) throws Exception
{
    monitor.beginTask(getName());

    FutureTask task = new FutureTask(getRunnable(), null);

    try {
        executorService.submit(task);
        int count = 0;
        while (!task.isDone())
        {
            if(monitor.isCanceled())
            {
                task.cancel(true);
            } else {
                monitor.updateTaskName(getName() + " running for : " + count + " seconds");
                Thread.currentThread().sleep(1000);
                count++;
            }
        }
        monitor.done();
    } catch (Exception e)
    {
        errorHandler.accept("Failed to complete " + getName(), e);
    }
}
 

/**
 * Calculates the statistics of the given {@link ExampleSet} in a {@link ProgressThread}. Once the statistics are
 * calculated, will update the stats on all {@link AttributeStatisticsPanel}s.
 *
 * @param exampleSet
 * 		the example of which to recalculate the statistics
 */
private void calculateStatistics(final ExampleSet exampleSet) {

	// wrap into a future task so that cancelling with an interrupt is possible
	FutureTask<Void> task = new FutureTask<>(() -> {
		exampleSet.recalculateAllAttributeStatistics();
		barrier.countDown();
		return null;
	});

	//execute with indeterminate progress thread
	worker = new ProgressThread("statistics_calculation") {

		@Override
		public void run() {
			task.run();
		}

		@Override
		protected void executionCancelled() {
			task.cancel(true);
			aborted = true;
			barrier.countDown();
		}
	};
	worker.setIndeterminate(true);
	worker.start();
}
 

/**
 * Variant method supporting a timeout.
 * 
 * @param task
 * @param timeout
 * @param unit
 * @throws InterruptedException
 * @throws ExecutionException
 * @throws TimeoutException
 * 
 *             TODO Add variants of memberAdd() and friends that accept
 *             a timeout. They should use this method rather than
 *             {@link #runActorTask(ActorTask)}.
 */
private void runActorTask(final ActorTask task, final long timeout,
        final TimeUnit unit) throws InterruptedException,
        ExecutionException, TimeoutException {
    if (!singleThreadActor) {
        /*
         * Timeout support requires an executor service to run the actor
         * tasks.
         */
        throw new UnsupportedOperationException();
    }
    final FutureTask<Void> ft = new FutureTaskMon<Void>(task);
    synchronized (knownActorTasks) {
        if (!knownActorTasks.add(ft))
            throw new AssertionError();
    }
    getActorExecutor().execute(ft);
    try {
        ft.get(timeout, unit);
    } finally {
        ft.cancel(true/* mayInterruptIfRunning */);
        synchronized (knownActorTasks) {
            if (!knownActorTasks.remove(ft))
                throw new AssertionError();
        }
    }
}
 

public void cancel(final long messageId) {
    final FutureTask<Message> future = task.remove(messageId);
    if (future != null) {
        LOGGER.error(String.format("Cancel Message unexpected (id=%s).", messageId));
        future.cancel(true);
    }
}
 

@Override
public void cancelDraw()
{
    super.cancelDraw();

    FutureTask task = (FutureTask) mDrawThreadTask;
    if (null != task) {
        task.cancel(true);
    }
}
 

/**
 * main entry function for fetching from servers
 */
public synchronized ByteBuffer[] fetch() {
  // clear previous results
  recvBuf = null;
  stats.clear();

  if (servers == null || servers.size() == 0 ||
      requestBuf == null || fetchTimeoutSeconds <= 0) {
    return recvBuf;
  }

  ExecutorService executor = Executors.newSingleThreadExecutor();
  MultiFetch multiFetch = new MultiFetch();
  FutureTask<?> task = new FutureTask(multiFetch, null);
  executor.execute(task);
  try {
    task.get(fetchTimeoutSeconds, TimeUnit.SECONDS);
  } catch(InterruptedException ie) {
    // attempt to cancel execution of the task.
    task.cancel(true);
    LOG.error("interrupted during fetch: "+ie.toString());
  } catch(ExecutionException ee) {
    // attempt to cancel execution of the task.
    task.cancel(true);
    LOG.error("exception during fetch: "+ee.toString());
  } catch(TimeoutException te) {
    // attempt to cancel execution of the task.  
    task.cancel(true);
    LOG.error("timeout for fetch: "+te.toString());
  }

  executor.shutdownNow();
  multiFetch.close();
  return recvBuf;
}
 

private void trackEvent(final boolean isTransaction) throws IOException {
    final MySQLConnector connector = new MySQLConnector(metaInfo.getDbInfo());
    connector.handshake();

    getNewBinlogOffset(metaInfo, connector);

    final LogDecoder decoder = new LogDecoder();
    decoder.handle(LogEvent.ROTATE_EVENT);
    decoder.handle(LogEvent.FORMAT_DESCRIPTION_EVENT);
    decoder.handle(LogEvent.QUERY_EVENT);
    decoder.handle(LogEvent.XID_EVENT);

    final LogContext context = new LogContext();
    String binlogFile = binlogInfo.getBinlogFile();
    boolean firstFlag = true;

    do {
        if (!isBinlogExist(binlogFile, connector)) {
            // binlog file not exist use the previous binlogInfo
            break;
        }
        binlogInfo.setBinlogFile(binlogFile);
        updateConnectionProps(connector);
        registerSlave(connector);
        sendDumpCommand(metaInfo.getSlaveId(), binlogInfo, connector);

        if (!firstFlag) { // mostly call this
            if (isTransaction) {
                if (isNearestCommit(decoder, binlogInfo, connector, context)) {
                    break;
                }
            } else {
                if (isNearestEvent(decoder, binlogInfo, connector, context)) {
                    break;
                }
            }
        } else {
            /**
             * 只有首次查找binlog的时候才需要开启future task
             *
             * 原因： 有可能从库最新的binlog时间都比主库dump的时间戳小，避免长时间等待
             *
             */
            firstFlag = false;
            FutureTask<Boolean> future = new FutureTask<Boolean>(new Callable<Boolean>() {
                public Boolean call() throws IOException {
                    if (isTransaction) {
                        if (isNearestCommit(decoder, binlogInfo, connector, context)) {
                            return true;
                        }
                    } else {
                        if (isNearestEvent(decoder, binlogInfo, connector, context)) {
                            return true;
                        }
                    }
                    return false;
                }
            });
            switchExecutors.execute(future);

            try {
                if (future.get(EXECUTE_TIMEOUT, TimeUnit.MILLISECONDS)) {
                    break;
                }
            } catch (Throwable e) {
                future.cancel(true);

                // timeout exception
                if (nearestCommit != null) {
                    // considering using transaction primarily
                    LogEvent event = nearestCommit;
                    binlogInfo.setBinlogPos(event.getLogPos());
                    binlogInfo.setBinlogWhen(event.getWhen());
                    break;
                }
                /**
                 * 如果一开始就超时异常 并且记录的事件都为null
                 */
                binlogInfo.setBinlogPos(DEFAULT_BINLOG_START_OFFSET);
                break;
            }
        }

        binlogFile = getPreBinlogFile(binlogInfo.getBinlogFile());
        connector.disconnect();
        connector.handshake();
    } while (true);

    connector.disconnect();
}
 

static <V> void cancel(FutureTask<V> task, boolean mayInterruptIfRunning) {
    task.cancel(mayInterruptIfRunning);
    checkCancelled(task);
}
 

/**
     * The {@link GangliaListener} can block awaiting a datagram packet. If no
     * packet arrives, then it could hang there since
     * {@link DatagramSocket#receive(java.net.DatagramPacket)} does not notice
     * an interrupt. NIO for multicast is not available in JDK 6 (it was added
     * in JDK 7). This test verifies that an interrupt is noticed and that the
     * listener terminates in a timely manner.
     */
    public void test_gangliaListener_shutdown() throws UnknownHostException,
            InterruptedException {
       
        final IGangliaMessageHandler handler = new IGangliaMessageHandler() {
            
            @Override
            public void accept(IGangliaMessage msg) {
                // Ignore.
            }
        };
        
        final GangliaListener gangliaListener = new GangliaListener(
                InetAddress.getByName(IGangliaDefaults.DEFAULT_GROUP),//
                IGangliaDefaults.DEFAULT_PORT, //
                new GangliaMessageDecoder31(),//
                handler//
                );
        
        ExecutorService executorService = null;

        FutureTask<Void> ft = null;

        try {

            executorService = Executors.newSingleThreadExecutor();

            ft = new FutureTask<Void>(gangliaListener);

            // Run the listener.
            executorService.submit(ft);

            Thread.sleep(1000/* ms */);

            assertTrue(gangliaListener.isListening());

            ft.cancel(true/* mayInterruptIfRunning */);

            Thread.sleep(1000/* ms */);

         /**
          * FIXME This assertion can not be made with Java 6 per the notes on
          * this test and on the GangliaListener implementation. Java 6 does not
          * support non-blocking IO and multicast, so the IO is blocking and the
          * interrupt is not noticed.  I have modified the test by disabling the
          * assert and linked the test to the ticket.  We should fix this by
          * a refactor of the GangliaListener to use the Java 7 support for 
          * non-blocking IO and multicast.
          * 
          * @see <a href="http://trac.bigdata.com/ticket/1188">
          *      com.bigdata.ganglia.TestGangliaListenerShutdown fails due to
          *      blocking NIO. </a>
          */
//            assertFalse(gangliaListener.isListening());
            log.error("Test is internally disabled due to lack of non-blocking IO and multicast in Java 6. See #1188.");
            
        } finally {

            /*
             * Stop host/application metric collection here.
             */
            if (executorService != null)
                executorService.shutdownNow();

        }

    }
 

/**
 * {@inheritDoc}
 * <p>
 * Note: The embedded GangliaService is executed on the main thread pool. We
 * need to terminate the GangliaService in order for the thread pool to
 * shutdown.
 */
@Override
public void stopService(final boolean immediateShutdown) {

    final FutureTask<Void> ft = gangliaFuture.getAndSet(null);

    if (ft != null) {

        ft.cancel(immediateShutdown/* mayInterruptIfRunning */);

    }

    // Clear the state reference.
    gangliaService.set(null);

}
 

/**
 * @param offset
 * @param limit
 * @param capacity
 */
public InnerIterator1(final long offset, final long limit,
        final int capacity) {

    // this.offset = offset;
    //
    // this.limit = limit;
    //
    // this.capacity = capacity;

    this.buffer = new BlockingBuffer<ISPO>(sourceAccessPath
            .getChunkCapacity());

    FutureTask<Void> future = null;
    try {

        /*
         * Note: We do NOT get() this Future. This task will run
         * asynchronously.
         * 
         * The Future is canceled IF (hopefully WHEN) the iterator
         * is closed.
         * 
         * If the task itself throws an error, then it will use
         * buffer#abort(cause) to notify the buffer of the cause (it
         * will be passed along to the iterator) and to close the
         * buffer (the iterator will notice that the buffer has been
         * closed as well as that the cause was set on the buffer).
         *
         * @see <a href="https://sourceforge.net/apps/trac/bigdata/ticket/707">
         *      BlockingBuffer.close() does not unblock threads </a>
         */

        // Wrap task as FutureTask.
        future = new FutureTask<Void>(newRunIteratorsTask(buffer));

        // set the future on the BlockingBuffer.
        buffer.setFuture(future);

        // submit task for execution.
        sourceAccessPath.getIndexManager().getExecutorService()
                .submit(future);

        /*
         * The outer access path will impose the "DISTINCT SPO"
         * constraint.
         */
        // /*
        // * Wrap the asynchronous iterator with one that imposes
        // * a distinct (s,p,o) filter.
        // */
        // src = sourceAccessPath.getRelation()
        // .distinctSPOIterator(buffer.iterator());
        final IFilter filter = sourceAccessPath.getPredicate()
                .getAccessPathFilter();
        if (filter != null) {
            src = new ChunkedWrappedIterator<ISPO>(new Striterator(
                    buffer.iterator()).addFilter(filter));
        } else {
            src = buffer.iterator();
        }

    } catch (Throwable ex) {

        try {

            buffer.close();

            if (future != null) {

                future.cancel(true/* mayInterruptIfRunning */);

            }

        } catch (Throwable t) {

            log.error(t, t);

        }

        throw new RuntimeException(ex);

    }

}
 

/**
 * The value expression for the SERVICE reference is a constant (fast
 * path).
 * 
 * @throws Exception
 */
private void doServiceCallWithConstant() throws Exception {

    final BigdataURI serviceURI = ServiceCallUtility
            .getConstantServiceURI(serviceRef);

    if (serviceURI == null)
        throw new AssertionError();
    
    // Lookup a class to "talk" to that Service URI.
    final ServiceCall<? extends Object> serviceCall = resolveService(serviceURI);

    try {

        final ICloseableIterator<IBindingSet[]> sitr = context
                .getSource();

        while (sitr.hasNext()) {

            final IBindingSet[] chunk = sitr.next();

            final ServiceCallChunk serviceCallChunk = new ServiceCallChunk(
                    serviceURI, serviceCall, chunk);

            final FutureTask<Void> ft = new FutureTask<Void>(
                    new ServiceCallTask(serviceCallChunk));
            
            context.getExecutorService().execute(ft);
            
            try {

                ft.get(timeout, TimeUnit.MILLISECONDS);
                
            } catch (TimeoutException ex) {
                
                if (!silent)
                    throw ex;
                
            } finally {

                ft.cancel(true/* mayInterruptIfRunning */);
                
            }

        }

        // Flush the sink.
        context.getSink().flush();
        
        // Done.
        return;

    } finally {
        
        context.getSource().close();

        context.getSink().close();

    } 
    
}