2019-05-12

Celery源码分析（一）-------------从命令执行到生成Worker

从今天起开始Celery源码分析系列文章。有需要的同学可以关注一上下，主要目的就是理清celery的核心对象及处理流程，方便大家学习。大家有需要详细讲解的部分可以告诉我，如果有时间会详细讲解和解答，感谢大家的支持。

最开始的几篇先从整体流程的角度进行梳理和分析，对Celery的关键对象和整体架构有了认识之后。后面再逐渐结合详细的代码进行分析。

第一篇从一个常用的命令”celery -A tasks worker”开始，讲述从命令执行开始到生成Worker对象的流程。

下面是涉及到的对象的时序图：

对照时序图，主要过程如下：

1.每一个命令对应不同的Command子类，“celery woker”命令对应的”celery.bin.worker::worker”类就是Command类的一个子类。

2.命令执行后，会调用CeleryCommand的execute_from_commandline函数，这个函数会根据参数进行一些特殊的处理操作，然后调用Command类的setup_app_from_commandline方法，这个方法比较重要，因此用红色标出。这个函数的作用是导入我们编写的tasks模块，然后得到我们自己写的tasks中的app对象，这个对象是一个”celery.app.base::Celery”对象。

3.Command的execute方法查找对应命令的类并生成对象，这里是Worker命令对象。不同的命令对象通过call方法调用自身的run函数。

4.然后命令会调用自己的app对象，这个app就是2中从我们自己任务模块中获取到的app.通过app对象的subclass_with_self方法生成一个celery.apps.worker::Worker对象。subclass_with_self方法会动态创建一个celery.apps.worker::Worker类的子类，这个子类会包含我们的app对象作为类属性，这也是subclass_with_self方法名字的由来。

这就是从命令执行到生成Worker对象的过程，下一节会分析Worker对象的作用及初始化流程。

作者：self-motivation
来源：CSDN
原文：https://blog.csdn.net/happyAnger6/article/details/53869262
版权声明：本文为博主原创文章，转载请附上博文链接！

2019-05-11

编程语言►Golang

Go实现控制任程序的生命周期

runner/runner.go:

package runner

import (
“errors”
“os”
“os/signal”
“time”
)

type Runner struct {
interrupt chan os.Signal

complete chan error

timeout <-chan time.Time

tasks []func(int)

}

var ErrTimeout = errors.New(“received timeout”)

var ErrInterrupt = errors.New(“received interrupt”)

func New(d time.Duration) *Runner {
return &Runner{
interrupt: make(chan os.Signal, 1),
complete: make(chan error),
timeout: time.After(d),
}
}

func (r *Runner) Add(tasks …func(int)){
r.tasks = append(r.tasks, tasks…)
}

func (r *Runner) Start() error {
signal.Notify(r.interrupt, os.Interrupt)

go func() {
    r.complete <- r.run()
}()

select {
case err := <-r.complete:
    return err
case <-r.timeout:
    return ErrTimeout
}

}

func (r *Runner) run() error {
for id, task := range r.tasks {
if r.gotInterrupt() {
return ErrInterrupt
}

    task(id)
}

return nil

}

func (r *Runner) gotInterrupt() bool {
select {
case <-r.interrupt:
signal.Stop(r.interrupt)
return true

1
2
3

default:
    return false
}

}

runner/main/main.go:
package main

import (
“log”
“time”
“os”
“runner”
)

const timeout = 3 * time.Second

func main(){
log.Println(“Starting work.”)

r := runner.New(timeout)

r.Add(createTask(), createTask(), createTask())

if err := r.Start(); err != nil {
    switch err {
    case runner.ErrTimeout:
        log.Println("Terminating due to timeout.")
        os.Exit(1)
    case runner.ErrInterrupt:
        log.Println("Terminating due to interrupt.")
        os.Exit(2)
    }

    log.Println("Process ended.")
}

}

