# 1. 引言
HTTP是大部分互联网应用接口的首选协议，但是由于HTTP协议短连接且是单向请求(request/response)的特性，决定了调用方要获得实时结果，需要不断的轮询(Polling)服务接口。从而造成大量无意义的请求及服务器相应的开销。针对此现状，许多方案应运而生。比如基于XMPP pubsub的方案、基于HTTP的web-hook的方案、适合即时通讯的comet方案等。但是由于HTTP的简洁及标准的力量，上述方案都没有得到大规模的流行HTTP Polling的现状暂时无人能够改变。



PubSubHubbub是Google推出的一个基于Web-hook方式的解决方案，它包括PubSubHubbub协议及一个开源的参考实现(Reference Implementation)

# 2. 原理
原理及数据流图再官网的slide上已经有详细的描述，这里以静态图补充：
![](https://www.ericturbo.top/static/img/40ac6c8950a1c9190a145598d19f7ca3.image.webp)
# 3.价值
## Publisher发布方
许多Blog服务提供者来说，RSS对它们来说是一个鸡肋，对运营及广告等业务没什么帮助，但是却流量很大。因此他们经常非常矛盾的维护着这个接口。如果PubSubHubbub能够在业界大范围的适用，至少从访问压力层面解除了BSP对提供RSS接口之忧。

**特例 Realtime RSS(Twitter, 微博服务等)**

Twitter/微博等realtime RSS可以从此方案受益，按照常规的方案，订阅方为了获取realtime的结果，几乎需要以每分钟1次的频率来访问RSS API, 如果订阅方能够以PubSubHubbub的方式来访问RSS，那么RSS API的请求量几乎可以降为0

## Subscriber订阅方
Subscriber比如RSS阅读器，搜索引擎等类似业务。Google Reader看似PubSubHubbub最大的赢家。
另外在有hub的前提下，即使Publisher不支持PubSubHubbub, subscriber可以通过hub直接取到feed内容，就是说类似阅读器这样的应用现在就可以完全切换到PubSubHubbub体系上。

# 在python中的应用
Python的订阅发布模式工具：RabbitMQ，Pypi，PubSubHubbub， NATS。RabbitMq中有发布订阅模式，学习这个，可以更好地理解此协议

PubSubHubbub的价值

# 1.关于Rss的介绍
RSS(really simple Syndication)是一种消息来源格式规范，用以聚合经常发布更新数据的网站，例如博客文章、新闻、音频或视频的网摘。RSS文件包含了全文或是节录的文字，再加上发用者所订阅之网摘布数据和授权的元数据。



## 1.1 Rss的意义
对某一行业密切相关的几百个甚至几千个Rss的种子进行的聚合，将能快速、全面了解某一行的最新动态；对某一行业的几十个甚至几百个网站进行完整的数据下载，并进行数据挖掘，将能了解某一主题再该行业发展的来龙去脉。

## 1.2 运用
RSS在科研院所的应用。 高能物理信息监测对象为全球高能物理同行机构：实验室、行业学会、国际协会、各国主管科研政府机构、重点综合性科学出版物、高能物理试验项目和实验设施。监控的信息类型为：新闻、论文、会议报告、分析评论、预印本、案例研究、多媒体、图书、招聘信息等。

高能物理文献信息所采用最现金的开源内容管理系统Drupal，开源搜索技术Apache Solr，以及Google员工开发的能实时订阅新闻的PubSubHubbub技术和Amazon的OpenSearch，建立了一套高能物理信息检测系统，有别于传统的Rss订阅和推送，实现了几乎实时的信息抓取和任意关键词、任意类别、复合条件新闻的主动推送。

# 2. 网页搜索爬虫时效性系统
网页爬虫的目标：
* 覆盖率高
* 死链率低
* 实效性好
爬虫实效性的目标也差不多，主要是实现新网页快速和全面的收录。下图为时效性系统的整体架构：


![image.png](/static/img/fce9a00fc1d49215af3a301c219d4e98.image.webp)
  
主要分为：
* Rss/sitemap子系统
* 网页泛爬的调度系统Webmain scheduler
* 时效性模块 Vertical Scheduler
* DNS服务，抓取的时候，一般是几十台甚至几百台抓取集群，如果每一台都有防御的话，对DNS的压力比较大，所以一般有一个DNS的服务模块来做全局的服务。

涉及到实效性的模块包括以下几个：
1. Rss/sitemap系统：时效性系统利用Rss/sitemap的过程是挖掘种子，定时抓取，解析链接发布时间，将较新的网页优先抓取并索引。
2. 泛爬系统：泛爬系统设计良好的话有助于提交时效性网页的高覆盖率，但泛爬需要尽可能缩短调度周期
3. 种子调度系统：主要是一个时效性的种子库，这个种子库里面有一些信息调度系统会不断地扫描这个数据库，然后发给抓取集群，这个集群抓取完全进行一些抽取链接的处理，接下来把这些按类别发出去，各个垂直频道会获取到时效性的数据。
4. 种子的挖掘：涉及到页面接卸或其他的一些挖掘手段，可以通过站点地图，还有导航条来构建，还要基于页面结构特征和页面百变得更规律
5. 种子的更新机制：记录每个种子的抓取历史，follow的链接信息，定期根据种子的外链更新特征，重新计算种子的更新周期。
6. 抓取系统与JavaScript解析：使用浏览器进行抓取，搭建一个基于浏览器抓取的抓取集群。或采用开源项目，如Qtwebkit。

[参考链接](https://www.yisu.com/zixun/128054.html)

> 帮助：
1. [Drupal介绍_php开发](https://www.leixue.com/ask/what-is-drupal)
2. [Apache Solr_java开发](https://baike.baidu.com/item/apache%20solr/1548066?fr=aladdin)
3. [open search](https://www.zhihu.com/question/24564105)

[mkdocs的官网](https://squidfunk.github.io/mkdocs-material/)

我做的第一个wiki网站
![image.png](/static/img/9dea764b7868721d25def57a048d23c4.image.webp)




使用VanBlog的时间1年多了

![image.png](/static/img/95639a0d7831f9b12519f60497ecf9f1.image.webp)

个人感觉还是很不错的，页面简洁，而且稳定性高，不需要考虑很多东西，启动方便。

特此留下记录，纪念我曾经折腾的mkdocs，里边的一些流程图等没有生效，所以不太好使，而且需要跟开发似的，push到代码库中才行。



写博客的初衷，就像图片写到的：当你啥都不会的时候，先从模仿开始。

加油~~~

搬迁-将mkdocs的内容迁移到此blog中

## How to Choose The Method
### 1. One Task Vs Multiple Tasks
可以通过指定目标函数和每次调用目标函数的参数可迭代对象来向进程池发出多个调用。可以通过以下函数实现这一点：map()、map_async()、imap()、imap_async()、starmap() 和 starmap_async()。



发出单个任务：

- 使用 apply()
- 使用 apply_async()

发出多个任务：

- 使用 map()
- 使用 map_async()
- 使用 imap()
- 使用 imap_unordered()
- 使用 starmap()
- 使用 starmap_async()
### 2. Blocking vs Non-Blocking
阻塞调用：

- 使用 apply()
- 使用 map()
- 使用 starmap()

非阻塞调用：

- 使用 apply_async()
- 使用 map_async()
- 使用 imap()
- 使用 imap_unordered()
- 使用 starmap_async()
### 3. Lazy vs Non-lazy
它们立即向进程池发出所有任务。这意味着提供的可迭代对象被遍历，并且所有对目标函数和产生的参数的调用都被转换成任务并保存在内存中。

我们可以将这些函数称为非惰性的。它们包括 map()、map_async()、starmap() 和 starmap_async()。

其他函数则逐一向进程池发出任务，只有在进程池中有空闲空间执行新任务时才会发出。这意味着提供的可迭代对象会逐个遍历，按需创建和发出任务。

我们可以将这些函数称为惰性的，可能更节省内存。它们包括 imap() 和 imap_unordered()。

惰性调用（逐一）：

- 使用 imap()
- 使用 imap_unordered()

非惰性调用（一次性）：

- 使用 apply()
- 使用 apply_async()
- 使用 map()
- 使用 map_async()
- 使用 starmap()
- 使用 starmap_async()
### 4. Single Arguments vs Multiple Arguments
没有参数的目标函数：

- 使用 apply()
- 使用 apply_async()

带有一个参数的目标函数：

- 使用 apply()
- 使用 apply_async()
- 使用 map()
- 使用 map_async()
- 使用 imap()
- 使用 imap_unordered()
- 使用 starmap()
- 使用 starmap_async()
具有多个参数的目标函数：

- 使用 apply()
- 使用 apply_async()
- 使用 starmap()
- 使用 starmap_async()
### 5. Ordered Results vs Unordered Results
有序结果：

- 使用 map()
- 使用 map_async()
- 使用 imap()
- 使用 starmap()
- 使用 starmap_async()

无序结果：

- 使用 imap_unordered()
### 6. Result Callback vs No Result Callbacks
支持回调函数：

- 使用 apply_async()
- 使用 map_async()
- 使用 starmap_async()

不支持回调函数：

- 使用 apply()
- 使用 map()
- 使用 imap()
- 使用 imap_unordered()
- 使用 starmap()
## Compare Methods
### 1. apply vs apply_async
apply() 和 apply_async() 都可以用于向进程池发出一次性任务。

它们的主要区别如下：

- apply() 函数会阻塞，而 apply_async() 函数不会阻塞。
- apply() 函数会返回目标函数的结果，而 apply_async() 函数会返回一个 AsyncResult 对象。
- apply() 函数不支持回调函数，而 apply_async() 函数支持回调函数。
### 2. map vs map_async
map() 和 map_async() 都可以用于通过进程池向可迭代对象中的所有项发出调用函数的任务。

它们的主要区别如下：

- map() 函数会阻塞，而 map_async() 函数不会阻塞。
- map() 函数会返回从目标函数返回的可迭代对象，而 map_async() 函数会返回一个 AsyncResult 对象。
- map() 函数不支持回调函数，而 map_async() 函数可以在返回值和错误上执行回调函数。

### 3. imap vs imap_unordered
imap() 和 imap_unordered() 函数有很多共同点，例如：

- imap() 和 imap_unordered() 都可以通过进程池向可迭代对象中的所有项发出调用函数的任务。
- imap() 和 imap_unordered() 都是 map() 函数的惰性版本。
- imap() 和 imap_unordered() 函数都会立即返回一个返回值的可迭代对象。

然而，这两个函数之间有一个关键的区别：

- imap() 返回的可迭代对象会按照任务完成的顺序逐个产生结果，而 imap_unordered() 函数则会以任务完成的任意顺序产生结果。
### 4. starmap vs starmap_async
starmap() 和 starmap_async() 都可以用于通过进程池发出调用带有多个参数的函数的任务。

它们的主要区别如下：

- starmap() 函数会阻塞，而 starmap_async() 函数不会阻塞。
- starmap() 函数会返回从目标函数返回的可迭代对象，而 starmap_async() 函数会返回一个 AsyncResult 对象。
- starmap() 函数不支持回调函数，而 starmap_async() 函数可以在返回值和错误上执行回调函数。

### 5. pool.map vs built-in map()
Pool.map() 函数和内置的 map() 函数都可以遍历提供的可迭代对象，并调用目标函数，将可迭代对象的每个项传递给目标函数执行。

它们的主要区别如下：

- Pool.map() 函数会立即遍历可迭代对象以发出所有任务，而内置的 map() 函数是惰性的，只有在遍历从目标函数返回的返回值的可迭代对象时才会遍历输入的可迭代对象。
- Pool.map() 函数会使用子进程并行执行每个调用目标函数的任务，而内置的 map() 函数会在当前进程中执行每个调用目标函数的任务。
- Pool.map() 函数支持单个可迭代对象作为目标函数的参数，而内置的 map() 函数支持多个可迭代对象，每个可迭代对象对应目标函数的一个参数。
- Pool.map() 函数支持指定块大小（chunksize）来批量调用目标函数的任务，而内置的 map() 函数不支持批量调用目标函数的任务。

### 6. pool.starmap() vs itertools.starmap()
Pool.starmap() 和 itertools.starmap() 函数都执行一个目标函数，该函数可能具有多个参数，使用一个提供的可迭代对象，其中每个项都是一个参数迭代器，用于每个函数调用的参数。

它们的主要区别如下：

- Pool.starmap() 函数会立即遍历可迭代对象以发出所有任务，而内置的 itertools.starmap() 函数是惰性的，只有在遍历从目标函数返回的返回值的可迭代对象时才会遍历输入的可迭代对象。
- Pool.starmap() 函数会使用子进程并行执行每个调用目标函数的任务，而内置的 itertools.starmap() 函数会在当前进程中执行每个调用目标函数的任务。
- Pool.starmap() 函数支持指定块大小（chunksize）来批量调用目标函数的任务，而 itertools.starmap() 函数不支持批量调用目标函数的任务。
### 7. imap() vs map()
imap() 和 map() 都可以用于通过进程池向可迭代对象中的所有项发出调用函数的任务。

它们的主要区别如下：

- imap() 函数逐一向进程池发出任务，而 map() 函数一次性向进程池发出所有任务。
- 在遍历返回值时，imap() 函数会阻塞直到每个任务完成，而 map() 函数会阻塞直到所有任务完成。

## Common Questions
### 1. How to Issue a single Task to the Process Pool
您可以使用 apply() 或 apply_async() 函数向进程池发出单个任务。
### 2. how to issue Tasks When the Target Function has no arguments
您可以使用 apply() 或 apply_async() 函数发出调用无参数目标函数的任务。
### 3. how to call map() for a function with multiple Arguments
您可以使用 starmap() 或 starmap_async() 函数向进程池发出针对接受多个参数的目标函数的任务。
### 4. how to issue tasks Asynchronously
您可以使用 apply_async()、map_async() 和 starmap_async() 函数异步向进程池发出任务。

此外，imap() 和 imap_unordered() 函数不会阻塞。
### 5. is the imap_unordered() Asynchronous?
不完全正确，但有点像。

imap_unordered() 的异步性质不同于 apply_async()、map_async() 和 starmap_async() 函数。具体来说，imap_unordered() 函数不会返回 AsyncResult 对象。

然而，imap_unordered() 函数和 imap() 函数一样，也不会阻塞。它会立即返回，并且只有在尝试通过遍历返回的可迭代对象来获取任务结果时才会阻塞。

因此，可以说imap_unordered()函数可以异步使用。
### 6. is there a imap_async() Function
No

这可能是因为 API 提供的 imap_unordered() 函数本身就是异步的，因为它不会阻塞。然而，imap_unordered() 函数确实不会返回 AsyncResult 对象。
### 7. why Ever use apply()
如果调用 apply() 会阻塞，为什么还要使用它呢？为什么不直接调用函数呢？

主要原因是为了让函数调用在一个单独的子进程中执行。
### 8. How do you call many different Target Functions
我们可以使用 apply() 或 apply_async() 在循环中调用多个不同的目标函数。

我们也可以使用多个分开的调用来通过 map()、map_async()、imap()、imap_unordered()、starmap() 和 starmap_async() 函数调用多个不同的函数。
### 9. why bother use imap() instead of map
您会使用 imap() 而不是 map()，这样可以在结果可用时立即开始处理，而不是阻塞等待所有结果可用。

此外，imap() 使用更少的内存，因为它会延迟遍历提供的输入可迭代对象，根据需要逐步发出任务。
### 10. how do you best set chunksize?

map()、imap() 和 starmap() 函数都接受一个 "chunksize" 参数。

chunksize 指定了要分组并发送到工作子进程中的目标函数调用的数量。

这可以通过减少将数据参数发送到子进程和从子进程接收结果的计算开销，加速整体任务的执行。

最佳的 chunksize 值取决于您的应用程序，具体取决于正在执行的任务、它们的持续时间、发送到每个任务的数据以及从每个任务返回的数据。

您可以通过一些试验和仔细的基准测试找到最佳的 chunksize 值。

一些可以尝试的值包括：

- chunksize=1，即不分块的默认值。
- chunksize = 任务数 / 工作进程数，平均分配。
- chunksize = 任务数 / 工作进程数 * 0.25，稍微少于平均分配的一部分。

Process Pool how to Choose and compare

## need to initialize Worker Process
我们可以使用诸如 `apply()` 的函数向进程池发出临时任务，也可以使用诸如 `map()` 的函数将相同的函数应用于一个可迭代的项目集合。发出的任务的结果可以同步获取，或者可以稍后使用诸如 `apply_async()` 和 `map_async()` 的异步版本的函数获取任务的结果。

在使用进程池时，我们可能需要初始化一个变量、数据或资源，以便每个工作进程在执行其任务时都可以使用该资源。例如，可能需要为每个进程分配一个日志记录的句柄或连接到远程服务器的连接，并在执行任务时保持打开和重复使用。

我们需要一种方法，在执行任何任务之前，调用一个函数来初始化进程池中每个工作进程。



## how to initialize Worker Process
例如：

```python
# 工作进程初始化函数
def worker_init():
    # ...

# 创建进程池并初始化工作进程
pool = multiprocessing.pool.Pool(initializer=worker_init)
```

如果我们的工作进程初始化函数需要参数，可以通过 `initargs` 参数将这些参数传递给进程池的构造函数，`initargs` 接受一个有序的列表或元组，用于传递自定义初始化函数的参数。

例如：

```python
# 工作进程初始化函数
def worker_init(arg1, arg2, arg3):
    # ...

# 创建进程池并初始化工作进程
pool = multiprocessing.pool.Pool(initializer=worker_init, initargs=(arg1, arg2, arg3))
```

现在我们知道如何在进程池中初始化工作进程了，让我们看一个示例。

## Example of initizing worker process
```python
# SuperFastPython.com
# example of initializing worker processes in the process pool
from time import sleep
from multiprocessing.pool import Pool
 
# task executed in a worker process
def task():
    # report a message
    print('Worker executing task...', flush=True)
    # block for a moment
    sleep(1)
 
# initialize a worker in the process pool
def initialize_worker():
    # report a message
    print('Initializing worker...', flush=True)
 
# protect the entry point
if __name__ == '__main__':
    # create and configure the process pool
    with Pool(2, initializer=initialize_worker) as pool:
        # issue tasks to the process pool
        for _ in range(4):
            pool.apply_async(task)
        # close the process pool
        pool.close()
        # wait for all tasks to complete
        pool.join()
```

输出结果：
```
Initializing worker...
Worker executing task...
Initializing worker...
Worker executing task...
Worker executing task...
Worker executing task...
```
## Exapmle of Thread and Process Names the initialize
```python
# SuperFastPython.com
# example of initializing workers in a process pool and reporting threads and processes
from time import sleep
from multiprocessing.pool import Pool
from multiprocessing import current_process
from threading import current_thread
 
# task executed in a worker process
def task():
    # get the current process
    process = current_process()
    # get the current thread
    thread = current_thread()
    # report a message
    print(f'Worker executing task, process={process.name}, thread={thread.name}', flush=True)
    # block for a moment
    sleep(1)
 
# initialize a worker in the process pool
def initialize_worker():
    # get the current process
    process = current_process()
    # get the current thread
    thread = current_thread()
    # report a message
    print(f'Initializing worker, process={process.name}, thread={thread.name}', flush=True)
 
# protect the entry point
if __name__ == '__main__':
    # create and configure the process pool
    with Pool(2, initializer=initialize_worker) as pool:
        # issue tasks to the process pool
        for _ in range(4):
            pool.apply_async(task)
        # close the process pool
        pool.close()
        # wait for all tasks to complete
        pool.join()
```
运行这个示例会创建线程池，并配置它以便使用我们的自定义函数初始化工作进程，就像之前一样。

四个任务被提交到进程池，并且主进程等待所有任务完成。

在初始化工作进程和执行任务时，报告了工作进程及其进程内线程的名称。

有趣的是，我们可以看到执行任务的进程和初始化工作进程的进程是相同的，而且在每种情况下使用的线程也是相同的。

例如，第一个工作进程的名称是“SpawnPoolWorker-1”，它在主线程“MainThread”中执行初始化函数和任务函数。

这突显了我们可以在工作进程中使用全局变量和线程局部存储等机制来在工作进程执行的任务之间共享数据。

结果：
```
Initializing worker, process=SpawnPoolWorker-1, thread=MainThread
Worker executing task, process=SpawnPoolWorker-1, thread=MainThread
Initializing worker, process=SpawnPoolWorker-2, thread=MainThread
Worker executing task, process=SpawnPoolWorker-2, thread=MainThread
Worker executing task, process=SpawnPoolWorker-1, thread=MainThread
Worker executing task, process=SpawnPoolWorker-2, thread=MainThread
```






## Example of Accessing an initialized Variable in a worker

```python
# SuperFastPython.com
# example of sharing data with all workers via the initializer
from time import sleep
from multiprocessing.pool import Pool
 
# task executed in a worker process
def task():
    # declare global variable
    global custom_data
    # report a message with global variable
    print(f'Worker executing with: {custom_data}', flush=True)
    # block for a moment
    sleep(1)
 
# initialize a worker in the process pool
def worker_init(custom):
    # declare global variable
    global custom_data
    # assign the global variable
    custom_data = custom
    # report a message
    print(f'Initializing worker with: {custom_data}', flush=True)
 
# protect the entry point
if __name__ == '__main__':
    # define data to share with all workers
    data = 'Global State'
    # create and configure the process pool
    with Pool(2, initializer=worker_init, initargs=(data,)) as pool:
        # issue tasks to the process pool
        for _ in range(4):
            pool.apply_async(task)
        # close the process pool
        pool.close()
        # wait for all tasks to complete
        pool.join()
```

结果：
```
Initializing worker with: Global State
Worker executing with: Global State
Initializing worker with: Global State
Worker executing with: Global State
Worker executing with: Global State
Worker executing with: Global State
```