C++11 —— 使用 thread 实现线程池

一个逗逗 2020-01-01

1. 引言

在新的 C++11 标准中，引入并发编程的一些基础组件：线程（thread）、互斥锁（mutex）、条件变量（condition_variable） 等，凭借这些，就足够我设计一个平台无关的 线程池 组件了。下面就详细介绍一下这个线程池组件。

2. 结构设计图

C++11 —— 使用 thread 实现线程池

需要特别说明的是，这个线程池组件，在增加了“存在关联性的任务对象顺序执行”的功能后，原本的任务队列就分成了两级任务队列，目的是为了降低 “任务提交” 与 “任务提取” 之间（属于一种生产/消费的关系）的锁竞争。

3. 源码说明

源码有点多，这里就不贴出来了，直接给下载地址：https://github.com/Gaaagaa/xthreadpool 。主要的线程池类 x_threadpool_t 在 xthreadpool.h 中已完整实现，在实际项目应用中，只需要 xthreadpool.h 这一个文件就足够了。

测试程序的编译命令：

  1. MSVC++2017：cl /EHsc main.cpp
  2. gcc ：g++ -Wall -std=c++11 -lpthread -o main main.cpp

技术特点：

  1. 使用 C++11 的 thread 实现，可跨平台，亲测的编译器有 MSVC++2017、gcc 4.8.5、gcc 8.2.0；
  2. 支持传统的面向对象编程的任务对象类接口：继承抽象任务对象接口类，实现多态（可结合对象池的模式进行资源复用）；
  3. 支持泛型接口的任务对象，如：C 函数接口、lambda 表达式、仿函数对象、类对象的成员函数调用；
  4. 支持动态调整工作线程的数量：通过 resize() 接口实现，该功能当前的实现方式还不够好，仍有优化的方式；
  5. 支持运行时的线程状态检测（判断当前工作线程是否需要退出）；
  6. 存在关联性的任务对象可顺序执行（这一特点只针对一些特别的应用场景，后续示例中会展示）。

4. 使用说明与示例代码

4.1 启动与关闭

#include "xthreadpool.h"
#include <chrono>
#include <stdio.h>

int main(int argc, char * argv[])
{
    // 工作线程数量
    // 若为 0，将取 hardware_concurrency() 返回值的 2倍 + 1
    int nthreads = 4;

    // 线程池对象
    x_threadpool_t xht_pool;

    // 启动线程池
    if (!xht_pool.startup(nthreads))
    {
        printf("startup return false!\n");
        return -1;
    }

    //======================================
    // 提交任务对象
    // ......

    //======================================

    // 等待所有任务执行完成
    while (xht_pool.task_count() > 0)
        std::this_thread::sleep_for(std::chrono::milliseconds(1));

    // 关闭线程池
    xht_pool.shutdown();

    return 0;
}

4.2 提交“传统 C 回调函数”的任务对象

#include "xthreadpool.h"
#include <chrono>
#include <stdio.h>

void func_task(int task_id, int task_iter)
{
    int count = 1;
    do
    {
        printf("func_task[%d, %d] => count: %d\n", task_id, task_iter, count);
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    } while (count++ < 10);
}

int main(int argc, char * argv[])
{
    // 线程池对象
    x_threadpool_t xht_pool;

    // 启动线程池
    if (!xht_pool.startup(0))
    {
        printf("startup return false!\n");
        return -1;
    }

    //======================================
    // 提交任务对象 : C 函数接口的任务

    for (int iter = 0; iter < 100; iter += 10)
    {
        xht_pool.submit_task_ex(func_task, iter, iter * iter);
    }

    //======================================

    // 等待所有任务执行完成
    while (xht_pool.task_count() > 0)
        std::this_thread::sleep_for(std::chrono::milliseconds(1));

    // 关闭线程池
    xht_pool.shutdown();

    return 0;
}

4.3 提交“仿函数对象”的任务对象

#include "xthreadpool.h"
#include <chrono>
#include <stdio.h>

/**
 * @struct functor_task_A
 * @brief  仿函数模式的任务对象类。
 */
struct functor_task_A
{
    // constructor/destructor
public:
    functor_task_A(int xtask_id = 0) : m_xtask_id(xtask_id) { }

public:
    void operator()() const
    {
        int count = 1;
        do
        {
            printf("functor_task_A[%d] => count: %d\n", m_xtask_id, count);
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        } while (count++ < 10);
    }

    // data members
private:
    int m_xtask_id;
};

/**
 * @struct functor_task_B
 * @brief  仿函数模式的任务对象类。
 */
struct functor_task_B
{
    // constructor/destructor
public:
    functor_task_B(int xtask_id = 0) : m_xtask_id(xtask_id) { }

public:
    void operator()(int flag) const
    {
        int count = 1;
        do
        {
            printf("functor_task_B[%d, %d] => count: %d\n", m_xtask_id, flag, count);
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        } while (count++ < 10);
    }

