最近在做嵌入式开发相关的事情，接触到一种新的操作系统，叫VxWorks。VxWorks不仅是一种嵌入式实时操作系统，又是可以运行的最小基本程序。其内部有BSP（Board Support Package，板级支持包），便于进行驱动程序的编写。此外，VxWorks具有强实时性、微内核设计、可裁剪性、可移植性和可靠性等特点，能较好地满足嵌入式开发的需求。
 
实时多任务
 
非实时应用系统可以利用单进程或多进程结构按照事件发生的顺序一次处理，而实时系统则要求按照发生事件的轻重缓急（优先级）来安排处理顺序。对于某些任务，强实时系统甚至要求在给定时间内处理完成。这一点从根本上决定了实时多任务机制的基本功能。
 
任务的状态
 
任务被创建后，其状态可以在有限的范围内迁移，从而构成一个简单的有限状态机。任何任务在某一时刻只能处于一种状态，基本状态包括：挂起状态（Suspend）、准备状态（Ready）、阻塞状态（Pend）、延迟状态（Delay）和运行状态（Running）。各种状态的含义及转移状态图与非实时系统类似，此处不做赘述。
 
任务的调度方式
 
任务调度最主要的因素是任务的优先级（Priority）。VxWorks支持256个优先级（0~255），数字越小优先级越高。任务调度的另一个因素是调度算法。VxWorks支持两种调度算法，第一种是基于优先级的抢占式调度算法，另一种是同一优先级任务之间所采用的时间片轮转算法，系统默认为第一种算法。
 
 
基于优先级的抢占式调度算法。
 
  
 
算法的基本思想：顾名思义，一个具有更高优先级的任务一旦进入ready状态，将抢占当前运行任务的CPU资源，进行上下文切换后进入运行状态。
 
 
算法的优点很明显，且符合大多数实时系统的调度需求，因此是最常见的一种选择。
 
 
缺点是：在多个任务具有相同优先级的情况下，如果当前执行的任务一直不被阻塞，它将一直占有CPU，从而造成其他同优先级或者低优先级任务无法执行。
 
 
时间片轮转调度算法
 
  
 
基本思想：对于处于同一优先级且处于Ready状态的任务来说，算法能使它们分时共享CPU。将任务分为多组，每一组任务具有相同的优先级。从组内的第一个任务开始，每个任务在执行一个时间片后让出CPU，下一个任务开始执行直到组内任务都执行过之后，同组第一个任务重新得到CPU，依次循环。
 
任务编程
 
任务的创建
 
任务创建主要采用taskSpawn函数（还可以用taskInit函数和taskActivate函数，这两者使用时需要同时使用，所以一般使用taskSpawn）。
 
int taskSpawn
 (
     char    *name,          /*任务的名字*/
     int     priority,       /*任务的优先级*/
     int     options,        /*任务的操作（可选择）*/
     int     stackSize,      /*任务分配的堆栈大小*/
     FUNCPTR entryPt,        /*任务的入口地址（入口函数）*/
     int     arg1,           /*函数可能带的参数（入口函数的实际参数，不足10个用0补足）*/
     int     arg2,
     ...
     int     arg10
 )
 
option参数的选择范围如下：
 
 
VX_FP_TASK (0x0008)：运行时使用浮点运算器，将浮点寄存器作为任务上下文的一部分
 
 
VX_PRIVATE_ENV (0x0180)：允许使用任务的环境变量
 
 
VX_NO_STAKC_FILL (0x0100)：人物创建时不以0xEE去填充堆栈，不能使用checkStack()
 
 
VX_UNBREAKABLE (0x0002)：不允许使用断点调试功能
 
调用成功，返回任务ID；否则返回ERROR并利用errno告知错误原因：
 
 
S_intLib_NOT_ISR_CALLABLE：程序不能从一个中断服务程序中调用
 
 
S_objLib_OBJ_ID_ERROR：不正确的Task ID
 
 
S_memLib_NOT_ENOUGH_MEMORY：没有足够的内存
 
 
S_memLib_BLOCK_ERROR：不能对内存分区互斥访问
 
 系统中同时只允许一个进程使用的资源成为临界资源；
 访问临界资源的代码成为临界区
 
系统任务
 
 
根任务：tUsrRoot。内核执行的第一个任务，该任务完成初始化功能并负责创建其他必要的系统任务。
 
 
日志任务：tLogTask。用来输出系统日志信息的任务。在其他任务中，可以调用LogMsg（）向tLogTask发送消息日至。与printf相比，好处是无需发起I/O操作，并节省格式化信息所需的大量堆栈空间。中断服务程序可以用LogMsg，不能使用printf。
 
 
异常处理任务：tExcTask。用来完成中断不能执行的功能。
 
 
网络任务：tNetTask。用于提供网络所需要使用的任务级功能。
 
 
目标代理任务：tWdbTask。当目标代理设置为运行在任务模式下时，创建这个任务，处理来自Tornado的目标服务请求。
 