func createTask() func(int) {
return func(id int){
log.Printf(“Processor - Task #%d.”, id)
time.Sleep(time.Duration(id) * time.Second)
}
}

作者：self-motivation
来源：CSDN
原文：https://blog.csdn.net/happyAnger6/article/details/70558324
版权声明：本文为博主原创文章，转载请附上博文链接！

2019-05-11

开源软件►Scrapy

scrapy源码分析（三）---------ExecutionEngine（二）一个request的周期

上一篇中讲解了ExecutionEngine的主循环流程，下面就具体讲解下不需要搁置时，如何处理一个request，从下载页面到解析页面，最后到数据处理的整个流程。

几个核心的类介绍如下：

1.Scraper:刮取器。用于对下载后的结果进行处理，主要使用ItemPipelineManager对数据进行入数据库等操作。

2.Downloader：下载器。对同时下载网页的并发度进行控制，同时通过DownloaderMiddlewareManager来对request,response进行各个中间件的操作。并通过HTTP11DownloadHandler来使用twisted的连接池进行网页下载操作。

工作流程图如下:

作者：self-motivation
来源：CSDN
原文：https://blog.csdn.net/happyAnger6/article/details/53401912
版权声明：本文为博主原创文章，转载请附上博文链接！

2019-05-11

开源软件►Scrapy

scrapy源码分析（二）----------ExecutionEngine（一）主循环

ExecutionEngine是scrapy的核心模块之一，顾名思义是执行引擎。
它驱动了整个爬取的开始，进行，关闭。

它又使用了如下几个主要模块来为其工作：
1.slot:它使用Twisted的主循环reactor来不断的调度执行Engine的”_next_request”方法，这个方法也是核心循环方法。下面的
流程图用伪代码描述了它的工作流程，理解了它就理解了引擎的核心功能。
另外slot也用于跟踪正在进行下载的request。

2.downloader:下载器。主要用于网页的实际下载

3.scraper:数据抓取器。主要用于从网页中抓取数据的处理。也就是ItemPipeLine的处理。

根据上面的分析可知，主要是_next_request在不断的进行工作，因此这个函数重点分析，流程图如下:

流程详解：
1.这个_next_request方法有2种调用途径，一种是通过reactor的5s心跳定时启动运行，另一种则是在流程中需要时主动调用。

2.如果没有暂停，则运行。判断是否需要搁置？这个判断条件如右边紫色框中讲的，有4种需要搁置的条件。如果不需要搁置，
则执行3;如果需要搁置，则执行4.

3.获取一个request,这个获取是从队列中获取。获取到则通过下载器下载（这个是Deferred实现的，因此是异步的）。如果
没有request了，则执行4;如果一直有，则不断的执行2.

4.判断start_requests如果有剩余且不需要搁置，则获取一个，并调用crawl方法，这个方法只是将request放入队列。这样，
3中就能获取到了;如果没有start_requests了或者需要搁置则执行5.

5.判断spider是否空闲，这里需要判断没有任何下载任务，没有任务数据处理任务，没有start_requests了，没有阻塞的requests了。
只有都满足才可能关闭并结束。

后面教程将会对执行引擎的下载器，slot,数据scraper等进行详细讲解，欢迎关注。

作者：self-motivation
来源：CSDN
原文：https://blog.csdn.net/happyAnger6/article/details/53385856
版权声明：本文为博主原创文章，转载请附上博文链接！

2019-05-11

开源软件►Scrapy

scrapy源码分析（一）---------------------总执行流程概览

scrapy是一个基于twisted实现的开源爬虫，要读懂其源码，需要对twisted的异步编程模型有一定了解。可以通过之前3篇deferred的相关教程了解。

下面是总结的执行一个爬虫任务的整体执行流程，请将图片放大查看，即运行”scrapy crawl xxxSpider”的执行流程：

流程中主要的颜色框的含义如下：

1.红色框是模块或者类。

2.紫色框是向模块或者类发送的消息，一般为函数调用。

3.红色框垂直以下的黑色框即为本模块或者对象执行流程的伪代码描述。

几个关键的模块和类介绍如下：

cmdline:命令行执行模块，主要用于配置的获取，并执行相应的ScrapyCommand。

ScrapyCommand:命令对象，用于执行不同的命令。对于crawl任务，主要是调用CrawlerProcess的crawl和start方法。

CrawlerProcess:顾名思义，爬取进程，主要用于管理Crawler对象，可以控制多个Crawler对象来同时进行多个不同的爬取任务，并调用Crawler的crawl方法。

Crawler:爬取对象，用来控制一个爬虫的执行，里面会通过一个执行引擎engine对象来控制spider从打开到启动等生命周期。

ExecutionEngine:执行引擎，主要控制整个调度过程，通过twisted的task.LoopingCall来不断的产生爬取任务。

请关注后面的教程，将会详细介绍各个模块的作用和关键代码实现。

作者：self-motivation
来源：CSDN
原文：https://blog.csdn.net/happyAnger6/article/details/53367108
版权声明：本文为博主原创文章，转载请附上博文链接！

2019-05-11

编程语言►python

集合Sets

Sets
Python还提供了集合类型。集合是没有重复元素的无序集合。集合的基本使用包括成员检测和消除重复元素。集合对象也支持数学上的并集，交集，差集，异或运算。

{}或者set() 函数可以用来创建集合。注意：创建一个空集合必须使用set()，而不能使用{}。因为{}是一个空字典。

Here is a brief demonstration:

basket = {‘apple’, ‘orange’, ‘apple’, ‘pear’, ‘orange’, ‘banana’}
print(basket) # 重复元素会被删除
{‘orange’, ‘banana’, ‘pear’, ‘apple’}
‘orange’ in basket # 快速成员检测
True
‘crabgrass’ in basket
False

下面展示两个单词的集合操作

…
a = set(‘abracadabra’)
b = set(‘alacazam’)
a # a中的唯一元素
{‘a’, ‘r’, ‘b’, ‘c’, ‘d’}
a - b # 只在a中的元素
{‘r’, ‘d’, ‘b’}
a b # 在a或b中的元素
{‘a’, ‘c’, ‘r’, ‘d’, ‘b’, ‘m’, ‘z’, ‘l’}
a & b # 在a,b中都出现的元素
{‘a’, ‘c’}
a ^ b # 没有同时在a,b中出现的元素
{‘r’, ‘d’, ‘b’, ‘m’, ‘z’, ‘l’}
和列表生成式类似，集合也支持集合生成式：

a = {x for x in ‘abracadabra’ if x not in ‘abc’}
a

{‘r’, ‘d’}

作者：self-motivation
来源：CSDN
原文：https://blog.csdn.net/happyAnger6/article/details/88700686
版权声明：本文为博主原创文章，转载请附上博文链接！

2019-05-11

python►python入门教程►编程语言

list

下同是使用list方法的一些例子:

a = [66.25, 333, 333, 1, 1234.5]
print(a.count(333), a.count(66.25), a.count(‘x’))
2 1 0
a.insert(2, -1)
a.append(333)
a
[66.25, 333, -1, 333, 1, 1234.5, 333]
a.index(333)
1
a.remove(333)
a
[66.25, -1, 333, 1, 1234.5, 333]
a.reverse()
a
[333, 1234.5, 1, 333, -1, 66.25]
a.sort()
a
[-1, 1, 66.25, 333, 333, 1234.5]
a.pop()
1234.5
a
[-1, 1, 66.25, 333, 333]

可以发现，insert,remove,sort方法返回None,这是python中可变数据结构的一个设计原则（就地改变返回None)。
像栈一样使用Lists
列表的方法可以使我们很容易地像栈一样使用list。