    // data members
private:
    int m_xtask_id;
};

int main(int argc, char * argv[])
{
    // 线程池对象
    x_threadpool_t xht_pool;

    // 启动线程池
    if (!xht_pool.startup(0))
    {
        printf("startup return false!\n");
        return -1;
    }

    //======================================
    // 提交任务对象 : 仿函数对象

    for (int iter = 0; iter < 100; iter += 10)
    {
        xht_pool.submit_task_ex((functor_task_A(iter)));
    }

    for (int iter = 0; iter < 100; iter += 10)
    {
        xht_pool.submit_task_ex((functor_task_B(iter)), iter * iter / 2);
    }

    //======================================

    // 等待所有任务执行完成
    while (xht_pool.task_count() > 0)
        std::this_thread::sleep_for(std::chrono::milliseconds(1));

    // 关闭线程池
    xht_pool.shutdown();

    return 0;
}

4.4 提交“调用类对象的成员函数”的任务对象

#include "xthreadpool.h"
#include <chrono>
#include <stdio.h>

/**
 * @class memfunc_task
 * @brief 调用成员函数的任务对象类。
 */
class memfunc_task
{
    // constructor/destructor
public:
    memfunc_task(int xtask_id = 0) : m_xtask_id(xtask_id) {  }

    // overrides
public:
    /**********************************************************/
    /**
     * @brief 任务对象执行流程的操作接口。
     */
    void memfunc(int task_iter)
    {
        int count = 1;
        do
        {
            printf("memfunc_task[%d, %d] => count: %d\n", m_xtask_id, task_iter, count);
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        } while (count++ < 10);
    }

    // data members
private:
    int m_xtask_id;
};

int main(int argc, char * argv[])
{
    // 线程池对象
    x_threadpool_t xht_pool;

    // 启动线程池
    if (!xht_pool.startup(0))
    {
        printf("startup return false!\n");
        return -1;
    }

    //======================================
    // 提交任务对象

    // 注意，这个栈区对象的生命期需要在线程池关闭前存活，
    // 至少要保证所提交的任务都执行完成时，才可结束该对象的生命期
    memfunc_task mftask(0);

    for (int iter = 0; iter < 100; iter += 10)
    {
        xht_pool.submit_task_ex(&memfunc_task::memfunc, &mftask, iter);
    }

    //======================================

    // 等待所有任务执行完成
    while (xht_pool.task_count() > 0)
        std::this_thread::sleep_for(std::chrono::milliseconds(1));

    // 关闭线程池
    xht_pool.shutdown();

    return 0;
}

4.5 提交“重载后的任务对象类”的任务对象

#include "xthreadpool.h"
#include <chrono>
#include <stdio.h>

/**
 * @class user_task
 * @brief 用户自定义的任务对象类。
 */
class user_task : public x_task_t
{
    // constructor/destructor
public:
    user_task(int xtask_id = 0) : m_xtask_id(xtask_id) {  }

    // overrides
public:
    /**********************************************************/
    /**
     * @brief 任务对象执行流程的操作接口。
     */
    virtual void run(x_running_checker_t * xchecker_ptr) override
    {
        int count = 1;
        do
        {
            printf("[%d]user_task[%d] => count: %d\n", (int)xchecker_ptr->thread_index(), m_xtask_id, count);
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        } while ((count++ < 10) && xchecker_ptr->is_enable_running());
    }

    // data members
private:
    int m_xtask_id;
};

int main(int argc, char * argv[])
{
    // 线程池对象
    x_threadpool_t xht_pool;

    // 启动线程池
    if (!xht_pool.startup(0))
    {
        printf("startup return false!\n");
        return -1;
    }

    //======================================
    // 提交任务对象 : 重载的任务对象

    for (int iter = 0; iter < 100; iter += 10)
    {
        xht_pool.submit_task((x_task_ptr_t)(new user_task(iter)));
    }

    //======================================

    // 等待所有任务执行完成
    while (xht_pool.task_count() > 0)
        std::this_thread::sleep_for(std::chrono::milliseconds(1));

    // 关闭线程池
    xht_pool.shutdown();

    return 0;
}

4.6 提交“lambda表达式”的任务对象

#include "xthreadpool.h"
#include <chrono>
#include <stdio.h>