Shell命令观察和控制
 
 
tt：显示任务的函数栈的调用状态
 
 
ts：挂起某个任务
 
 
tr：恢复某个任务
 
 
td：删除某个任务
 
欢迎关注个人网站，https://zhangfeng-fitz.com

 
今天说一个在实际项目中特别实用的解决并发耗时问题的办法：异步任务处理。这里采用 redis list 结构来实现。涉及知识点：
 
1、redis list 结构
 
2、阻塞、非阻塞、同步、异步的概念
 
3、如何实现一个异步处理任务实战
 
同步、异步、阻塞、非阻塞
 
首先来说同步和异步，这两个概念是针对通信双方消息传送的响应来说的，如果 A 请求 B，B 马上响应 A，这是同步，而如果 A 请求 B ，B 说好的，我已经接受任务，然后把任务交由 C，而 C 是一个专门负责处理任务的，这种模型就是异步。
 
而阻塞和非阻塞，通常对应于程序请求响应，比如我们常见的 IO 处理，一个进程在处理 IO 操作时，需要等待 IO 操作完成，如果等着，就是阻塞，如果这个时候它不等着，而是转身先去做别的事情，走到 IO 操作完成再回来处理，这种就是非阻塞。
 
在网上看到一个比较好的关于这两组概念的区别，即同步和异步是描述行为的，即我是如果去处理当有类似阻塞这种情况的请求，比如 IO 操作，比如耗时操作，而阻塞和非阻塞描述的是一种状态。
 
redis 列表介绍
 
首先在面试中经常会问一个问题：redis 支持哪几种数据结构？
 
网上一般的回答是五种，分别是 string、list、set、sorted set、hash，在官方文档还有其他三种，Bitmaps、HyperLogLogs，还有 stream，参考链接如下：
 
https://redis.io/topics/data-types-intro
 
其中 stream 类型是在 5.0 中引入的。这里不详细介绍每种的用法，感兴趣的读者可以自行查看官方文档。因为要用到 list 这种数据结构，所以简单介绍下。
 
redis 的 list 结构，也可以称为是队列，它和 Python 中的 list 相似，它可以按照插入的顺序存储数据，但不同的是 Python 中的 list 底层是数组实现的，而 redis 中的 list 底层是链表实现的，所以 redis 中的 list 无论是在头部还是在尾部插入元素，时间复杂度都是常量级别的。通常的操作命令：
 
> rpush mylist 1 2 3 4 5 "foo bar"
 
(integer) 9
 
> lrange mylist 0 -1
 
1) "first"
 
2) "A"
 
3) "B"
 
4) "1"
 
5) "2"
 
6) "3"
 
7) "4"
 
8) "5"
 
9) "foo bar"
 
当然，与之对应的还有 lpush, lpop 等，所有操作指令可以在以下链接进行查看：
 
https://redis.io/commands#list
 
数据结构存在的意义是根据不同的场景可以达到更高的效率，那 list 结构有什么作用呢？通常有以下两个作用：
 
1、存储最近的数据，比如用一个列表存储用户访问的记录，每次访问时插入，而如果需要取最近访问的 10 条，只需要使用 lrange(key, 0, 9) 来获取即可
 
2、存储任务，即作为一个要处理的任务列表。也就是我们这篇文章里要介绍的异步任务，它去存储需要异步处理的任务列表
 
此外，除了普通的列表，还有两种特殊列表的存在，一种是限制列表，即限制列表的长度，比如我们只需要记录用户最近访问的30条记录，这样做的好处是保证数据不会无限增长，从业务需求上来说也是合理的，因为比较老的数据已经没有太大的价值(当然如果非要保存这些旧的数据，也可以考虑持久化)。我们可以使用 ltrim 指令来完成，示例如下：
 
> rpush mylist 1 2 3 4 5
 
(integer) 5
 
> ltrim mylist 0 2
 
OK
 
> lrange mylist 0 -1
 
1) "1"
 
2) "2"
 
3) "3"
 
另外一种是阻塞列表，即使用 brpop 或者 blpop 。我们都知道 redis 是单线程的，通常是一个命令执行完，才会去执行下一个命令，那试想这样一种情况，如果一个 redis 列表一直为空，我们一直循环去判断它有没有数据，这样是不是很浪费资源，做了很多无用功，所以通常我们需要在读数据时，判断如果没有数据就阻塞几秒，然后再去读数据。这里可以说得有些难理解，后面我们讲了异步实现代码之后，就好理解了。我们先来看一下 brpop 使用示例：
 
# 如果没有数据，停 5s
 
> brpop tasks 5
 
1) "tasks"
 
2) "do_something"
 