使用append()来向栈顶添加元素，然后使用不带索引的pop()从线顶弹出元素。下面是例子:

stack = [3, 4, 5]
stack.append(6)
stack.append(7)
stack
[3, 4, 5, 6, 7]
stack.pop()
7
stack
[3, 4, 5, 6]
stack.pop()
6
stack.pop()
5
stack
[3, 4]
像队列一样使用list
将list作为队列使用并不高效，因为从list尾部添加和弹出元素很快，但是从list头部插入和弹出元素则比较慢（因为需要移动元素)

要实现队列的功能，可以使用从两端添加和弹出元素都很快的collections.deque . 下面是例子:

from collections import deque
queue = deque([“Eric”, “John”, “Michael”])
queue.append(“Terry”) # Terry arrives
queue.append(“Graham”) # Graham arrives
queue.popleft() # The first to arrive now leaves
‘Eric’
queue.popleft() # The second to arrive now leaves
‘John’
queue # Remaining queue in order of arrival
deque([‘Michael’, ‘Terry’, ‘Graham’])
列表生成式
列表生成式提供了方便创建lists的方法。

通常我们有需要在另外的序列或可迭代对象的每个元素上施加一些操作来生成新的列表，或者创建一个满足一些特定条件的子序列。

比如，创建一个平方的列表:

squares = []
for x in range(10):
… squares.append(x**2)
…
squares
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
上面创建的x变量会一直存在到循环结束。可以通过下面的方法消除这种副作用::

squares = list(map(lambda x: x**2, range(10)))

或者，通过下面更简洁，可读性更好的等价方法:
squares = [x**2 for x in range(10)]

列表生成式由[]内的一个表达式和紧跟其后的for,if语句构成。列表生成式的结果是满足这些条件的值。

比如，下面生成由两个列表中不相等元素构成的元组列表:

[(x, y) for x in [1,2,3] for y in [3,1,4] if x != y]
[(1, 3), (1, 4), (2, 3), (2, 1), (2, 4), (3, 1), (3, 4)]

和下面是等价的:

combs = []
for x in [1,2,3]:
… for y in [3,1,4]:
… if x != y:
… combs.append((x, y))
…
combs
[(1, 3), (1, 4), (2, 3), (2, 1), (2, 4), (3, 1), (3, 4)]

注意，for和if语句的顺序是怎样的。
如果表达式是一个元组，则()是不可少的。

vec = [-4, -2, 0, 2, 4]

