聊聊jesque的WorkerImpl与WorkerPool

本文主要讲一下jesque的WorkerImpl与WorkerPool。

resque

Resque是一个使用redis来创建后台任务的ruby组件。而jesque是其java版本。通常用来做延时队列。

WorkerImpl

        List<String> queues = Arrays.asList(delayedQueue);
        final Worker worker = new WorkerImpl(jesqueConfig,queues, new MapBasedJobFactory(map(entry("DemoJob", DemoJob.class))));
        final Thread workerThread = new Thread(worker);
        workerThread.start();

这是worker实例 jesque-2.1.2-sources.jar!/net/greghaines/jesque/worker/WorkerImpl.java

/**
     * Starts this worker. Registers the worker in Redis and begins polling the queues for jobs.<br>
     * Stop this worker by calling end() on any thread.
     */
    @Override
    public void run() {
        if (this.state.compareAndSet(NEW, RUNNING)) {
            try {
                renameThread("RUNNING");
                this.threadRef.set(Thread.currentThread());
                this.jedis.sadd(key(WORKERS), this.name);
                this.jedis.set(key(WORKER, this.name, STARTED), new SimpleDateFormat(DATE_FORMAT).format(new Date()));
                this.listenerDelegate.fireEvent(WORKER_START, this, null, null, null, null, null);
                this.popScriptHash.set(this.jedis.scriptLoad(ScriptUtils.readScript(POP_LUA)));
                this.lpoplpushScriptHash.set(this.jedis.scriptLoad(ScriptUtils.readScript(LPOPLPUSH_LUA)));
                this.multiPriorityQueuesScriptHash
                        .set(this.jedis.scriptLoad(ScriptUtils.readScript(POP_FROM_MULTIPLE_PRIO_QUEUES)));
                poll();
            } catch (Exception ex) {
                LOG.error("Uncaught exception in worker run-loop!", ex);
                this.listenerDelegate.fireEvent(WORKER_ERROR, this, null, null, null, null, ex);
            } finally {
                renameThread("STOPPING");
                this.listenerDelegate.fireEvent(WORKER_STOP, this, null, null, null, null, null);
                this.jedis.srem(key(WORKERS), this.name);
                this.jedis.del(key(WORKER, this.name), key(WORKER, this.name, STARTED), key(STAT, FAILED, this.name),
                        key(STAT, PROCESSED, this.name));
                this.jedis.quit();
                this.threadRef.set(null);
            }
        } else if (RUNNING.equals(this.state.get())) {
            throw new IllegalStateException("This WorkerImpl is already running");
        } else {
            throw new IllegalStateException("This WorkerImpl is shutdown");
        }
    }

实现了runnable方法,里头poll方法无限循环

   /**
     * Polls the queues for jobs and executes them.
     */
    protected void poll() {
        int missCount = 0;
        String curQueue = null;
        while (RUNNING.equals(this.state.get())) {
            try {
                if (threadNameChangingEnabled) {
                    renameThread("Waiting for " + JesqueUtils.join(",", this.queueNames));
                }
                curQueue = getNextQueue();
                if (curQueue != null) {
                    checkPaused();
                    // Might have been waiting in poll()/checkPaused() for a while
                    if (RUNNING.equals(this.state.get())) {
                        this.listenerDelegate.fireEvent(WORKER_POLL, this, curQueue, null, null, null, null);
                        final String payload = pop(curQueue);
                        if (payload != null) {
                            process(ObjectMapperFactory.get().readValue(payload, Job.class), curQueue);
                            missCount = 0;
                        } else {
                            missCount++;
                            if (shouldSleep(missCount) && RUNNING.equals(this.state.get())) {
                                // Keeps worker from busy-spinning on empty queues
                                missCount = 0;
                                Thread.sleep(EMPTY_QUEUE_SLEEP_TIME);
                            }
                        }
                    }
                }
            } catch (InterruptedException ie) {
                if (!isShutdown()) {
                    recoverFromException(curQueue, ie);
                }
            } catch (JsonParseException | JsonMappingException e) {
                // If the job JSON is not deserializable, we never want to submit it again...
                removeInFlight(curQueue);
                recoverFromException(curQueue, e);
            } catch (Exception e) {
                recoverFromException(curQueue, e);
            }
        }
    }

不断地pop和process

/**
     * Materializes and executes the given job.
     * 
     * @param job the Job to process
     * @param curQueue the queue the payload came from
     */
    protected void process(final Job job, final String curQueue) {
        try {
            this.processingJob.set(true);
            if (threadNameChangingEnabled) {
                renameThread("Processing " + curQueue + " since " + System.currentTimeMillis());
            }
            this.listenerDelegate.fireEvent(JOB_PROCESS, this, curQueue, job, null, null, null);
            this.jedis.set(key(WORKER, this.name), statusMsg(curQueue, job));
            final Object instance = this.jobFactory.materializeJob(job);
            final Object result = execute(job, curQueue, instance);
            success(job, instance, result, curQueue);
        } catch (Throwable thrwbl) {
            failure(thrwbl, job, curQueue);
        } finally {
            removeInFlight(curQueue);
            this.jedis.del(key(WORKER, this.name));
            this.processingJob.set(false);
        }
    }