异步任务处理代码实现
 
实现的原理其实也很简单，当客户端发起请求，服务端接收到请求，可能需要完成很多复杂的逻辑，有些甚至是包括一些耗时的操作，这个时候我们可以将这个耗时的操作当作一个任务，把它塞到 redis 的 list 中，而在另一边开启一个常驻脚本，即 while True 的循环，一直去从 list 中读取数据(管理这种常驻脚本，也可以使用 pm2，或者其他工具，这里不做介绍，简单地直接 run 一个 python 脚本)。
 
这里我们模拟服务端接收到请求后，往 redis list 里塞数据，实现代码如下：
 
# -*- coding: utf-8 -*-
 
"""
 
@Time : 2019/7/6 下午10:51
 
@Author : yrr
 
@File : task_client.py
 
@Desc : push task to redis list
 
"""
 
# need install redis， pip install redis
 
import redis
 
import json
 
def main():
 
conn = redis.Redis(host='localhost', port=6379, db=0)
 
# Redis Lists are simply lists of strings, sorted by insertion order.
 
# so we need json.dumps
 
conn.lpush("async_task_list", json.dumps({
 
# 参数可自定义规则
 
'name': 'test',
 
'data': {'name': 'demon'}
 
}))
 
if __name__ == '__main__':
 
main()
 
而异步任务脚本实现如下：
 
# -*- coding: utf-8 -*-
 
"""
 
@Time : 2019/7/6 下午10:57
 
@Author : yrr
 
@File : task_async.py
 
@Desc : task async handle
 
"""
 
import redis
 
import json
 
def handle_test(data):
 
print("has get the data by redis list...")
 
print("the name is %s" % data['name'])
 
TASKS_FUNC_MAP = {'test': handle_test}
 
def main():
 
conn = redis.Redis(host='localhost', port=6379, db=0)
 
while True:
 
# 需要保证异常不影响正常异步任务
 
# 所以需要用 try 来控制
 
try:
 
# 这里需要用到阻塞式列表，因为一直 for 循环会消耗cpu资源
 
# 可以设置阻塞时间
 
# 这也是一个面试考点
 
task = conn.brpop('async_task_list', 2)
 
if not task:
 
continue
 
_, data = task
 
print("curent handle data: %s" % (data,))
 
data = json.loads(data)
 
fname = TASKS_FUNC_MAP.get(data['name'])
 
if not fname:
 
# 通常这里需要报错，暂时处理为打印错误信息
 
print('not func handle, please check')
 
continue
 
fname(data)
 
except Exception as e:
 
# 这里涉及到错误任务的重试机制，可以根据业务需求来做
 
# 这里暂时不处理，只打印出异步信息
 
print(e)
 
if __name__ == '__main__':
 
main()
 
实现效果如下：https://www.zhihu.com/video/1130982109149556736

 
目录
 
一、前言
二、验证目的
三、验证流程
1、工作线程直接无锁处理
2、工作线程加锁处理
3、使用信号槽发送任务
4、使用postEvent
  
四、结论

 
一、前言
 
命名这个博客的名称也让我费了点时间，因为之前我在网上找过相关资料，但并没有这方面的资料和讨论，也可能是我太钻牛角尖非要搞个清楚。之前我做IOCP库的时候，想把这个高并发网络库做的更精益求精，用了很多晦涩的逻辑和技术来实现无锁化，但这只是建立在我一直认为发送事件的效率比锁的速度要快，今天终于有时间来验证一下这个想法，但结果却让我大跌眼镜。当然也有可能是我的思路或者方法有问题，希望有大神了解这方面的可以指点指教。
 
二、验证目的
 
因为IOCP的实现中，会开多个工作线程来接受数据和处理任务，也会遇到处理临界区资源的情况，我的无锁化实现实质就是将这些数据都Post给主线程，让主线程来处理所有的任务，这样的好处就是让工作线程尽快回到队列中取数据，且主线程处理任务就不用加锁，运行效率会更高（我自认为的）。现在为了验证到底是工作线程加锁来处理任务、还是工作线程将任务Post给主线程的效率高。
 
三、验证流程
 
为了简化程序，我只用了一个工作线程和主线程来完成验证，任务很简单，就是给一个全局变量加1，直到给定的一个最大值。
 
1、工作线程直接无锁处理
 
这个是最快速的，让我们先看看最快的速度是多少。
 相关代码：
 
#define COUNT 5000000

    void run()
    {
        while(count <= COUNT)
        {
            count++;
        }
        qApp->exit();
    }
 
测试结果为15ms左右。
 
2、工作线程加锁处理
 
假设变量count是个临界区，对它修改时要加锁。
 相关代码：
 