int main(int argc, char * argv[])
{
    // 线程池对象
    x_threadpool_t xht_pool;

    // 启动线程池
    if (!xht_pool.startup(0))
    {
        printf("startup return false!\n");
        return -1;
    }

    //======================================
    // 提交任务对象 : lambda 表达式

    for (int iter = 0; iter < 100; iter += 10)
    {
        xht_pool.submit_task_ex(
            [iter]() -> void
            {
                int task_id = iter;
                for (int jter = 0; jter < 10; ++jter)
                {
                    printf("lambda A task id : %d -> %d\n", task_id, task_id + jter);
                    std::this_thread::sleep_for(std::chrono::milliseconds(100));
                }
            });
    }

    for (int iter = 0; iter < 100; iter += 10)
    {
        xht_pool.submit_task_ex(
            [iter](const std::string & str_name) -> void
            {
                int task_id = iter;
                for (int jter = 0; jter < 10; ++jter)
                {
                    printf("lambda B task id : %d -> %d name : %s\n", task_id, task_id + jter, str_name.c_str());
                    std::this_thread::sleep_for(std::chrono::milliseconds(100));
                }
            }, std::string("Lambda B"));
    }

    //======================================

    // 等待所有任务执行完成
    while (xht_pool.task_count() > 0)
        std::this_thread::sleep_for(std::chrono::milliseconds(1));

    // 关闭线程池
    xht_pool.shutdown();

    return 0;
}

4.7 任务对象执行过程中检测工作线程是否可继续运行

x_threadpool_t 内部提供了一个类 x_running_checker_t 可检测当前工作线程是否可继续执行下去（或者说，是否要立即终止执行的任务流程）。提供此功能，主要目的是针对那些耗时长的任务对象（处于运行状态），在线程池关闭或者需要动态调整（减少）工作线程数量时，能够优雅地终止任务流程。参看如下代码是如何使用的：

#include "xthreadpool.h"
#include <chrono>
#include <stdio.h>

int main(int argc, char * argv[])
{
    // 线程池对象
    x_threadpool_t xht_pool;

    // 启动线程池
    if (!xht_pool.startup(0))
    {
        printf("startup return false!\n");
        return -1;
    }

    //======================================
    // 提交任务对象 : lambda 表达式（检测工作线程的退出标识）

    for (int iter = 0; iter < 100; iter += 10)
    {
        xht_pool.submit_task_ex(
            [iter](x_running_checker_t * xchecker_ptr) -> void
            {
                int task_id = iter;
                for (int jter = 0; (jter < 10) && xchecker_ptr->is_enable_running(); ++jter)
                {
                    printf("rchecker[%d] lambda A task id : %d -> %d\n", (int)xchecker_ptr->thread_index(), task_id, task_id + jter);
                    std::this_thread::sleep_for(std::chrono::milliseconds(100));
                }
            },
            x_running_checker_t::xholder());
    }

    //======================================

    // 等待所有任务执行完成
    while (xht_pool.task_count() > 0)
        std::this_thread::sleep_for(std::chrono::milliseconds(1));

    // 关闭线程池
    xht_pool.shutdown();

    return 0;
}

4.8 使某一批任务对象在线程池中可按提交次序而顺序执行

单看 “按提交次序而顺序执行” 这句话，可能不好理解，先看后面提到的这样应用场景。

A 对象产生（需要执行）的任务对象序列为[ A1, A2, ..., Am ]，并按此顺序依次提交到线程池中；与此同时，B 对象产生（需要执行）的任务对象序列为[ B1, B2, ..., Bn ]，也按此顺序依次提交到线程池中。而不管是 A 对象还是 B 对象，都期望按所提交顺序执行任务对象。这种情况下，在线程池中的任务队列可能有如下图所示的排列：

C++11 —— 使用 thread 实现线程池

从本质上看，我们只需要保证 A（或 B） 对象产生的任务在同一时刻只能有一个工作线程执行，这就能保证其顺序性。
面对这种应用场景，我在设计任务对象的抽象基类 x_task_t 时，就为此做好了相关的扩展接口。先看下 x_task_t 代码中的两个接口：

struct x_task_t
{
    ......

    // extensible interfaces
public:
    ......

    /**********************************************************/
    /**
     * @brief 判断 任务对象 是否挂起。
     * @note  若任务对象处于挂起状态，工作线程提取任务时，则跳过该对象。
     */
    virtual bool is_suspend(void) const { return false; }

    /**********************************************************/
    /**
     * @brief 设置任务对象的运行标识。
     * 
     * @note
     * <pre>
     *   工作线程在提取到任务对象后，则立即调用 set_running_flag(true) 操作；
     *   执行 run() 操作返回后，又调用 set_running_flag(false) 操作。
     * </pre>
     */
    virtual void set_running_flag(bool xrunning_flag) { }

    ......
};

再看下 x_threadpool_t 工作线程 “提取任务对象” 以及 “执行任务对象” 的部分实现流程（注意任务对象 is_suspend() 和 set_running_flag() 两个接口被调用的地方）：

class x_threadpool_t
{
    ......