而process这个方法,就是实例化目标job,然后execute

/**
     * Executes the given job.
     * 
     * @param job the job to execute
     * @param curQueue the queue the job came from
     * @param instance the materialized job
     * @throws Exception if the instance is a {@link Callable} and throws an exception
     * @return result of the execution
     */
    protected Object execute(final Job job, final String curQueue, final Object instance) throws Exception {
        if (instance instanceof WorkerAware) {
            ((WorkerAware) instance).setWorker(this);
        }
        this.listenerDelegate.fireEvent(JOB_EXECUTE, this, curQueue, job, instance, null, null);
        final Object result;
        if (instance instanceof Callable) {
            result = ((Callable<?>) instance).call(); // The job is executing!
        } else if (instance instanceof Runnable) {
            ((Runnable) instance).run(); // The job is executing!
            result = null;
        } else { // Should never happen since we're testing the class earlier
            throw new ClassCastException(
                    "Instance must be a Runnable or a Callable: " + instance.getClass().getName() + " - " + instance);
        }
        return result;
    }

而execute就是调用call或者run方法。 从这里可以看出是单线程阻塞的,如果一个job比较耗时,是会影响其他job的触发和执行。

WorkerPool

jesque-2.1.2-sources.jar!/net/greghaines/jesque/worker/WorkerPool.java

/**
     * Create a WorkerPool with the given number of Workers and the given <code>ThreadFactory</code>.
     * @param workerFactory a Callable that returns an implementation of Worker
     * @param numWorkers the number of Workers to create
     * @param threadFactory the factory to create pre-configured Threads
     */
    public WorkerPool(final Callable<? extends Worker> workerFactory, final int numWorkers,
            final ThreadFactory threadFactory) {
        this.workers = new ArrayList<>(numWorkers);
        this.threads = new ArrayList<>(numWorkers);
        this.eventEmitter = new WorkerPoolEventEmitter(this.workers);
        for (int i = 0; i < numWorkers; i++) {
            try {
                final Worker worker = workerFactory.call();
                this.workers.add(worker);
                this.threads.add(threadFactory.newThread(worker));
            } catch (RuntimeException re) {
                throw re;
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    }
/**
     * {@inheritDoc}
     */
    @Override
    public void run() {
        for (final Thread thread : this.threads) {
            thread.start();
        }
        Thread.yield();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void end(final boolean now) {
        for (final Worker worker : this.workers) {
            worker.end(now);
        }
    }

workerpool维护了一组worker实例,起线程池的作用,尽可能提高job的并发度。

doc

  • resque
  • jesque

原文发布于微信公众号 - 码匠的流水账(geek_luandun)

原文发表时间:2017-11-15

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏JAVA技术站

基于MyBatis注解扩展,实现无需配置就可拥有增删改 顶

      如果项目本身基于SpringMVC+Mybatis构建,不需添加任何配置Mapper类只需继承

5433
来自专栏菩提树下的杨过

Flash/Flex学习笔记(34):AS3中的自定义事件

类似C#中自定义事件需要一个自定义的EventArgs子类一样,AS3也需要开发者自定义一个Event类的子类,这里我们假设一种场景:设计一个Person(人物...

2037
来自专栏大内老A

ASP.NET MVC的Model元数据与Model模板:将”ListControl”引入ASP.NET MVC

我们不仅可以创建相应的模板来根据Model元数据控制种类型的数据在UI界面上的呈现方法,还可以通过一些扩展来控制Model元数据本身。在某些情况下通过这两者的结...

3676
来自专栏函数式编程语言及工具

Akka(8): 分布式运算:Remoting-远程查找式

  Akka是一种消息驱动运算模式,它实现跨JVM程序运算的方式是通过能跨JVM的消息系统来调动分布在不同JVM上ActorSystem中的Actor进行运算,...

4259
来自专栏木宛城主

PowerShell 获取Site Collection下被签出的文件

由于权限的设置,当文件被签出时导致别人不可见了,这对校验文件个数的人来说着实是件烦恼的事。幸好利用PowerShell,可以获取Site Collection下...

2017
来自专栏码匠的流水账

聊聊pg jdbc的queryTimeout及next方法

本文主要介绍一下pg jdbc statement的queryTimeout及resultSet的next方法

3371
来自专栏分布式系统进阶

Influxdb中Select查询请求结果涉及到的一些数据结构

相当于c里面的链表元素,itr指向下一个元素的指针,buf表示当前元素,即FloatPoint类型的链表的迭代器.

1572
来自专栏王磊的博客

javascript数字格式化通用类——accounting.js使用

简介 accounting.js 是一个非常小的JavaScript方法库用于对数字,金额和货币进行格式化。并提供可选的Excel风格列渲染。它没有依赖任何JS...

6036
来自专栏码匠的流水账

聊聊spring for kafka对producer的封装与集成

本文主要解析一下spring for apache kafka对原生的kafka client producer的封装与集成。

1482
来自专栏Java 技术分享

Ajax 案例之三级联动

3256

扫码关注云+社区

领取腾讯云代金券