org.elasticsearch.threadpool.ThreadPool.generic()方法的使用及代码示例

x33g5p2x  于2022-01-30 转载在 其他  
字(10.7k)|赞(0)|评价(0)|浏览(195)

本文整理了Java中org.elasticsearch.threadpool.ThreadPool.generic()方法的一些代码示例，展示了ThreadPool.generic()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。ThreadPool.generic()方法的具体详情如下：
包路径：org.elasticsearch.threadpool.ThreadPool
类名称：ThreadPool
方法名：generic

ThreadPool.generic介绍

[英]Get the generic ExecutorService. This executor service Executor#execute(Runnable) method will run the Runnable it is given in the ThreadContext of the thread that queues it.

Warning: this ExecutorService will not throw RejectedExecutionExceptionif you submit a task while it shutdown. It will instead silently queue it and not run it.
[中]获取通用执行器服务。这个executor service executor#execute（Runnable）方法将运行Runnable，它是在队列线程的线程上下文中给出的。
警告：如果在任务关闭时提交任务，此ExecutorService将不会抛出RejectedExecutionException。相反，它将静默地对其排队，而不是运行它。

代码示例

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. /**
  2.  * The executor service for this transport service.
  3.  *
  4.  * @return the executor service
  5.  */
  6. private ExecutorService getExecutorService() {
  7.   return threadPool.generic();
  8. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. private void notifyPingReceived(final PingRequest pingRequest) {
  2.   threadPool.generic().execute(new Runnable() {
  3.     @Override
  4.     public void run() {
  5.       for (Listener listener : listeners) {
  6.         listener.onPingReceived(pingRequest);
  7.       }
  8.     }
  9.   });
  10. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. @Override
  2.   public void onNodeDisconnected(DiscoveryNode node) {
  3.     AbstractRunnable runnable = new AbstractRunnable() {
  4.       @Override
  5.       public void onFailure(Exception e) {
  6.         logger.warn("failed to handle transport disconnect for node: {}", node);
  7.       }
  8.       @Override
  9.       protected void doRun() {
  10.         handleTransportDisconnect(node);
  11.       }
  12.     };
  13.     threadPool.generic().execute(runnable);
  14.   }
  15. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. @Override
  2.   public void accept(final IndexShard.ShardFailure shardFailure) {
  3.     final ShardRouting shardRouting = shardFailure.routing;
  4.     threadPool.generic().execute(() -> {
  5.       synchronized (IndicesClusterStateService.this) {
  6.         failAndRemoveShard(shardRouting, true, "shard failure, reason [" + shardFailure.reason + "]", shardFailure.cause,
  7.           clusterService.state());
  8.       }
  9.     });
  10.   }
  11. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. private void notifyNodeFailure(final DiscoveryNode node, final String reason) {
  2.   try {
  3.     threadPool.generic().execute(new Runnable() {
  4.       @Override
  5.       public void run() {
  6.         for (Listener listener : listeners) {
  7.           listener.onNodeFailure(node, reason);
  8.         }
  9.       }
  10.     });
  11.   } catch (EsRejectedExecutionException ex) {
  12.     logger.trace(() -> new ParameterizedMessage(
  13.         "[node  ] [{}] ignoring node failure (reason [{}]). Local node is shutting down", node, reason), ex);
  14.   }
  15. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. @Override
  2.   protected void handleMergeException(final Directory dir, final Throwable exc) {
  3.     engineConfig.getThreadPool().generic().execute(new AbstractRunnable() {
  4.       @Override
  5.       public void onFailure(Exception e) {
  6.         logger.debug("merge failure action rejected", e);
  7.       }
  8.       @Override
  9.       protected void doRun() throws Exception {
  10.         /*
  11.          * We do this on another thread rather than the merge thread that we are initially called on so that we have complete
  12.          * confidence that the call stack does not contain catch statements that would cause the error that might be thrown
  13.          * here from being caught and never reaching the uncaught exception handler.
  14.          */
  15.         failEngine("merge failed", new MergePolicy.MergeException(exc, dir));
  16.       }
  17.     });
  18.   }
  19. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. @Override
  2. protected void onTimeout(List<? extends BatchedTask> tasks, TimeValue timeout) {
  3.   threadPool.generic().execute(
  4.     () -> tasks.forEach(
  5.       task -> ((UpdateTask) task).listener.onFailure(task.source,
  6.         new ProcessClusterEventTimeoutException(timeout, task.source))));
  7. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. private void notifyMasterFailure(final DiscoveryNode masterNode, final Throwable cause, final String reason) {
  2.   if (notifiedMasterFailure.compareAndSet(false, true)) {
  3.     try {
  4.       threadPool.generic().execute(() -> {
  5.         for (Listener listener : listeners) {
  6.           listener.onMasterFailure(masterNode, cause, reason);
  7.         }
  8.       });
  9.     } catch (EsRejectedExecutionException e) {
  10.       logger.error("master failure notification was rejected, it's highly likely the node is shutting down", e);
  11.     }
  12.     stop("master failure, " + reason);
  13.   }
  14. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. private void handleResponse(final int responseSlot, final MultiSearchResponse.Item item) {
  2.   responses.set(responseSlot, item);
  3.   if (responseCounter.decrementAndGet() == 0) {
  4.     assert requests.isEmpty();
  5.     finish();
  6.   } else {
  7.     if (thread == Thread.currentThread()) {
  8.       // we are on the same thread, we need to fork to another thread to avoid recursive stack overflow on a single thread
  9.       threadPool.generic().execute(() -> executeSearch(requests, responses, responseCounter, listener));
  10.     } else {
  11.       // we are on a different thread (we went asynchronous), it's safe to recurse
  12.       executeSearch(requests, responses, responseCounter, listener);
  13.     }
  14.   }
  15. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. /** starts a new joining thread if there is no currently active one and join thread controlling is started */
  2. public void startNewThreadIfNotRunning() {
  3.   assert Thread.holdsLock(stateMutex);
  4.   if (joinThreadActive()) {
  5.     return;
  6.   }
  7.   threadPool.generic().execute(new Runnable() {
  8.     @Override
  9.     public void run() {
  10.       Thread currentThread = Thread.currentThread();
  11.       if (!currentJoinThread.compareAndSet(null, currentThread)) {
  12.         return;
  13.       }
  14.       while (running.get() && joinThreadActive(currentThread)) {
  15.         try {
  16.           innerJoinCluster();
  17.           return;
  18.         } catch (Exception e) {
  19.           logger.error("unexpected error while joining cluster, trying again", e);
  20.           // Because we catch any exception here, we want to know in
  21.           // tests if an uncaught exception got to this point and the test infra uncaught exception
  22.           // leak detection can catch this. In practise no uncaught exception should leak
  23.           assert ExceptionsHelper.reThrowIfNotNull(e);
  24.         }
  25.       }
  26.       // cleaning the current thread from currentJoinThread is done by explicit calls.
  27.     }
  28.   });
  29. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. threadPool.generic().execute(() -> {
  2.   closeLock.writeLock().lock();
  3.   try {

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. public void startRecovery(final IndexShard indexShard, final DiscoveryNode sourceNode, final RecoveryListener listener) {
  2.   // create a new recovery status, and process...
  3.   final long recoveryId = onGoingRecoveries.startRecovery(indexShard, sourceNode, listener, recoverySettings.activityTimeout());
  4.   threadPool.generic().execute(new RecoveryRunner(recoveryId));
  5. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. private void performStateRecovery(boolean enforceRecoverAfterTime, String reason) {
  2.   final Gateway.GatewayStateRecoveredListener recoveryListener = new GatewayRecoveryListener();
  3.   if (enforceRecoverAfterTime && recoverAfterTime != null) {
  4.     if (scheduledRecovery.compareAndSet(false, true)) {
  5.       logger.info("delaying initial state recovery for [{}]. {}", recoverAfterTime, reason);
  6.       threadPool.schedule(recoverAfterTime, ThreadPool.Names.GENERIC, () -> {
  7.         if (recovered.compareAndSet(false, true)) {
  8.           logger.info("recover_after_time [{}] elapsed. performing state recovery...", recoverAfterTime);
  9.           gateway.performStateRecovery(recoveryListener);
  10.         }
  11.       });
  12.     }
  13.   } else {
  14.     if (recovered.compareAndSet(false, true)) {
  15.       threadPool.generic().execute(new AbstractRunnable() {
  16.         @Override
  17.         public void onFailure(Exception e) {
  18.           logger.warn("Recovery failed", e);
  19.           // we reset `recovered` in the listener don't reset it here otherwise there might be a race
  20.           // that resets it to false while a new recover is already running?
  21.           recoveryListener.onFailure("state recovery failed: " + e.getMessage());
  22.         }
  23.         @Override
  24.         protected void doRun() throws Exception {
  25.           gateway.performStateRecovery(recoveryListener);
  26.         }
  27.       });
  28.     }
  29.   }
  30. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. final CountDownLatch latch = new CountDownLatch(1);
  2. threadPool.generic().execute(() -> {
  3.   closeLock.writeLock().lock();
  4.   try {

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. try (ThreadContext.StoredContext ignore = threadContext.stashContext()) {
  2.   threadContext.markAsSystemContext();
  3.   threadPool.generic().execute(() -> upgradeTemplates(changes.get().v1(), changes.get().v2()));

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. threadPool.generic().execute(pingSender);
  2. threadPool.schedule(TimeValue.timeValueMillis(scheduleDuration.millis() / 3), ThreadPool.Names.GENERIC, pingSender);
  3. threadPool.schedule(TimeValue.timeValueMillis(scheduleDuration.millis() / 3 * 2), ThreadPool.Names.GENERIC, pingSender);

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. case EXISTING_STORE:
  2.   markAsRecovering("from store", recoveryState); // mark the shard as recovering on the cluster state thread
  3.   threadPool.generic().execute(() -> {
  4.     try {
  5.       if (recoverFromStore()) {
  6.   markAsRecovering("from snapshot", recoveryState); // mark the shard as recovering on the cluster state thread
  7.   SnapshotRecoverySource recoverySource = (SnapshotRecoverySource) recoveryState.getRecoverySource();
  8.   threadPool.generic().execute(() -> {
  9.     try {
  10.       final Repository repository = repositoriesService.repository(recoverySource.snapshot().getRepository());
  11.     assert requiredShards.isEmpty() == false;
  12.     markAsRecovering("from local shards", recoveryState); // mark the shard as recovering on the cluster state thread
  13.     threadPool.generic().execute(() -> {
  14.       try {
  15.         if (recoverFromLocalShards(mappingUpdateConsumer, startedShards.stream()

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. private void submitStateUpdateTask(final String source, final ClusterStateTaskConfig config,
  2.                   final Function<ClusterState, ClusterState> executor,
  3.                   final ClusterApplyListener listener) {
  4.   if (!lifecycle.started()) {
  5.     return;
  6.   }
  7.   try {
  8.     UpdateTask updateTask = new UpdateTask(config.priority(), source, new SafeClusterApplyListener(listener, logger), executor);
  9.     if (config.timeout() != null) {
  10.       threadPoolExecutor.execute(updateTask, config.timeout(),
  11.         () -> threadPool.generic().execute(
  12.           () -> listener.onFailure(source, new ProcessClusterEventTimeoutException(config.timeout(), source))));
  13.     } else {
  14.       threadPoolExecutor.execute(updateTask);
  15.     }
  16.   } catch (EsRejectedExecutionException e) {
  17.     // ignore cases where we are shutting down..., there is really nothing interesting
  18.     // to be done here...
  19.     if (!lifecycle.stoppedOrClosed()) {
  20.       throw e;
  21.     }
  22.   }
  23. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. /**
  2.  * Executed on the node that should be running the task to find and return the running task. Falls back to
  3.  * {@link #getFinishedTaskFromIndex(Task, GetTaskRequest, ActionListener)} if the task isn't still running.
  4.  */
  5. void getRunningTaskFromNode(Task thisTask, GetTaskRequest request, ActionListener<GetTaskResponse> listener) {
  6.   Task runningTask = taskManager.getTask(request.getTaskId().getId());
  7.   if (runningTask == null) {
  8.     // Task isn't running, go look in the task index
  9.     getFinishedTaskFromIndex(thisTask, request, listener);
  10.   } else {
  11.     if (request.getWaitForCompletion()) {
  12.       // Shift to the generic thread pool and let it wait for the task to complete so we don't block any important threads.
  13.       threadPool.generic().execute(new AbstractRunnable() {
  14.         @Override
  15.         protected void doRun() throws Exception {
  16.           taskManager.waitForTaskCompletion(runningTask, waitForCompletionTimeout(request.getTimeout()));
  17.           waitedForCompletion(thisTask, request, runningTask.taskInfo(clusterService.localNode().getId(), true), listener);
  18.         }
  19.         @Override
  20.         public void onFailure(Exception e) {
  21.           listener.onFailure(e);
  22.         }
  23.       });
  24.     } else {
  25.       TaskInfo info = runningTask.taskInfo(clusterService.localNode().getId(), true);
  26.       listener.onResponse(new GetTaskResponse(new TaskResult(false, info)));
  27.     }
  28.   }
  29. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. threadPool.generic().execute(new AbstractRunnable() {
  2.   @Override
  3.   public void onFailure(Exception e) {

内容来源于网络，如有侵权，请联系作者删除！

threadpool

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com