    x_task_ptr_t get_task(void)
    {
        ......

        for (std::list< x_task_ptr_t >::iterator itlst = m_lst_run_tasks.begin();
             (itlst != m_lst_run_tasks.end()) && is_enable_get_task();
             ++itlst)
        {
            if ((nullptr == *itlst) || !(*itlst)->is_suspend())
            {
                xtask_ptr = *itlst;
                m_lst_run_tasks.erase(itlst);
                m_xst_lst_tasks.fetch_sub(1);
                break;
            }
        }

        if (nullptr != xtask_ptr)
        {
            xtask_ptr->set_running_flag(true);
        }

        return xtask_ptr;
    }

    ......

    void thread_run(size_t xthread_index)
    {
        ......

            xtask_ptr = get_task();
            if (nullptr == xtask_ptr)
            {
                if (get_lst_task_size() > 0)
                    thread_yield(xcounter);
                continue;
            }

            if (xht_checker.is_enable_running())
            {
                xtask_ptr->run(&xht_checker);
            }

            // 执行完任务对象后，将任务对象转换为 非挂起状态，
            // 加锁进行操作，是为了与 get_task() 内的操作保持队列的同步
            {
                // 标识当前不可提取待执行的任务对象，迫使 get_task() 内部迅速解锁
                m_xst_get_task.fetch_add(1);

                m_lock_run_task.lock();
                xtask_ptr->set_running_flag(false);
                m_lock_run_task.unlock();

                m_xst_get_task.fetch_sub(1);
            }

        ......
    }

    ......
};

我们可通过重载 x_task_t 所提供的两个扩展接口：is_suspend() 和 set_running_flag() 进行状态切换（即在调用 set_running_flag() 时让相关任务切换 is_suspend() 挂起/非挂起 两种状态），就可实现顺序执行。

示例代码如下所示：

#include "xthreadpool.h"
#include <chrono>
#include <string>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>

////////////////////////////////////////////////////////////////////////////////

class x_order_task_t;

/**
 * @class objectT
 * @brief 产生 x_order_task_t 任务对象序列的测试类。
 */
class objectT
{
    // constructor/destructor
public:
    objectT(const std::string & xstr_name)
        : m_xrunning_flag(false)
        , m_seqno_taskid(0)
        , m_xstr_name(xstr_name)
    {

    }

    // public interfaces
public:
    inline void set_running(bool xrunning_flag)
    {
        m_xrunning_flag = xrunning_flag;
    }

    inline bool is_running(void) const
    {
        return m_xrunning_flag;
    }

    inline const std::string & name(void) const
    {
        return m_xstr_name;
    }

    x_order_task_t * new_task(void);

    // data members
private:
    bool         m_xrunning_flag;
    int          m_seqno_taskid;
    std::string  m_xstr_name;
};

/**
 * @class x_order_task_t
 * @brief 保证顺序执行的测试任务对象。
 */
class x_order_task_t : public x_threadpool_t::x_task_t
{
    // constructor/destructor
public:
    x_order_task_t(objectT * obj_owner, int taskid)
        : m_obj_owner(obj_owner)
        , m_taskid(taskid)
    {

    }

    virtual ~x_order_task_t(void) { }

    // overrides
public:
    virtual void run(x_running_checker_t * xchecker_ptr) override
    {
        printf("%s:%02d\n", m_obj_owner->name().c_str(), m_taskid);
    }

    virtual bool is_suspend(void) const override
    {
        return m_obj_owner->is_running();
    }

    virtual void set_running_flag(bool xrunning_flag) override
    {
        m_obj_owner->set_running(xrunning_flag);
    }

    // data members
private:
    objectT * m_obj_owner;
    int       m_taskid;
};

x_order_task_t * objectT::new_task(void)
{
    return (new x_order_task_t(this, m_seqno_taskid++));
}

//====================================================================

int main(int argc, char * argv[])
{
    // 线程池对象
    x_threadpool_t xht_pool;

    // 启动线程池
    if (!xht_pool.startup(0, true))
    {
        printf("startup return false!\n");
        return -1;
    }

    //======================================
    // 便于测试而增加的代码
    // 提交任务对象 : 使线程池的所有工作线程处于暂停状态

    bool is_pause_pool = true;
    for (int iter = 0, nthds = (int)xht_pool.size(); iter < nthds + 1; ++iter)
    {
        xht_pool.submit_task_ex(
            [&is_pause_pool]() -> void
            {
                while (is_pause_pool)
                {
                    std::this_thread::sleep_for(std::chrono::milliseconds(1));
                }
            });
    }

    //======================================
    // 提交任务对象 : 按提交顺序执行任务对象

    srand(time(NULL));

    objectT objA("objA");
    objectT objB("objB");
    for (int iter = 0; iter < 100; ++iter)
    {
        if (0 == (rand() % 2))
        {
            xht_pool.submit_task(objA.new_task());
        }
        else
        {
            xht_pool.submit_task(objB.new_task());
        }
    }

    // 使线程池开始工作（便于测试而增加的代码）
    is_pause_pool = false;

    //======================================

    // 等待所有任务执行完成
    while (xht_pool.task_count() > 0)
        std::this_thread::sleep_for(std::chrono::milliseconds(1));

    // 关闭线程池
    xht_pool.shutdown();

    return 0;
}