#define COUNT 5000000

    void run()
    {
        while(1)
        {
            mutex.lock();
            if(count <= COUNT)
            {
                count++;
                mutex.unlock();
            }
            else
            {
                mutex.unlock();
                break;
            }
            //sleep(0);
        }
        //qDebug() << "3:" << Worker::time.elapsed();
        qApp->exit();
    }

 
测试结果为100ms左右。
 
3、使用信号槽发送任务
 
开始我认为效率最高的处理办法吧。
 相关代码：
 
signals:
    void s_opt();
	//。。。
    connect(this, SIGNAL(s_opt()), this, SLOT(opt()), Qt::AutoConnection);
	//。。。
    Q_INVOKABLE void opt() {
        count++;
    }
	//。。。
    void run()
    {
        for(int i = 0; i <= COUNT; i++)
        {
            emit s_opt();
        }
        QMetaObject::invokeMethod(qApp, "quit", Qt::AutoConnection);
    }
 
满怀信心的点了运行，然后闷逼的等了一会，给我显示个出来：7709ms。
 思来想去，是不是我点运行的姿势不对？多点了几次，也就这个值左右。
 我感觉不可思议，但也不死心，把代码改了一下：
 

    void run()
    {
        for(int i = 0; i <= COUNT; i++)
        {
            QMetaObject::invokeMethod(pobj, "opt", Qt::AutoConnection); 
        } 
        QMetaObject::invokeMethod(qApp, "quit", Qt::AutoConnection);
    }
 
但还是试试吧。点了运行后，结果果然差不多。
 还不死心，继续改了一下，将AutoConnection改为QueuedConnection，继续测试！
 没想到结果是：29070ms！
 我有点蒙逼了，两个问题：1是用信号槽或invokeMethod不至于慢上几十倍吧？2是AutoConnection在跨线程不是默认QueuedConnection吗？
 
4、使用postEvent
 
思来想去，再试试发送事件的办法吧，毕竟信号槽在跨线程的实现上也是基于这个的。
 相关代码：
 
    void run()
    { 
        for(int i = 0; i <= COUNT; i++)
        {
            CustomEvent *customEvent = new CustomEvent;
            QCoreApplication::postEvent(this, customEvent); 
        } 
        QMetaObject::invokeMethod(qApp, "quit", Qt::AutoConnection);
    }
    bool event(QEvent *event)
    {
        if (event->type() == CustomEvent::eventType())
        {
            count++; 
            return true;
        }
        return Worker::event(event);
    }
 
小心翼翼的按下运行，结果显示为：29002ms。
 果然问题在这里，难道是堆上建立Event对象耗费时间吗？那把代码稍微再改改：
 
    void run()
    { 
        for(int i = 0; i <= COUNT / 1000; i++)
        {
            CustomEvent *customEvent = new CustomEvent[1000];
            for(int j = 0; j < 1000; j++)
            {
                QCoreApplication::postEvent(this, customEvent + j);
            } 
        } 
        QMetaObject::invokeMethod(qApp, "quit", Qt::AutoConnection);
    }
 
一次申请1000个Event对象，速度应该会提高些吧？
 生气的按下运行，程序直接奔溃。找了半天也不知什么原因。。
 
四、结论
 
这个结论让我吃惊不小，没想到用锁都比事件传递的效率快，但无需质疑的一点是，因为IOCP建议工作线程要尽快回到系统循环中，所以如果遇到IO或高密集计算，还是得让其他线程来处理。
 
至于为什么发送事件会如此耗时，个人认为除了新建、删除事件对象会耗时外，postEvent函数是由于线程安全的，所以本身应该也有一些耗时处理，加上唤醒主线程这些额外的损耗，综合这些原因才导致慢了这么多。
 
这里没有讨论的一种是原子操作，但理论上应该是比第一种（线程直接操作）慢一点，比第二种（加锁）快。这里就不做讨论了。

Netty 中有很多场景依赖定时任务实现，比较典型的有客户端连接的超时控制、通信双方连接的心跳检测等场景。在学习 Netty Reactor 线程模型时，我们知道 NioEventLoop 不仅负责处理 I/O 事件，而且兼顾执行任务队列中的任务，其中就包括定时任务。为了实现高性能的定时任务调度，Netty 引入了时间轮算法驱动定时任务的执行。时间轮到底是什么呢？为什么 Netty 一定要用时间轮来处理定时任务呢？JDK 原生的实现方案不能满足要求吗？
 
 
 
说明：本文参考的 Netty 源码版本为 4.1.42.Final。
 
 
 
 
 
文章目录
 
 

  
定时任务的基础知识
  
Timer
  
DelayedQueue
  
ScheduledThreadPoolExecutor
  
时间轮原理分析
  
Netty HashedWheelTimer 源码解析
  

   

    
接口定义
    
快速上手
    
内部结构
    
添加任务
    
工作线程 Worker
    
停止时间轮
   
 
  
时间轮