聊聊storm worker的executor与task
序
本文主要研究一下storm worker的executor与task
Worker
storm-2.0.0/storm-client/src/jvm/org/apache/storm/daemon/worker/Worker.java
public static void main(String[] args) throws Exception { Preconditions.checkArgument(args.length == 5, "Illegal number of arguments. Expected: 5, Actual: " + args.length); String stormId = args[0]; String assignmentId = args[1]; String supervisorPort = args[2]; String portStr = args[3]; String workerId = args[4]; Map<String, Object> conf = ConfigUtils.readStormConfig(); Utils.setupDefaultUncaughtExceptionHandler(); StormCommon.validateDistributedMode(conf); Worker worker = new Worker(conf, null, stormId, assignmentId, Integer.parseInt(supervisorPort), Integer.parseInt(portStr), workerId); worker.start(); Utils.addShutdownHookWithForceKillIn1Sec(worker::shutdown); }
- main方法创建Worker,然后调用start
Worker.start
storm-2.0.0/storm-client/src/jvm/org/apache/storm/daemon/worker/Worker.java
public void start() throws Exception { LOG.info("Launching worker for {} on {}:{} with id {} and conf {}", topologyId, assignmentId, port, workerId, ConfigUtils.maskPasswords(conf)); // because in local mode, its not a separate // process. supervisor will register it in this case // if ConfigUtils.isLocalMode(conf) returns false then it is in distributed mode. if (!ConfigUtils.isLocalMode(conf)) { // Distributed mode SysOutOverSLF4J.sendSystemOutAndErrToSLF4J(); String pid = Utils.processPid(); FileUtils.touch(new File(ConfigUtils.workerPidPath(conf, workerId, pid))); FileUtils.writeStringToFile(new File(ConfigUtils.workerArtifactsPidPath(conf, topologyId, port)), pid, Charset.forName("UTF-8")); } final Map<String, Object> topologyConf = ConfigUtils.overrideLoginConfigWithSystemProperty(ConfigUtils.readSupervisorStormConf(conf, topologyId)); ClusterStateContext csContext = new ClusterStateContext(DaemonType.WORKER, topologyConf); IStateStorage stateStorage = ClusterUtils.mkStateStorage(conf, topologyConf, csContext); IStormClusterState stormClusterState = ClusterUtils.mkStormClusterState(stateStorage, null, csContext); StormMetricRegistry.start(conf, DaemonType.WORKER); Credentials initialCredentials = stormClusterState.credentials(topologyId, null); Map<String, String> initCreds = new HashMap<>(); if (initialCredentials != null) { initCreds.putAll(initialCredentials.get_creds()); } autoCreds = ClientAuthUtils.getAutoCredentials(topologyConf); subject = ClientAuthUtils.populateSubject(null, autoCreds, initCreds); Subject.doAs(subject, (PrivilegedExceptionAction<Object>) () -> loadWorker(topologyConf, stateStorage, stormClusterState, initCreds, initialCredentials) ); }
- 这里主要是调用loadWorker
Worker.loadWorker
storm-2.0.0/storm-client/src/jvm/org/apache/storm/daemon/worker/Worker.java
private AtomicReference<List<IRunningExecutor>> executorsAtom; private Object loadWorker(Map<String, Object> topologyConf, IStateStorage stateStorage, IStormClusterState stormClusterState, Map<String, String> initCreds, Credentials initialCredentials) throws Exception { workerState = new WorkerState(conf, context, topologyId, assignmentId, supervisorPort, port, workerId, topologyConf, stateStorage, stormClusterState, autoCreds); // Heartbeat here so that worker process dies if this fails // it's important that worker heartbeat to supervisor ASAP so that supervisor knows // that worker is running and moves on doHeartBeat(); executorsAtom = new AtomicReference<>(null); // launch heartbeat threads immediately so that slow-loading tasks don't cause the worker to timeout // to the supervisor workerState.heartbeatTimer .scheduleRecurring(0, (Integer) conf.get(Config.WORKER_HEARTBEAT_FREQUENCY_SECS), () -> { try { doHeartBeat(); } catch (IOException e) { throw new RuntimeException(e); } }); workerState.executorHeartbeatTimer .scheduleRecurring(0, (Integer) conf.get(Config.EXECUTOR_METRICS_FREQUENCY_SECS), Worker.this::doExecutorHeartbeats); workerState.registerCallbacks(); workerState.refreshConnections(null); workerState.activateWorkerWhenAllConnectionsReady(); workerState.refreshStormActive(null); workerState.runWorkerStartHooks(); List<Executor> execs = new ArrayList<>(); for (List<Long> e : workerState.getLocalExecutors()) { if (ConfigUtils.isLocalMode(topologyConf)) { Executor executor = LocalExecutor.mkExecutor(workerState, e, initCreds); execs.add(executor); for (int i = 0; i < executor.getTaskIds().size(); ++i) { workerState.localReceiveQueues.put(executor.getTaskIds().get(i), executor.getReceiveQueue()); } } else { Executor executor = Executor.mkExecutor(workerState, e, initCreds); for (int i = 0; i < executor.getTaskIds().size(); ++i) { workerState.localReceiveQueues.put(executor.getTaskIds().get(i), executor.getReceiveQueue()); } execs.add(executor); } } List<IRunningExecutor> newExecutors = new ArrayList<IRunningExecutor>(); for (Executor executor : execs) { newExecutors.add(executor.execute()); } executorsAtom.set(newExecutors); //...... setupFlushTupleTimer(topologyConf, newExecutors); setupBackPressureCheckTimer(topologyConf); LOG.info("Worker has topology config {}", ConfigUtils.maskPasswords(topologyConf)); LOG.info("Worker {} for storm {} on {}:{} has finished loading", workerId, topologyId, assignmentId, port); return this; }
- 这里通过workerState.getLocalExecutors()获取List<Long> executorId的集合
- 然后通过Executor.mkExecutor创建指定数量的Executor,然后调用execute()方法转换为ExecutorShutdown,然后保存到AtomicReference<List<IRunningExecutor>> executorsAtom
WorkerState.getLocalExecutors
storm-2.0.0/storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java
// local executors and localTaskIds running in this worker final Set<List<Long>> localExecutors; public Set<List<Long>> getLocalExecutors() { return localExecutors; } public WorkerState(Map<String, Object> conf, IContext mqContext, String topologyId, String assignmentId, int supervisorPort, int port, String workerId, Map<String, Object> topologyConf, IStateStorage stateStorage, IStormClusterState stormClusterState, Collection<IAutoCredentials> autoCredentials) throws IOException, InvalidTopologyException { this.autoCredentials = autoCredentials; this.conf = conf; this.localExecutors = new HashSet<>(readWorkerExecutors(stormClusterState, topologyId, assignmentId, port)); //...... } private List<List<Long>> readWorkerExecutors(IStormClusterState stormClusterState, String topologyId, String assignmentId, int port) { LOG.info("Reading assignments"); List<List<Long>> executorsAssignedToThisWorker = new ArrayList<>(); executorsAssignedToThisWorker.add(Constants.SYSTEM_EXECUTOR_ID); Map<List<Long>, NodeInfo> executorToNodePort = getLocalAssignment(conf, stormClusterState, topologyId).get_executor_node_port(); for (Map.Entry<List<Long>, NodeInfo> entry : executorToNodePort.entrySet()) { NodeInfo nodeInfo = entry.getValue(); if (nodeInfo.get_node().equals(assignmentId) && nodeInfo.get_port().iterator().next() == port) { executorsAssignedToThisWorker.add(entry.getKey()); } } return executorsAssignedToThisWorker; } private Assignment getLocalAssignment(Map<String, Object> conf, IStormClusterState stormClusterState, String topologyId) { if (!ConfigUtils.isLocalMode(conf)) { try (SupervisorClient supervisorClient = SupervisorClient.getConfiguredClient(conf, Utils.hostname(), supervisorPort)) { Assignment assignment = supervisorClient.getClient().getLocalAssignmentForStorm(topologyId); return assignment; } catch (Throwable tr1) { //if any error/exception thrown, fetch it from zookeeper return stormClusterState.remoteAssignmentInfo(topologyId, null); } } else { return stormClusterState.remoteAssignmentInfo(topologyId, null); } }
- WorkerState在构造器里头通过readWorkerExecutors获取在本worker运行的executorIds
- 通过getLocalAssignment方法获取Assignment,然后通过get_executor_node_port方法获取Map<List<Long>, NodeInfo> executorToNodePort
- getLocalAssignment通过supervisorClient.getClient().getLocalAssignmentForStorm(topologyId)获取Assignment,如果出现异常则通过stormClusterState.remoteAssignmentInfo从zookeeper获取
StormClusterStateImpl.remoteAssignmentInfo
storm-2.0.0/storm-client/src/jvm/org/apache/storm/cluster/StormClusterStateImpl.java
public Assignment remoteAssignmentInfo(String stormId, Runnable callback) { if (callback != null) { assignmentInfoCallback.put(stormId, callback); } byte[] serialized = stateStorage.get_data(ClusterUtils.assignmentPath(stormId), callback != null); return ClusterUtils.maybeDeserialize(serialized, Assignment.class); }
- 根据topologyId从ClusterUtils.assignmentPath获取路径,然后去zookeeper获取数据
- 数据采用thrift序列化,取回来需要反序列化
ClusterUtils.assignmentPath
storm-2.0.0/storm-client/src/jvm/org/apache/storm/cluster/ClusterUtils.java
public static final String ZK_SEPERATOR = "/"; public static final String ASSIGNMENTS_ROOT = "assignments"; public static final String ASSIGNMENTS_SUBTREE = ZK_SEPERATOR + ASSIGNMENTS_ROOT; public static String assignmentPath(String id) { return ASSIGNMENTS_SUBTREE + ZK_SEPERATOR + id; }
- 路径为/assignments/{topology},比如/assignments/DemoTopology-1-1539163962
Executor.mkExecutor
storm-2.0.0/storm-client/src/jvm/org/apache/storm/executor/Executor.java
public static Executor mkExecutor(WorkerState workerState, List<Long> executorId, Map<String, String> credentials) { Executor executor; WorkerTopologyContext workerTopologyContext = workerState.getWorkerTopologyContext(); List<Integer> taskIds = StormCommon.executorIdToTasks(executorId); String componentId = workerTopologyContext.getComponentId(taskIds.get(0)); String type = getExecutorType(workerTopologyContext, componentId); if (ClientStatsUtil.SPOUT.equals(type)) { executor = new SpoutExecutor(workerState, executorId, credentials); } else { executor = new BoltExecutor(workerState, executorId, credentials); } int minId = Integer.MAX_VALUE; Map<Integer, Task> idToTask = new HashMap<>(); for (Integer taskId : taskIds) { minId = Math.min(minId, taskId); try { Task task = new Task(executor, taskId); idToTask.put(taskId, task); } catch (IOException ex) { throw Utils.wrapInRuntime(ex); } } executor.idToTaskBase = minId; executor.idToTask = Utils.convertToArray(idToTask, minId); return executor; }
- 根据组件类型创建SpoutExecutor或者BoltExecutor
- 然后创建tasks并绑定到executor
Executor.execute
storm-2.0.0/storm-client/src/jvm/org/apache/storm/executor/Executor.java
/** * separated from mkExecutor in order to replace executor transfer in executor data for testing. */ public ExecutorShutdown execute() throws Exception { LOG.info("Loading executor tasks " + componentId + ":" + executorId); String handlerName = componentId + "-executor" + executorId; Utils.SmartThread handler = Utils.asyncLoop(this, false, reportErrorDie, Thread.NORM_PRIORITY, true, true, handlerName); LOG.info("Finished loading executor " + componentId + ":" + executorId); return new ExecutorShutdown(this, Lists.newArrayList(handler), idToTask, receiveQueue); }
- 这里使用Utils.asyncLoop创建Utils.SmartThread并且调用start启动
Utils.asyncLoop
storm-2.0.0/storm-client/src/jvm/org/apache/storm/utils/Utils.java
/** * Creates a thread that calls the given code repeatedly, sleeping for an interval of seconds equal to the return value of the previous * call. * * The given afn may be a callable that returns the number of seconds to sleep, or it may be a Callable that returns another Callable * that in turn returns the number of seconds to sleep. In the latter case isFactory. * * @param afn the code to call on each iteration * @param isDaemon whether the new thread should be a daemon thread * @param eh code to call when afn throws an exception * @param priority the new thread's priority * @param isFactory whether afn returns a callable instead of sleep seconds * @param startImmediately whether to start the thread before returning * @param threadName a suffix to be appended to the thread name * @return the newly created thread * * @see Thread */ public static SmartThread asyncLoop(final Callable afn, boolean isDaemon, final Thread.UncaughtExceptionHandler eh, int priority, final boolean isFactory, boolean startImmediately, String threadName) { SmartThread thread = new SmartThread(new Runnable() { public void run() { try { final Callable<Long> fn = isFactory ? (Callable<Long>) afn.call() : afn; while (true) { if (Thread.interrupted()) { throw new InterruptedException(); } final Long s = fn.call(); if (s == null) { // then stop running it break; } if (s > 0) { Time.sleep(s); } } } catch (Throwable t) { if (Utils.exceptionCauseIsInstanceOf( InterruptedException.class, t)) { LOG.info("Async loop interrupted!"); return; } LOG.error("Async loop died!", t); throw new RuntimeException(t); } } }); if (eh != null) { thread.setUncaughtExceptionHandler(eh); } else { thread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { public void uncaughtException(Thread t, Throwable e) { LOG.error("Async loop died!", e); Utils.exitProcess(1, "Async loop died!"); } }); } thread.setDaemon(isDaemon); thread.setPriority(priority); if (threadName != null && !threadName.isEmpty()) { thread.setName(thread.getName() + "-" + threadName); } if (startImmediately) { thread.start(); } return thread; }
- 这里run方法无限循环调用fn.call(),也就是调用Executor.call().call()方法
- BoltExecutor.call主要是调用receiveQueue.consume方法
- SpoutExecutor.call除了调用receiveQueue.consume方法,还调用了spouts.get(j).nextTuple()
receiveQueue.consume
storm-2.0.0/storm-client/src/jvm/org/apache/storm/utils/JCQueue.java
/** * Non blocking. Returns immediately if Q is empty. Returns number of elements consumed from Q */ public int consume(JCQueue.Consumer consumer) { return consume(consumer, continueRunning); } /** * Non blocking. Returns immediately if Q is empty. Runs till Q is empty OR exitCond.keepRunning() return false. Returns number of * elements consumed from Q */ public int consume(JCQueue.Consumer consumer, ExitCondition exitCond) { try { return consumeImpl(consumer, exitCond); } catch (InterruptedException e) { throw new RuntimeException(e); } } /** * Non blocking. Returns immediately if Q is empty. Returns number of elements consumed from Q * * @param consumer * @param exitCond */ private int consumeImpl(Consumer consumer, ExitCondition exitCond) throws InterruptedException { int drainCount = 0; while (exitCond.keepRunning()) { Object tuple = recvQueue.poll(); if (tuple == null) { break; } consumer.accept(tuple); ++drainCount; } int overflowDrainCount = 0; int limit = overflowQ.size(); while (exitCond.keepRunning() && (overflowDrainCount < limit)) { // 2nd cond prevents staying stuck with consuming overflow Object tuple = overflowQ.poll(); ++overflowDrainCount; consumer.accept(tuple); } int total = drainCount + overflowDrainCount; if (total > 0) { consumer.flush(); } return total; }
- consume方法主要是调用consumer的accept方法
Task
storm-2.0.0/storm-client/src/jvm/org/apache/storm/daemon/Task.java
public class Task { private static final Logger LOG = LoggerFactory.getLogger(Task.class); private final TaskMetrics taskMetrics; private Executor executor; private WorkerState workerData; private TopologyContext systemTopologyContext; private TopologyContext userTopologyContext; private WorkerTopologyContext workerTopologyContext; private Integer taskId; private String componentId; private Object taskObject; // Spout/Bolt object private Map<String, Object> topoConf; private BooleanSupplier emitSampler; private CommonStats executorStats; private Map<String, Map<String, LoadAwareCustomStreamGrouping>> streamComponentToGrouper; private HashMap<String, ArrayList<LoadAwareCustomStreamGrouping>> streamToGroupers; private boolean debug; public Task(Executor executor, Integer taskId) throws IOException { this.taskId = taskId; this.executor = executor; this.workerData = executor.getWorkerData(); this.topoConf = executor.getTopoConf(); this.componentId = executor.getComponentId(); this.streamComponentToGrouper = executor.getStreamToComponentToGrouper(); this.streamToGroupers = getGroupersPerStream(streamComponentToGrouper); this.executorStats = executor.getStats(); this.workerTopologyContext = executor.getWorkerTopologyContext(); this.emitSampler = ConfigUtils.mkStatsSampler(topoConf); this.systemTopologyContext = mkTopologyContext(workerData.getSystemTopology()); this.userTopologyContext = mkTopologyContext(workerData.getTopology()); this.taskObject = mkTaskObject(); this.debug = topoConf.containsKey(Config.TOPOLOGY_DEBUG) && (Boolean) topoConf.get(Config.TOPOLOGY_DEBUG); this.addTaskHooks(); this.taskMetrics = new TaskMetrics(this.workerTopologyContext, this.componentId, this.taskId); } //...... }
Executor.accept
storm-2.0.0/storm-client/src/jvm/org/apache/storm/executor/Executor.java
@Override public void accept(Object event) { AddressedTuple addressedTuple = (AddressedTuple) event; int taskId = addressedTuple.getDest(); TupleImpl tuple = (TupleImpl) addressedTuple.getTuple(); if (isDebug) { LOG.info("Processing received message FOR {} TUPLE: {}", taskId, tuple); } try { if (taskId != AddressedTuple.BROADCAST_DEST) { tupleActionFn(taskId, tuple); } else { for (Integer t : taskIds) { tupleActionFn(t, tuple); } } } catch (Exception e) { throw new RuntimeException(e); } }
- accept方法主要是对每个taskId,挨个调用tupleActionFn方法
- BoltExecutor.tupleActionFn主要是从task获取boltObject,然后调用boltObject.execute(tuple);
- SpoutExecutor.tupleActionFn主要是从RotatingMap<Long, TupleInfo> pending取出TupleInfo,然后进行成功或失败的ack
ExecutorShutdown
storm-2.0.0/storm-client/src/jvm/org/apache/storm/executor/ExecutorShutdown.java
public class ExecutorShutdown implements Shutdownable, IRunningExecutor { private static final Logger LOG = LoggerFactory.getLogger(ExecutorShutdown.class); private final Executor executor; private final List<Utils.SmartThread> threads; private final ArrayList<Task> taskDatas; private final JCQueue receiveQueue; //...... @Override public void credentialsChanged(Credentials credentials) { TupleImpl tuple = new TupleImpl(executor.getWorkerTopologyContext(), new Values(credentials), Constants.SYSTEM_COMPONENT_ID, (int) Constants.SYSTEM_TASK_ID, Constants.CREDENTIALS_CHANGED_STREAM_ID); AddressedTuple addressedTuple = new AddressedTuple(AddressedTuple.BROADCAST_DEST, tuple); try { executor.getReceiveQueue().publish(addressedTuple); executor.getReceiveQueue().flush(); } catch (InterruptedException e) { throw new RuntimeException(e); } } public void loadChanged(LoadMapping loadMapping) { executor.reflectNewLoadMapping(loadMapping); } @Override public JCQueue getReceiveQueue() { return receiveQueue; } @Override public boolean publishFlushTuple() { return executor.publishFlushTuple(); } @Override public void shutdown() { try { LOG.info("Shutting down executor " + executor.getComponentId() + ":" + executor.getExecutorId()); executor.getReceiveQueue().close(); for (Utils.SmartThread t : threads) { t.interrupt(); } for (Utils.SmartThread t : threads) { LOG.debug("Executor " + executor.getComponentId() + ":" + executor.getExecutorId() + " joining thread " + t.getName()); t.join(); } executor.getStats().cleanupStats(); for (Task task : taskDatas) { if (task == null) { continue; } TopologyContext userContext = task.getUserContext(); for (ITaskHook hook : userContext.getHooks()) { hook.cleanup(); } } executor.getStormClusterState().disconnect(); if (executor.getOpenOrPrepareWasCalled().get()) { for (Task task : taskDatas) { if (task == null) { continue; } Object object = task.getTaskObject(); if (object instanceof ISpout) { ((ISpout) object).close(); } else if (object instanceof IBolt) { ((IBolt) object).cleanup(); } else { LOG.error("unknown component object"); } } } LOG.info("Shut down executor " + executor.getComponentId() + ":" + executor.getExecutorId()); } catch (Exception e) { throw Utils.wrapInRuntime(e); } } }
- ExecutorShutdown主要包装了一下shutdown的处理
小结
- worker启动之后从去zk的/assignments/{topology}路径,比如/assignments/DemoTopology-1-1539163962读取assignment信息
- 然后根据assignment信息获取Map<List<Long>, NodeInfo> executorToNodePort,然后通过Executor.mkExecutor创建Executor
- 创建Executor的时候根据assignment信息中的task信息创建Task绑定到Executor
- 之后调用executor的execute方法,这个方法启动Utils.SmartThread,该thread循环调用Executor.call().call()方法
BoltExecutor.call主要是调用receiveQueue.consume方法;SpoutExecutor.call除了调用receiveQueue.consume方法,还调用了spouts.get(j).nextTuple()
- receiveQueue.consume方法主要是调用Executor的accept方法,而accept方法主要是对每个taskId,挨个调用tupleActionFn方法
BoltExecutor.tupleActionFn主要是从task获取boltObject,然后调用boltObject.execute(tuple);SpoutExecutor.tupleActionFn主要是从RotatingMap<Long, TupleInfo> pending取出TupleInfo,然后进行成功或失败的ack
- worker可以理解为进程,executor即为该进程里头的线程数,而task则可以理解为spout或bolt的实例,默认是一个executor对应一个spout或bolt的task
- 增加worker或executor可以对supervisor进行扩容,这个过程称之为rebalance,而task则作为载体及任务的抽象从负载大的worker的executor转到新worker的executor上,实现rebalance(
rebalance命令只能重新调整worker、executor数量,无法改变task数量
)