看一下测试的输出结果：

objA:00
objB:00
objB:01
objB:02
objB:03
objB:04
objA:01
objA:02
objA:03
objB:05
objA:04
objB:06
objB:07
objB:08
objB:09
objA:05
objA:06
objB:10
objB:11
objB:12
objA:07
objA:08
objA:09
objA:10
objB:13
objA:11
objA:12
objB:14
objB:15
objB:16
objB:17
objB:18
objB:19
objA:13
objA:14
objA:15
objA:16
objA:17
objA:18
objA:19
objB:20
objA:20
objA:21
objA:22
objB:21
objB:22
objB:23
objA:23
objB:24
objA:24
objB:25
objB:26
objB:27
objB:28
objB:29
objA:25
objB:30
objA:26
objA:27
objB:31
objB:32
objB:33
objA:28
objA:29
objA:30
objB:34
objA:31
objA:32
objB:35
objA:33
objA:34
objB:36
objA:35
objA:36
objB:37
objA:37
objA:38
objA:39
objA:40
objB:38
objB:39
objB:40
objA:41
objA:42
objA:43
objA:44
objA:45
objB:41
objA:46
objA:47
objA:48
objB:42
objA:49
objB:43
objA:50
objA:51
objA:52
objA:53
objA:54
objB:44

经检测，这已经达到我们所期望的结果。

一个逗逗

一个逗逗

相关推荐

线程池源码分析-ThreadPoolExecutor

针对JDK1.7中的线程池。从上一篇文章中了解到：核心execute方法需要被子类来实现，所以我们就俩重点看看ThreadPoolExecutor是如何实现这个核心方法的。long keepAliveTime：和TimeUnit unit一起构成线程的最大

ljngya / 0评论 2020-07-30

线程池源码分析-FutureTask

针对JDK1.7中的线程池。* Executes the given command at some time in the future. The command. 可以提交一个Callable，并且返回一个Future用于追踪提交的任务。有一个状态变

齐天大圣数据候 / 0评论 2020-07-30

netty教程

Netty 是由 JBOSS 提供的一个 Java 开源框架。Netty 提供异步的、基于事件驱动的网络。应用程序框架，用以快速开发高性能、高可靠性的网络 IO 程序。个网络应用，相当于简化和流程化了 NIO 的开发过程。通信行业等获得了广泛的应用，知名的

爱传文档 / 0评论 2020-07-28

高并发系统设计（六）：如何减少频繁创建数据库连接的性能损耗？

数据库的调用方式是先获取数据库的连接，然后依靠这条连接从数据库中查询数据，最后关闭连接释放数据库资源。这种调用方式下，每次执行SQL都需要重新建立连接，频繁地建立数据库连接耗费时间长导致了访问慢的问题。来看一个实际的测试。我用"tcpdump -

jameszgw / 0评论 2020-06-25

第53天： Python 线程池

大家都知道当任务过多，任务量过大时如果想提高效率的一个最简单的方法就是用多线程去处理，比如爬取上万个网页中的特定数据，以及将爬取数据和清洗数据的工作交给不同的线程去处理，也就是生产者消费者模式，都是典型的多线程使用场景。那是不是意味着线程数量越多，程序的执

fengling / 0评论 2020-05-31

python 非阻塞模式 线程池

return "完成任务:{} 用时:{} 进程ID:{}".format. container.append # 添加返回内容到container，这里也可以对返回的函数进行其它的处理。tasks = ["任务1&qu

meylovezn / 0评论 2020-05-30

Arthas | 定位线上 Dubbo 线程池满异常

阿里云高级开发工程师。Dubbo 线程池满异常应该是大多数 Dubbo 用户都遇到过的一个问题，本文以 Arthas 3.1.7 版本为例，介绍如何针对该异常进行诊断，主要使用到 dashboard?Cloud Toolkit 是阿里云发布的免费本地 ID

大步流星 / 0评论 2020-05-28

常用的后端性能优化六种方式：缓存化+服务化+异步化等

硬件问题对性能的影响不容忽视。后来DBA同学帮忙定位到问题是硬件过旧导致，将机械硬盘升级成固态硬盘之后报警立马消失了，效果立竿见影！异步化可以利用线程池、消息队列等方式实现。使用线程池的时候一定要注意核心参数的设置，可以通过监控工具去观测实际创建、活跃、空

Ken专注后端技术 / 0评论 2020-05-07

时间轮算法 — 转

