前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >Netflix Conductor源码分析--Client层源码分析

Netflix Conductor源码分析--Client层源码分析

作者头像
小程故事多
发布2018-12-25 14:41:35
1.7K0
发布2018-12-25 14:41:35
举报
文章被收录于专栏:微服务生态微服务生态

一、Client层总体介绍

在正式介绍Client层源码前,我们先来看一下如何在client端与server端通信,demo代码如下:

代码语言:javascript
复制
TaskClient taskClient = new TaskClient();
taskClient.setRootURI("http://localhost:8080/api/");        //Point this to the server API
int threadCount = 2;            //number of threads used to execute workers.  To avoid starvation, should be same or more than number of workers
Worker worker1 = new OrderWorker("order");
Worker worker2 = new PaymentWorker("payment");
//Create WorkflowTaskCoordinator
WorkflowTaskCoordinator.Builder builder = new WorkflowTaskCoordinator.Builder();
WorkflowTaskCoordinator coordinator = builder.withWorkers(worker1, worker2).withThreadCount(threadCount).withTaskClient(taskClient).build();
//Start for polling and execution of the tasks
coordinator.init();

代码说明:

1、第一步需要创建TaskClient类并设置server端的API URL路径以便客户端能够与服务端通信。

2、创建任务工作者Worker对象,具体的任务是由Worker来执行。

3、将Worker对象传入WorkerflowTaskCoordinator对象中,WorkerflowTaskCoordinator负责启动线程池来执行Worker任务,同时维护与server端的心跳以及最新任务数据的拉取操作。

通过阅读上述代码引出了几个类名称的解释:

  • WorkerflowTaskCoordinator:工作流的协调者,负责管理Task Worker的线程池以及和服务端的通信。
  • TaskClient:conductor的任务管理客户端类,负责从server端轮询任务以及更新任务状态等。
  • Builder:用于创建WorkerflowTaskCoordinator实例的建造类。

这三个类的类图如图1-1所示,从图中可以看到类的依赖、组合等关系。

图1-1

图1-1展示是Client层最核心的三个类的依赖关系,我们接下来的源码解析就是围绕这三个类来展开。

整个Client模块的包结构和关键类如图1-2所示:

图1-2

其中:

  • config包是关于Client的一些配置类
  • exceptions包是自定义的client异常类
  • http包是与服务端通信的基础类,包括基础基类ClientBase,还有元数据、负载、客户端任务,工作流等通信类
  • task包主要包括工作流协调者和工作流任务统计类
  • worker包主要包括Worker工作者接口类

二、Client层源码执行的全流程解析

我们拿文章 深入浅出Netflix Conductor使用 中介绍的案例来讲解源码流程(文章中包括了任务、工作流的DSL定义以及如何使用),流程图形表示如图1-3所示:

图1-3

这张图的含义非常简单,用户走下单流程到order模块,如果下单成功则走payment支付模块进行支付,如果下单失败则走失败模块进行重试等操作。

在Swagger界面上输入如下参数启动工作流,如图1-4

图1-4

启动的过程实际上是通过Swagger API接口调用server端的相关类,而client端则是通过拉取的方式来得到需要自己执行任务的通知和输入参数。

启动完工作流之后Client端的代码进入WorkerflowTaskCoordinator中的init方法,代码如下所示:

代码语言:javascript
复制
public synchronized void init() {
    if(threadCount == -1) {
        threadCount = workers.size();
    }
    logger.info("Initialized the worker with {} threads", threadCount);
    this.workerQueue = new LinkedBlockingQueue<Runnable>(workerQueueSize);
    AtomicInteger count = new AtomicInteger(0);
    this.executorService = new ThreadPoolExecutor(threadCount, threadCount,
            0L, TimeUnit.MILLISECONDS,
            workerQueue,
            (runnable) -> {
                Thread thread = new Thread(runnable);
                thread.setName(workerNamePrefix + count.getAndIncrement());
                return thread;
            });
    this.scheduledExecutorService = Executors.newScheduledThreadPool(workers.size());
  
   //定时轮询server状态策略,默认每隔1秒进行轮询,根据任务名获取当前任务信息
    workers.forEach(worker -> {
        scheduledExecutorService.scheduleWithFixedDelay(()->pollForTask(worker), worker.getPollingInterval(), worker.getPollingInterval(), TimeUnit.MILLISECONDS);
    });
}

代码说明:

这段代码通过JDK中的scheduledExecutorService.scheduleWithFixedDelay方法每隔一秒对server端进行轮询,轮询任务的方法是pollForTask,代码如下:

代码语言:javascript
复制
private void pollForTask(Worker worker) {
    if(eurekaClient != null && !eurekaClient.getInstanceRemoteStatus().equals(InstanceStatus.UP)) {
        logger.debug("Instance is NOT UP in discovery - will not poll");
        return;
    }
    if(worker.paused()) {
        WorkflowTaskMetrics.incrementTaskPausedCount(worker.getTaskDefName());
        logger.debug("Worker {} has been paused. Not polling anymore!", worker.getClass());
        return;
    }
    String domain = Optional.ofNullable(PropertyFactory.getString(worker.getTaskDefName(), DOMAIN, null))
            .orElse(PropertyFactory.getString(ALL_WORKERS, DOMAIN, null));
    logger.debug("Polling {}, domain={}, count = {} timeout = {} ms", worker.getTaskDefName(), domain, worker.getPollCount(), worker.getLongPollTimeoutInMS());
    List<Task> tasks = Collections.emptyList();
    try{
        // get the remaining capacity of worker queue to prevent queue full exception
        int realPollCount = Math.min(workerQueue.remainingCapacity(), worker.getPollCount());
        if (realPollCount <= 0) {
            logger.warn("All workers are busy, not polling. queue size = {}, max = {}", workerQueue.size(), workerQueueSize);
            return;
        }
        //获取当前客户端的任务名称
        String taskType = worker.getTaskDefName();
        //根据当前客户端的任务名称从server端的状态机获取是否有自己要执行的任务,如果有任务则获取执行,只能获取一次。
        tasks = getPollTimer(taskType)
                .record(() -> taskClient.batchPollTasksInDomain(taskType, domain, worker.getIdentity(), realPollCount, worker.getLongPollTimeoutInMS()));
        incrementTaskPollCount(taskType, tasks.size());
        logger.debug("Polled {}, domain {}, received {} tasks in worker - {}", worker.getTaskDefName(), domain, tasks.size(), worker.getIdentity());
    } catch (Exception e) {
        WorkflowTaskMetrics.incrementTaskPollErrorCount(worker.getTaskDefName(), e);
        logger.error("Error when polling for tasks", e);
    }
    //根据获取的任务列表,以线程的方式启动执行任务
    for (Task task : tasks) {
        try {
            executorService.submit(() -> {
                try {
                    logger.debug("Executing task {}, taskId - {} in worker - {}", task.getTaskDefName(), task.getTaskId(), worker.getIdentity());
                    //这步就是执行用户自定义的任务逻辑
                    execute(worker, task);
                } catch (Throwable t) {
                    //执行失败,置任务状态为失败,并将失败结果返回到server端
                    task.setStatus(Task.Status.FAILED);
                    TaskResult result = new TaskResult(task);
                    handleException(t, result, worker, task);
                }
            });
        } catch (RejectedExecutionException e) {
            WorkflowTaskMetrics.incrementTaskExecutionQueueFullCount(worker.getTaskDefName());
            logger.error("Execution queue is full, returning task: {}", task.getTaskId(), e);
            returnTask(worker, task);
        }
    }
}

代码说明:

每隔一秒从服务端的(tasks/poll/batch/{taskType})获取当前需要执行的任务列表,任务只能获取一次不能重新获取。然后将任务通过异步线程的方式启动执行,每一个任务都是由用户自定义的逻辑实现,任务的返回值被封装到了TaskResult类中,execute方法的内容如下所示:

代码语言:javascript
复制
private void execute(Worker worker, Task task) {
    String taskType = task.getTaskDefName();
    try {
        if(!worker.preAck(task)) {
            logger.debug("Worker decided not to ack the task {}, taskId = {}", taskType, task.getTaskId());
            return;
        }
        if (!taskClient.ack(task.getTaskId(), worker.getIdentity())) {
            WorkflowTaskMetrics.incrementTaskAckFailedCount(worker.getTaskDefName());
            logger.error("Ack failed for {}, taskId = {}", taskType, task.getTaskId());
            returnTask(worker, task);
            return;
        }
    } catch (Exception e) {
        logger.error(String.format("ack exception for task %s, taskId = %s in worker - %s", task.getTaskDefName(), task.getTaskId(), worker.getIdentity()), e);
        WorkflowTaskMetrics.incrementTaskAckErrorCount(worker.getTaskDefName(), e);
        returnTask(worker, task);
        return;
    }
    com.google.common.base.Stopwatch stopwatch = com.google.common.base.Stopwatch.createStarted();
    TaskResult result = null;
    try {
        //前面大部分都是做监控和统计功能的,在这里不细说
        //这段代码是真正执行用户Task任务的代码,执行完后返回值被封装为TaskResult对象
        logger.debug("Executing task {} in worker {} at {}", task, worker.getClass().getSimpleName(), worker.getIdentity());
        result = worker.execute(task);
        result.setWorkflowInstanceId(task.getWorkflowInstanceId());
        result.setTaskId(task.getTaskId());
        result.setWorkerId(worker.getIdentity());
    } catch (Exception e) {
        logger.error("Unable to execute task {}", task, e);
        if (result == null) {
            task.setStatus(Task.Status.FAILED);
            result = new TaskResult(task);
        }
        handleException(e, result, worker, task);
    } finally {
        stopwatch.stop();
        WorkflowTaskMetrics.getExecutionTimer(worker.getTaskDefName())
                .record(stopwatch.elapsed(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS);
    }
    logger.debug("Task {} executed by worker {} at {} with status {}", task.getTaskId(), worker.getClass().getSimpleName(), worker.getIdentity(), task.getStatus());
    //更新任务状态,成功或者失败
    updateWithRetry(updateRetryCount, task, result, worker);
}

代码说明:

通过worker.execute方法执行用户定义的任务逻辑,不管是否成功都执行updatewithRetry方法更新server端的任务状态和任务执行返回结果。

访问的URL是/api/tasks。

三、完整流程时序图

本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
原始发表:2018.12.10 ,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 二、Client层源码执行的全流程解析
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档