首页  流程图  详情

Trino--coordinator 线程模型

2024-12-09 12:19:38   0  举报





分布式

分布式数据库

作者其他创作

大纲/内容

io.airlift.concurrent.Threads

ContextClassLoaderThreadFactory

ThreadFactory daemonThreadsNamed( String nameFormat)

DispatchManager

QueryTracker<DispatchQuery> queryTracker

void start()

corePoolSize

queryManagerExecutorPoolSize

保留在池中的线程数，即使它们是空闲的，除非设置了 allowCoreThreadTimeOut

maximumPoolSize

int MAX_VALUE = 0x7fffffff

线程池运行的最大线程数

keepAliveTime

DEFAULT_KEEPALIVE_MILLIS = 10L

keepAliveTime 当线程数大于核心时，这是多余的空闲线程在终止前等待新任务的最长时间。

unit

MILLISECONDS

keepAliveTime 的单位

workQueue

DelayedWorkQueue

每次取出队首延迟最小的任务，超时等待，然后执行；

threadFactory

DefaultThreadFactory

创建线程的工程

DispatchExecutor

ListeningScheduledExecutorService scheduledExecutor

ScheduledThreadPoolExecutor

ThreadPoolExecutor

ListeningScheduledExecutorService listeningDecorator( ScheduledExecutorService delegate)

ListeningExecutorService listeningDecorator( ExecutorService delegate)

java.util.concurrent.Executors.DefaultThreadFactory

ScheduledListeningDecorator

ScheduledExecutorService delegate

ListeningExecutorService executor

Executor dispatchExecutor

创建一个线程池，该线程池根据需要创建新线程，但会在可用时重用以前构造的线程，并在需要时使用提供的 ThreadFactory 创建新线程。

runState 提供主要的生命周期控制，具有以下值： * RUNNING：接受新任务并处理排队的任务 * SHUTDOWN：不接受新任务，但处理排队的任务 * STOP：不接受新任务，不处理排队的任务，中断进行中的任务 * TIDYING：所有任务都已终止，workerCount 为零 *转换到 TIDYING 状态 *运行 terminated() 钩子方法 * TERMINATED: terminated() 已经完成状态之间的数字顺序很重要，以便进行有序比较。 runState 随时间单调增加，但不需要达到每个状态。状态转换： *调用 shutdown() 时：RUNNING -> SHUTDOWN *调用 shutdownNow() 时：(RUNNING or SHUTDOWN) -> STOP *当任务队列和线程池都为空时：SHUTDOWN -> TIDYING *当线程池为空时：STOP -> TIDYING *当 terminated() 钩子方法完成时：TIDYING -> TERMINATED

该线程池根据需要创建新线程，但会在可用时重用以前构造的线程，并在需要时使用提供的 ThreadFactory 创建新线程。

QueuedStatementResource

DispatchManager dispatchManager

Rest API

查询后用于垃圾收集的线程池的大小。来自此池的线程用于从取消的查询中释放资源，以及强制执行内存限制、查询超时等。更多线程将允许更有效的内存管理，因此在某些情况下可能有助于避免内存不足异常。但是，拥有更多线程也可能会增加用于垃圾收集的 CPU 使用率，并且即使线程无事可做，也会有额外的恒定内存成本。

60L

SECONDS

SynchronousQueue（线程安全队列）

workQueue 在任务执行前用于保存任务的队列。该队列将仅保存由 execute 方法提交的 Runnable 任务。

创建线程的工厂

ThreadFactory threadFactory

Executors

dispatcher-query

ExecutorService newCachedThreadPool( ThreadFactory threadFactory)

Thread newThread(Runnable r)

CoordinatorModule

dispatchExecutor

启动线程执行 SQL

void execute(Runnable command)

corePoolSize = 0，不缓存线程maximumPoolSize，为最大值keepAliveTime，线程数大于corePoolSize 时，空闲线程的生存时间为1分钟（空闲线程：没有拿到任务的线程）整体策略：构建弹性的，按需分配的线程，及时消灭空闲的线程

ListeningScheduledExecutorService

@ResourceSecurity(PUBLIC) @GET @Path(\"queued/{queryId}/{slug}/{token}\") @Produces(APPLICATION_JSON) public void getStatus( @PathParam(\"queryId\

Executor

在将来的某个时间执行给定的命令。该命令可以在新线程、池线程或调用线程中执行，由 Executor 实现自行决定。

ScheduledExecutorService

启动定时任务，执行任务数：不少于5个线程

ThreadFactory delegate

QueryManagerConfig

int queryManagerExecutorPoolSize = 5

DecoratingListeningExecutorService

ListeningExecutorService delegate

ThreadFactory

Thread newThread(Runnable r)

1. 如果正在运行的线程少于 corePoolSize，尝试使用给定命令启动一个新线程作为其第一个任务。2. 如果一个任务可以成功排队，那么我们仍然需要仔细检查我们是否应该添加一个线程（因为自从上次检查后现有的线程已经死了）或者池在进入这个方法后关闭了。3.如果我们不能排队任务，那么我们尝试添加一个新线程。如果它失败了，我们知道我们已经关闭或饱和，因此拒绝该任务。

ThreadFactory delegate

void execute(Runnable command)

ListeningExecutorService

保留在池中的线程数，即使它们是空闲的，除非设置了 allowCoreThreadTimeOut

int MAX_VALUE = 0x7fffffff

keepAliveTime 当线程数大于核心时，这是多余的空闲线程在终止前等待新任务的最长时间。

SynchronousQueue

workQueue 在任务执行前用于保存任务的队列。该队列将仅保存由 execute 方法提交的 Runnable 任务。

ExecutorService

dispatch-executor

initialDelay

delay