　　自然界中定时任务无处不在，太阳每天东升西落，候鸟的迁徙，树木的年轮，人们每天按时上班，每个月按时发工资、交房租，四季轮换，潮涨潮落，等等，从某种意义上说，都可以认为是定时任务。大概很少有人想过，这些“定时”是怎样做到的。当然，计算机领域的同学们可能对此

nurvnurv / 0评论 2020-05-01

《Python》线程池、携程

concurrent.futures模块提供了高度封装的异步调用接口。Both implement the same interface, which is defined by the abstract Executor class.取代for循环sub

cenylon / 0评论 2020-04-29

华为Android三面成功通过，面试官都问了什么？

2020年的春招太难了，很多小伙伴说不知道怎样才能顺利通过面试，明明这些技术性问题自己也清楚，但是每次面试官问到的时候还是答不上来，再或者说，之前有学过这一类的，或者做过同类型的项目，但是就是很担心，不知道大厂到底问的一些什么问题，今天我就将我的经历整理出

fengyeezju / 0评论 2020-04-26

线程池原理详解及如何用C语言实现线程池

线程池是一种多线程处理形式，大多用于高并发服务器上，它能合理有效的利用高并发服务器上的线程资源;线程与进程用于处理各项分支子功能，我们通常的操作是：接收消息 ==> 消息分类 ==> 线程创建 ==> 传递消息到子线程 ==> 线

csde / 0评论 2020-03-05

任务队列方案详解（一）JVM线程池

我们都知道 web 服务的工作大多是接受 http 请求，并返回处理后的结果。服务器接受的每一个请求又可以看是一个任务。一般而言这些请求任务会根据请求的先后有序处理，如果请求任务的处理比较耗时，往往就需要排队了。而同时不同的任务直接可能会存在一些优先级的变

nbfcome / 0评论 2020-02-21

netty 的 worker 线程池

pipeline 添加 handler 的时候，如果没有指定线程池，则使用 channel 的 IO 线程池，即 NioEventLoop。所以，NioEventLoop 的作用是，轮询 SocketChannel 的网络事件，同时也可以处理 handle

huaye00 / 0评论 2020-02-18

被开发者抛弃的 Executors，错在哪儿？

在 Java 领域内，我们使用多线程的方式来实现并发编程。而线程本身是操作系统的一个概念，虽然不同的语言对线程都进行了一些封装，但是最终都是调用到操作系统中去创建和调度线程。既然线程是一项重要的系统资源，为了更合理的利用此资源，我们会使用池化技术来优化线

xdhywj / 0评论 2020-02-13

Spring 线程池

corePoolSize：线程池至少有2个线程是启动的，即使是空闲的也不会关闭。maxPoolSize：最大的线程数目，当corePoolSize繁忙时，会创建线程，启动的总的线程数不能大于maxPoolSize. 才会在corePoolSize的基础上,

greensomnuss / 0评论 2012-09-11

源码学习之线程池

大家面试过程中肯定被问道过线程池。因为在系统中频繁创建线程会造成很大的CPU消耗。而且用完的线程要等待GC回收也会造成消耗。下面我们就来学习下最常用的线程池ThreadPoolExecutor，线程池状态:RUNNING:取30位到32位111代表运行中,

xzkjgw / 0评论 2020-01-23

Java线程池源码阅读

　　线程池是池化技术的一种，对线程复用、资源回收、多任务执行有不错的实践。阅读源码，可以学习jdk的大师对于线程并发是怎么池化的，还有一些设计模式。同时，它也能给我们在使用它的时候多一种感知，出了什么问题可以马上意识到哪里的问题。使用如下api创建一个固定

少年阿涛 / 0评论 2020-01-12

高并发解决方案与并发编程,线程池,缓存,消息队列,分库分表java系统架构

通过不同的域名可以让浏览器直接访问资源服务器而不需要再访问应用服务器了。这个问题可以用先生成静态页面然后使用Ajax来读取并修改响应的数据，这样就可以一举两得来，既可以使用页面缓存也可以实时显示一些变化频率高的数据来。其实大家都知道，效率最高、消耗最小的就

MichelinMessi / 0评论 2020-01-07

分布式场景下Kafka消息顺序性的思考

如果业务中，对于kafka发送消息异步消费的场景，在业务上需要实现在消费时实现顺序消费， 利用kafka在partition内消息有序的特点，消息消费时的有序性。否则即使通过发送时指定partition，在消费端由于线程池的异步消费，消息之间的处理都是并发

jiangkai00 / 0评论 2020-01-04

都说衣不如新人不如故，技术是学新不学旧的？IPC+view+Handler+线程。

刚刚过去2019，新的一年2020年。都说衣不如新人不如故，技术是学新不学旧的？可是旧的知识不巩固，根基不固很容易在面试或者实战遇到很大的问题的。四种线程池的使用场景，线程池的几个参数的理解？提供定时执行、定期执行、单线程、并发数控制等功能。如果线程池中的