创建2倍元素的列表

[x*2 for x in vec]

[-8, -4, 0, 4, 8]

过滤负数

[x for x in vec if x >= 0]

[0, 2, 4]

在元素上调用函数

[abs(x) for x in vec]

[4, 2, 0, 2, 4]

调用元素的函数

freshfruit = [‘ banana’, ‘ loganberry ‘, ‘passion fruit ‘]

[weapon.strip() for weapon in freshfruit]

[‘banana’, ‘loganberry’, ‘passion fruit’]

创建(number, square)的元组列表

[(x, x**2) for x in range(6)]
[(0, 0), (1, 1), (2, 4), (3, 9), (4, 16), (5, 25)]

如果产生元组必须使用括号，否则会抛出异常

[x, x2 for x in range(6)]

File “”, line 1, in ?

[x, x2 for x in range(6)]

^
SyntaxError: invalid syntax

扁平化一个2维列表

vec = [[1,2,3], [4,5,6], [7,8,9]]

[num for elem in vec for num in elem]

[1, 2, 3, 4, 5, 6, 7, 8, 9]

列表生成式可以包含复杂的表达式和嵌套的函数调用:

from math import pi

[str(round(pi, i)) for i in range(1, 6)]

[‘3.1’, ‘3.14’, ‘3.142’, ‘3.1416’, ‘3.14159’]

列表生成式嵌套
列表生成式的初始表达式可以是任意的表达式，甚至可以是另外一个列表生成式。

考虑下面3*4的矩阵:

matrix = [
… [1, 2, 3, 4],
… [5, 6, 7, 8],
… [9, 10, 11, 12],
… ]

下面是矩阵的转置:

[[row[i] for row in matrix] for i in range(4)]
[[1, 5, 9], [2, 6, 10], [3, 7, 11], [4, 8, 12]]

嵌套的表达式在后面的for循环中依次计算:

transposed = []
for i in range(4):
… transposed.append([row[i] for row in matrix])
…
transposed
[[1, 5, 9], [2, 6, 10], [3, 7, 11], [4, 8, 12]]

也和下面等价:

transposed = []
for i in range(4):
… # the following 3 lines implement the nested listcomp
… transposed_row = []
… for row in matrix:
… transposed_row.append(row[i])
… transposed.append(transposed_row)
…
transposed
[[1, 5, 9], [2, 6, 10], [3, 7, 11], [4, 8, 12]]
在实际中，我们可能更喜欢使用内置的函数，zip()函数十分擅长这种操作

list(zip(*matrix))
[(1, 5, 9), (2, 6, 10), (3, 7, 11), (4, 8, 12)]

del语句
可以通过del语句在指定位置删除元素:它和pop()的区别是没有返回值。del语句还可以从列表中删除一个切片或者清空列表:

a = [-1, 1, 66.25, 333, 333, 1234.5]
del a[0]
a
[1, 66.25, 333, 333, 1234.5]
del a[2:4]
a
[1, 66.25, 1234.5]
del a[:]
a
[]
del 还可以用来删除整个变量:

del a

后续对a的引用会报错，除非又赋了新值。

作者：self-motivation
来源：CSDN
原文：https://blog.csdn.net/happyAnger6/article/details/88674851
版权声明：本文为博主原创文章，转载请附上博文链接！

2012-02-18

自我修养►数据结构与算法

大数运算经典:棋盘上的米粒。

有一个8*8共64格的棋盘，要求第一个格放1粒米，第二个格放2粒米，第三个放4粒，依次加倍，问放满棋盘共有多少粒米。

很显然，这是个很大的数，用long型也无法存放。所以只能通过模拟加法和乘法来达到计算的目的。

include

using namespace std;

//模拟乘2运算
void multi2(short a[])
{
short carry = 0; //进位
for(short i=999;i>=0;i–)
{
short tmp =a[i]*2;
a[i]=tmp%10+carry; //当前位结果
carry=tmp/10; //进位

}

return;
}

//模拟加运算

void plus(short sum[],short add[])
{
short carry = 0;
for(short i=999;i>=0;i–)
{
short tmp = sum[i]+add[i];
sum[i]=tmp%10 + carry;
carry=tmp/10;

}

return;

}

int main()
{
short sum[1000]={0};
short add[1000]={0};

add[999]=1;

for(short i=0;i<64;i++)
{

plus(sum,add);

multi2(add);
}

short k = 0; /结果中的0不打/
for(short j=0;j<1000;j++)
{
if(sum[j]!=0)
{
k=j;
break;
}
}

cout<<”result:”<<endl;
for(short i=k;i<1000;i++)
cout<<sum[i];

cout<<endl;
getchar();
return 0;

}