csdnuuu / 0评论 2020-01-03

python_并发编程——线程池

t_pool = ThreadPoolExecutor # 创建线程池对象,设置线程池大小，建议不超过cpu数*5. 结果：每过两秒钟随机打印5个数。t_pool.shutdown() # 线程池被执行完，才会执行主线程的内容。t_pool.map

georgeandgeorge / 0评论 2020-01-02

java线程池源码的简单分析

　　工作中用过线程池来进行多线程的操作，但是也仅仅是停留在使用方面，没有深入研究，现在通过源码来仔细研究下java的线程池。关于线程池的优缺点就不研究了，直接通过一个源码来看看java中线程池的原理。* 非核心线程的在没有任务执行的时候存活的时间设置为0，

Bloddy / 0评论 2019-12-28

Django异步任务线程池

当数据库数据量很大时（百万级），许多批量数据修改请求的响应会非常慢，一些不需要即时响应的任务可以放到后台的异步线程中完成，发起异步任务的请求就可以立即响应。选择用线程池的原因是：线程比进程更为可控。不像子进程，子线程会在所属进程结束时立即结束。线程可共享内

meylovezn / 0评论 2019-12-16

多线程与线程池

1，GIL全局解释器锁。CPython解释器自动加载GIL锁，基于单核CPU的GIL锁，为了保证数据的安全，单个进程的多线程不能利用多核，即可以并发，但是不能并行，多个进程可以并发，并行。进入多核时代，GIL锁则成了python的一个弱点。以为CPytho

meylovezn / 0评论 2019-12-13

Java: 线程池(ThreadPoolExecutor)中的参数说明

新建线程时，必须通过线程池提供，不允许在应用中自行显式创建线程。

fengyeezju / 0评论 2019-12-09

SpringBoot-技术专区-实战方案-应用监控线程池

　　废话不多说，做这个监控的背景很简单，我们的项目都是以spring boot框架为基础开发的，代码里所有的异步线程都是通过@Async标签标注的，并且标注的时候都是指定对应线程池的，如果不知@Async标注的，可以参考@Async异步线程池用法总结, 如

88996371 / 0评论 2019-12-01

Android线程

1）new CachedThreadPool 创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，如无可回收线程，则新建线程。2）new FixedThreadPool 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待

pengjin / 0评论 2019-12-01

Android线程池封装库

轻量级线程池封装库，支持线程执行过程中状态回调监测；支持创建异步任务，并且可以设置线程的名称，延迟执行时间，线程优先级，回调callback等；可以根据自己需要创建自己需要的线程池，一共有四种；线程异常时，可以打印异常日志，避免崩溃。关于线程池，对于开发来

jling / 0评论 2019-11-17

网络通信与治理，谁更在行？Envoy和Nginx架构层面的对比

Nginx是Envoy出现之前网络通信中间件领域非常有代表性的开源系统，功能强大，性能出色，扩展性很强，已经形成了强大的生态，成为HTTP流量管理领域事实上的标杆。Envoy作为后起之秀，虽然定位和目标上与Nginx有不少差异，但架构设计层面，Envoy和

guoshaoliang / 0评论 2019-11-12

程序员楼下闲聊：某次jvm崩溃排查

大望路某写字楼下。来来来，请你喝饮料，好好聊聊当时的情况。“猿A”是我的同事兼死党，和他详聊后大概明白了当时的场景。业务处理部分，使用了线程池FixedThreadPool。

weixiaohuai / 0评论 2019-11-04

周期性线程池与主要源码解析

之前学习ThreadPool的使用以及源码剖析，并且从面试的角度去介绍知识点的解答。今天小强带来周期性线程池的使用和重点源码剖析。工作线程的的执行过程：工作线程会从DelayQueue取出已经到期的任务去执行；执行结束后重新设置任务的到期时间，再次放回De

yangxiaobo / 0评论 2019-11-04

周期性线程池与主要源码解析

之前学习ThreadPool的使用以及源码剖析，并且从面试的角度去介绍知识点的解答。今天给大家介绍下周期性线程池的使用和重点源码剖析。工作线程的的执行过程：工作线程会从DelayQueue取出已经到期的任务去执行；执行结束后重新设置任务的到期时间，再次放回

SUDevops / 0评论 2019-11-03

面试必问的并发编程知识点，你知道多少？

众所周知，在Java的知识体系中，并发编程是非常重要的一环，也是面试的必问题，一个好的Java开发者是必须对并发编程这块有所了解的。而如何在并发场景中写出优良的代码，是一道绕不开的坎，也是考量一个 Java 开发者功底的关键技术。于是发奋学习，啃大部头书又

讨厌什么变成什么 / 0评论 2019-11-03

Elasticsearch常用配置及性能参数

cluster.name: estest 集群名称node.name: “testanya” 节点名称

AFei00 / 0评论 2017-09-15

第三十一天

Event是python的threading模块中的一个类，它的作用是可以通过一些线程来控制另一些线程。因为它是一个类，所以首先需要先实例化一个对象，通过对象来调用它的一些方法。该方法可以将线程变为false，即阻塞状态。# 让汽车线程进入运行或就绪状态。

georgeandgeorge / 0评论 2019-10-24

Tornado实现多进程/多线程的HTTP服务

返回内容用self.write方法输出。构造函数的参数是一个Handlers列表，通过正则表达式，将请求与Handler对应起来。通过dict将Handler需要的其他对象以参数的方式传递给Handler的initialize方法。因此，我们考虑用线程池的

chenzhanhai / 0评论 2019-04-09

python多线程扫描端口(线程池)

扫描服务器ip开放端口，用线程池ThreadPoolExecutor，i7的cpu可以开到600个左右现成,大概20s左右扫描完65535个端口，根据电脑配置适当降低线程数

小炮集揣 / 0评论 2019-09-04

Android 高级面试题

Handler，Message，looper和MessageQueue构成了安卓的消息机制，handler创建后可以通过sendMessage将消息加入消息队列，然后looper不断的将消息从MessageQueue中取出来，回调到Hander的handl

冰川孤辰 / 0评论 2019-09-08

手撕ThreadPoolExecutor线程池源码

通过对execute方法、addWorker方法、Worker类、runWorker方法、getTask方法、processWorkerExit从源码角度详细阐述，文末有彩蛋。* workerCountOf方法取出低29位的值，表示当前活动的线程数；* 如

keyxiaodj / 0评论 2019-09-08

Android的线程池：ExecutorService和Executors

ExecutorService是线程池的一个服务，可以随时关闭线程池，是继承Executor的。Executors是个工厂类，专门创建各种线程池。Android常用的线程池有一下几种，在Executors里面对应的方法：1. newFixedThreadP

Helene / 0评论 2012-05-31

线程池及增长策略和拒绝策略

ThreadPoolExecutor类实现了ExecutorService接口和Executor接口，可以设置线程池corePoolSize，最大线程池大小，AliveTime，拒绝策略等。2如果此时线程池中的数量等于corePoolSize，但是缓冲队列

85251048 / 0评论 2018-11-06

tomcat 线程池的使用

1，应用背景：项目压力测试，要求同时支持并发数量8000，5台tomcat配置。tomcat 中Connector是用来创建request和response对象，并用于处理请求和处理交换数据的；然后将请求发送给Engine，Engine处理完成之后再通过C

XHuiLin / 0评论 2018-10-25

Linux下用C实现线程池

简单的说，如果一个应用需要频繁的创建和销毁线程，而任务执行的时间又非常短，这样线程创建和销毁的带来的开销就不容忽视，这时也是线程池该出场的机会了。如果线程创建和销毁时间相比任务执行时间可以忽略不计，则没有必要使用线程池了。下面是Linux系统下用C语言创建

kvikon / 0评论 2011-12-11

servlet3的异步特性

Servlet3 比 Servlet2.4最大的进步应该是异步支持了，不再像过去那样，对于每个请求，只有一个线程在处理，必须等待这个线程阻塞处理完毕后，才能给 client 返回响应。使用异步处理，新开一条线程处理耗时的 task，这样线程不会一直堵塞，可

88407013 / 0评论 2015-01-28

dubbo线程池溢出解决

最近在处理项目，压测的时候发现，dubbo线程池溢出，查看官方文档找到解决方案，如下：根据自己的需要，改成需要参数

wangyangsoftware / 0评论 2018-03-29

关于找工作

我：最后把处理业务的逻辑封装成一个task提交给一个新建的业务线程池中执行，执行完之后由work线程池执行请求的write事件。

moodytong / 0评论 2017-03-25

elasticsearch高级配置之线程池设置

一个Elasticsearch节点会有多个线程池，但重要的是下面四个：索引：主要是索引数据和删除数据操作搜索：主要是获取，统计和搜索操作批量操作：主要是对索引的批量操作更新：主要是更新操作。可以通过给设置一个参数来改变线程池的类型，例如，把索引的线程池改成

数据的启示 / 0评论 2014-10-10

Netty初步之hello world

Java的网络操作一直比较复杂，虽然说在加入NIO之后简单了些，但还不是我这些菜鸟玩得起的，由于存在大量低层操作和协议处理，所以在使用上面还是很难。迄今为止，还没有用NIO写出稳定可靠的网络操作，也许这和具体的应用需求较少也有关系吧。Netty提供异步的、

arctan0 / 0评论 2012-09-25