发表于 | 分类于 | | 阅读次数:
| 字数统计:

缓存模式（Cache Aside Pattern）

什么是”Cache Aside Pattern”?

答： 旁路缓存方案的经验实践，这个实践又分读实践，写实践。

对于读请求
  • 先读cache，再读db
  • 如果，cache hit，则直接返回数据
  • 如果，cache miss，则访问db，并将数据set回缓存

如上图：

  1. 先从cache中尝试get数据，结果miss了
  2. 再从db中读取数据，从库，读写分离
  3. 最后把数据set回cache，方便下次读命中
对于写请求
  • 淘汰缓存，而不是更新缓存
  • 先操作数据库，再淘汰缓存

如上图：

  1. 第一步要操作数据库，第二步操作缓存
  2. 缓存，采用delete淘汰，而不是set更新

Cache Aside Pattern为什么建议淘汰缓存，而不是更新缓存？

答： 如果更新缓存，在并发写时，可能出现数据不一致。

如上图所示，如果采用set缓存。

在1和2两个并发写发生时，由于无法保证时序，此时不管先操作缓存还是先操作数据库，都可能出现：

  1. 请求1先操作数据库，请求2后操作数据库
  2. 请求2先set了缓存，请求1后set了缓存

导致，数据库与缓存之间的数据不一致。

所以，Cache Aside Pattern建议，delete缓存，而不是set缓存。

Cache Aside Pattern为什么建议先操作数据库，再操作缓存？

答： 如果先操作缓存，在读写并发时，可能出现数据不一致。

如上图所示，如果先操作缓存。

在1和2并发读写发生时，由于无法保证时序，可能出现：

  1. 写请求淘汰了缓存
  2. 写请求操作了数据库（主从同步没有完成）
  3. 读请求读了缓存（cache miss）
  4. 读请求读了从库（读了一个旧数据）
  5. 读请求set回缓存（set了一个旧数据）
  6. 数据库主从同步完成
    导致，数据库与缓存的数据不一致。

所以，Cache Aside Pattern建议，先操作数据库，再操作缓存。

Cache Aside Pattern方案存在什么问题？

答： 如果先操作数据库，再淘汰缓存，在原子性被破坏时：

  1. 修改数据库成功了
  2. 淘汰缓存失败了

导致，数据库与缓存的数据不一致。

原文来自58沈剑:
https://cloud.tencent.com/info/5e258c6f840423de68eda937db3e1c90.html

缓存问题

用一张脑图总结如下：

发表于 | 分类于 | | 阅读次数:
| 字数统计:

使用 slots

__slots__变量是用来限制class能添加的属性

例如

1
2
3
>>> class Student(object):
...     __slots__ = ('name', 'age') # 用tuple定义允许绑定的属性名称
...

1
2
3
4
5
6
7
>>> s = Student() # 创建新的实例
>>> s.name = 'Michael' # 绑定属性'name'
>>> s.age = 25 # 绑定属性'age'
>>> s.score = 99 # 绑定属性'score'
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
AttributeError: 'Student' object has no attribute 'score'

从上面代码示例我们可以看到由于score没有被放到__slots__中，所以不能绑定score属性，试图绑定score将得到AttributeError的错误。
使用__slots__要注意，__slots__定义的属性仅对当前类起作用，对继承的子类是不起作用的：

1
2
3
4
5
>>> class GraduateStudent(Student):
...     pass
...
>>> g = GraduateStudent()
>>> g.score = 9999

除非在子类中也定义__slots__，这样，子类允许定义的属性就是自身的__slots__加上父类的__slots__。

详情可以查看廖雪峰教程：
https://www.liaoxuefeng.com/wiki/897692888725344/923030542875328

关于six.with_metaclass(ABCMeta, object)的理解

元类就是创建类的类，在Python中，只有type类及其子类才可以当元类，type创建类时，参数格式：type(classname, parentclasses,attrs), classname是类名，字符串类型，parentclasses是类的所有父类，元组类型，attrs是类的所有{属性:值}，是字典类型。一切类的创建最终都会调用type.new(cls, classname, bases, attrs)，它会在堆中创建一个类对象，并返回该对象。

当通过type.new(cls, classname, bases, attrs)创建类时，cls就是该类的元类，它是type类或其子类。

ABCMeta是一个抽象类的元类，用来创建抽象类基类：Metaclass for defining Abstract Base Classes (ABCs).
six.with_metaclass是用元类来创建类的方法，调用一个内部类，在内部类的new函数中，返回一个新创建的临时类。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def with_metaclass(meta, *bases):
    """Create a base class with a metaclass."""
    # This requires a bit of explanation: the basic idea is to make a dummy
    # metaclass for one level of class instantiation that replaces itself with
    # the actual metaclass.
    class metaclass(type):
        def __new__(cls, name, this_bases, d):
            return meta(name, bases, d)
        @classmethod
        def __prepare__(cls, name, this_bases):
            return meta.__prepare__(name, bases)
    return type.__new__(metaclass, ‘temporary_class‘, (), {})

six.with_metaclass(ABCMeta, object)就通过内部类的new函数返回一个ABCMeta元类创建的临时类，作为PeopleBase类的基类。

在python里类也是对象，下面两个语句可以看PeopleBase类的类型，都是，即PeopleBase类是ABCMeta元类的对象，是由ABCMeta元类生成的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
class PeopleBase(six.with_metaclass(ABCMeta, object)):
    @abstractmethod
    def work(self, *args, **kwargs):
        #pass为空语句，占位用
        pass
    @abstractmethod
    def live(self, *args, **kwargs):
        pass
print(type(PeopleBase))
print(PeopleBase.__class__)
#<class ‘abc.ABCMeta‘>
#<class ‘abc.ABCMeta‘>

详情请查看：

python中MRO排序原理

拓扑排序

如上图所示每个点都是有指向性的：可能指向别人或者被别人指向。

拓扑顺序就是：每次找到一个只指向别人的点 (学术性说法：入度为0)，记录下来；然后忽略掉这个点和它所指出去的线，再找到下一个只指向别人的点，记录下来，直到剩最后一个点，所有记录的点的顺序就是拓扑顺序

上图中，只有点1只指向别人，输出1；去掉点1和它伸出的两根线外只有点2只指向别人，输出2；…类推下去，得到拓扑排序结构: 1 2 4 3 5

MRO 排序算法

MRO 排序应用了 C3 算法，想了解 C3 自己查吧…总之得到的结果类似于拓扑排序，下面有段简单的多继承代码和其对应的拓扑排序的抽象图

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
class D(object):
    pass
class E(object):
    pass
 
class F(object):
    pass
 
class C(D, F):
    pass
 
class B(E, D):
    pass
 
class A(B, C):
    pass
 
if __name__ == '__main__':
    print A.__mro__

得到的输出结果：

1
(<class '__main__.A'>, <class '__main__.B'>, <class '__main__.E'>, <class '__main__.C'>, <class '__main__.D'>, <class '__main__.F'>, <type 'object'>)

下面就是抽象出来的图

我们就用拓扑排序来分析，但是这里会碰到同时存在好几个点都是入度为0 (说人话，就是没有被别人指向的点)，这时按照树的排序来，即从左到右，从根到叶，这里 A 就是根。
所以具体情况就是：我们先找到了点 A只有它没有被别人指向，输出A；去掉A及其伸出的两条线，剩 B 和 C 点同时满足只指向别人，按照树的顺序从左到右，故先输出B；去掉线剩 E 和 C ，输出E ;去线剩 C，输出C；去线剩 D 和 F ，输出D；去线只剩F ，输出F；最后输出object;得到的输出结果：

1
A B E C D F object

原文地址：
https://www.jianshu.com/p/6651ed35104c

发表于 | 分类于 ， | | 阅读次数:
| 字数统计:

pycrypto 在Windows系统上安装比较麻烦，会出现意想不到的各种问题。因为pycrypto-2.6.1 之后就不在维护。因此系统兼容性就不那么友好。推荐PyCryptodome作为替代工具包。官方地址为：https://www.pycryptodome.org/en/latest/

1
2
# 1、Windows pip install pycryptodomex
# 2、Linux/Mac pip install pycryptodome

举个例子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
#!usr/bin/env python3
# -*- coding:utf-8 _*-
import base64
import hashlib
import json
import platform
if platform.system() != 'Windows':
    from Crypto.Cipher import AES
    from Crypto import Random
else:
    from Cryptodome.Cipher import AES
    from Cryptodome import Random
    
class AESCipher(object):
    def __init__(self, key):
        self.bs = 32
        self.key = hashlib.sha256(key.encode()).digest()
    def encrypt(self, raw):
        raw = self._pad(raw).encode('utf-8')
        iv = Random.new().read(AES.block_size)
        cipher = AES.new(self.key, AES.MODE_CBC, iv)
        return base64.b64encode(iv + cipher.encrypt(raw))
    def _pad(self, s):
        return s + (self.bs - len(s) % self.bs) * chr(self.bs - len(s) % self.bs)
def aes_encrypt(aes_key, app_client_id, app_client_secret):
    aes = AESCipher(aes_key)
    raw = dict(user=app_client_id, password=app_client_secret)
    data = aes.encrypt(json.dumps(raw))
    return bytes.decode(data)

发表于 | 分类于 | | 阅读次数:
| 字数统计:

简介

Flask-APScheduler是Flask扩展，增加了对APScheduler的支持。

  • 支持从Flask配置加载调度程序配置以及任务定义。
  • 支持WEB程序和调度程序共存。提供REST API管理任务调度。并为REST API提供身份验证。目前只支持Basic-Auth认证。

官方地址：https://github.com/viniciuschiele/flask-apscheduler

使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
from flask import Flask
from flask_apscheduler import APScheduler
class Config(object):
    JOBS = [
        {
            'id': 'job1',
            'func': 'jobs:job1',
            'args': (1, 2),
            'trigger': 'interval',
            'seconds': 10
        }
    ]
    SCHEDULER_API_ENABLED = True
def job1(a, b):
    print(str(a) + ' ' + str(b))
if __name__ == '__main__':
    app = Flask(__name__)
    app.config.from_object(Config())
    scheduler = APScheduler()
    # it is also possible to enable the API directly
    # scheduler.api_enabled = True
    scheduler.init_app(app)
    scheduler.start()
    app.run()

Flask Config 属性配置

1
2
3
4
5
6
7
8
9
SCHEDULER_JOBSTORES         # 配置存储任务存储器
SCHEDULER_EXECUTORS         # 配置执行器
SCHEDULER_JOB_DEFAULTS      # 配置调度默认属性
SCHEDULER_TIMEZONE          # 配置时区
SCHEDULER_AUTH              # 配置认证中心
SCHEDULER_API_ENABLED       # 配置是否开启API
SCHEDULER_API_PREFIX        # 配置API路由前缀
SCHEDULER_ENDPOINT_PREFIX   # 配置API路由后缀
SCHEDULER_ALLOWED_HOSTS     # 配置访问白名单

REST API 接口

1
2
3
4
5
6
7
8
9
10
11
12
"""
Add the routes for the scheduler API.
"""
self._add_url_route('get_scheduler_info', '', api.get_scheduler_info, 'GET')
self._add_url_route('add_job', '/jobs', api.add_job, 'POST')
self._add_url_route('get_job', '/jobs/<job_id>', api.get_job, 'GET')
self._add_url_route('get_jobs', '/jobs', api.get_jobs, 'GET')
self._add_url_route('delete_job', '/jobs/<job_id>', api.delete_job, 'DELETE')
self._add_url_route('update_job', '/jobs/<job_id>', api.update_job, 'PATCH')
self._add_url_route('pause_job', '/jobs/<job_id>/pause', api.pause_job, 'POST')
self._add_url_route('resume_job', '/jobs/<job_id>/resume', api.resume_job, 'POST')
self._add_url_route('run_job', '/jobs/<job_id>/run', api.run_job, 'POST')

访问地址为：http://localhost:5000/scheduler/jobs

如果配置了 SCHEDULER_API_PREFIX = "/api/v1/scheduler"，则访问地址为: http://localhost:5000/api/v1/scheduler/jobs

配置API认证

1
2
3
4
5
SCHEDULER_AUTH = HTTPBasicAuth()
@scheduler.authenticate
def authenticate(auth):
    return auth['username'] == 'guest' and auth['password'] == 'guest'

完整示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
from flask import Flask
from flask_apscheduler import APScheduler
from flask_apscheduler.auth import HTTPBasicAuth
class Config(object):
    JOBS = [
        {
            'id': 'job1',
            'func': '__main__:job1',
            'args': (1, 2),
            'trigger': 'interval',
            'seconds': 10
        }
    ]
    SCHEDULER_API_ENABLED = True
    SCHEDULER_AUTH = HTTPBasicAuth()
def job1(a, b):
    print(str(a) + ' ' + str(b))
if __name__ == '__main__':
    app = Flask(__name__)
    app.config.from_object(Config())
    scheduler = APScheduler()
    # it is also possible to set the authentication directly
    # scheduler.auth = HTTPBasicAuth()
    scheduler.init_app(app)
    scheduler.start()
    @scheduler.authenticate
    def authenticate(auth):
        return auth['username'] == 'guest' and auth['password'] == 'guest'
    app.run()

使用过程遇到的问题

Debug启动过程中重复运行的问题

1
2
3
4
5
6
7
8
# Bug: APScheduler in Flask executes twice
# https://github.com/viniciuschiele/flask-apscheduler/issues/58
app.run(debug=True, use_reloader=False)
# or 
manager.add_command("runserver", Server(use_debugger=False, use_reloader=False))

多进程中APScheduler重复运行的问题

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
# Fix Multiple instances of scheduler problem
# https://github.com/viniciuschiele/flask-apscheduler/issues/51
import platform
import atexit
if platform.system() != 'Windows':
    fcntl = __import__("fcntl")
    f = open("scheduler.lock", "wb")
    try:
        fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
        scheduler.start()
        logger.debug("Scheduler Started...")
    except Exception as e:
        logger.error('Exit the scheduler, error - {}'.format(e))
        scheduler.shutdown()
    def unlock():
        fcntl.flock(f, fcntl.LOCK_UN)
        f.close()
    atexit.register(unlock)
else:
    msvcrt = __import__("msvcrt")
    f = open("scheduler.lock", "wb")
    try:
        msvcrt.locking(f.fileno(), msvcrt.LK_NBLCK, 1)
        scheduler.start()
        logger.debug("Scheduler Started...")
    except Exception as e:
        logger.error('Exit the scheduler, error - {}'.format(e))
        scheduler.shutdown()
    def _unlock_file():
        try:
            f.seek(0)
            msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1)
        except IOError:
            raise
    atexit.register(_unlock_file)

扩展

Flask-APScheduler缺少可视化任务调度和监控界面，需要使用者自己根据REST API扩展实现。

参考

发表于 | 分类于 | | 阅读次数:
| 字数统计:

简介

APScheduler是类似Quartz的一个Python定时任务框架，实现了Quartz的所有功能，使用起来十分方便。提供了基于日期、固定时间间隔以及crontab类型的任务，并且可以持久化任务。如果将任务持久化到数据库中，那么它们也将在调度程序在重新启动后继续运行并保持其状态。重启启动调度程序后，它将运行它在脱机时运行的所有任务。

除此之外，APScheduler还可以用作特定于平台的调度程序（如cron守护程序或Windows任务调度程序）的跨平台，特定于应用程序的替代程序。但请注意，APScheduler本身不是守护程序或服务，也不附带任何命令行工具。它主要用于在现有应用程序中运行。也就是说，APScheduler确实为我们提供了一些构建块来构建调度程序服务或运行专用的调度程序进程。

官方地址：https://github.com/agronholm/apscheduler

使用

简单例子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
#!usr/bin/env python3
# -*- coding:utf-8 _*-
from datetime import datetime
import time
import os
from apscheduler.schedulers.background import BackgroundScheduler
def tick():
    print('Tick! The time is: %s' % datetime.now())
if __name__ == '__main__':
    # 1、创建后台执行的 schedulers
    scheduler = BackgroundScheduler()
    # 2、添加调度任务，触发器选择 interval(间隔)，每隔3秒执行一次
    scheduler.add_job(tick, 'interval', seconds=3)
    # 3、启动调度任务
    scheduler.start()
    print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))
    try:
        # This is here to simulate application activity (which keeps the main thread alive).
        while True:
            time.sleep(2)
    except (KeyboardInterrupt, SystemExit):
        # Not strictly necessary if daemonic mode is enabled but should be done if possible
        scheduler.shutdown()

jitter: 振动参数，给每次触发添加一个随机浮动秒数，一般适用于多服务器，避免同时运行造成服务拥堵。

1
2
# 每小时（上下浮动120秒区间内）运行`tick`
sched.add_job(tick, 'interval', hours=1, jitter=120)

自定义配置

方法一

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
#!usr/bin/env python3
# -*- coding:utf-8 _*-
from pytz import utc
from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.mongodb import MongoDBJobStore
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
def tick():
    print('Tick! The time is: %s' % datetime.now())
# 选择MongoDB作为任务存储数据库
jobstores = {
    'mongo': MongoDBJobStore(),
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
# 默认使用线程池
executors = {
    'default': ThreadPoolExecutor(20),
    'processpool': ProcessPoolExecutor(5)
}
# 默认参数配置
job_defaults = {
    'coalesce': False,  # 积攒的任务是否只跑一次，是否合并所有错过的Job
    'max_instances': 3,  # 默认同一时刻只能有一个实例运行，通过max_instances=3修改为3个。
    'misfire_grace_time': 30  # 30秒的任务超时容错
}
scheduler = BackgroundScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc)
scheduler.add_job(tick, 'interval', seconds=3)
scheduler.start()

方法二

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from apscheduler.schedulers.background import BackgroundScheduler
# The "apscheduler." prefix is hard coded
scheduler = BackgroundScheduler({
    'apscheduler.jobstores.mongo': {
         'type': 'mongodb'
    },
    'apscheduler.jobstores.default': {
        'type': 'sqlalchemy',
        'url': 'sqlite:///jobs.sqlite'
    },
    'apscheduler.executors.default': {
        'class': 'apscheduler.executors.pool:ThreadPoolExecutor',
        'max_workers': '20'
    },
    'apscheduler.executors.processpool': {
        'type': 'processpool',
        'max_workers': '5'
    },
    'apscheduler.job_defaults.coalesce': 'false',
    'apscheduler.job_defaults.max_instances': '3',
    'apscheduler.timezone': 'UTC',
})

方法三

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from pytz import utc
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ProcessPoolExecutor
jobstores = {
    'mongo': {'type': 'mongodb'},
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
executors = {
    'default': {'type': 'threadpool', 'max_workers': 20},
    'processpool': ProcessPoolExecutor(max_workers=5)
}
job_defaults = {
    'coalesce': False,
    'max_instances': 3
}
scheduler = BackgroundScheduler()
# ..这里可以添加任务
scheduler.configure(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc)

misfire_grace_time：如果一个job本来14:00有一次执行，但是由于某种原因没有被调度上，现在14:01了，这个14:00的运行实例被提交时，会检查它预订运行的时间和当下时间的差值（这里是1分钟），大于我们设置的30秒限制，那么这个运行实例不会被执行。
合并：最常见的情形是scheduler被shutdown后重启，某个任务会积攒了好几次没执行如5次，下次这个job被submit给executor时，执行5次。将coalesce=True后，只会执行一次

replace_existing: 如果在程序初始化时，是从数据库读取任务的，那么必须为每个任务定义一个明确的ID，并且使用replace_existing=True，否则每次重启程序，你都会得到一份新的任务拷贝，也就意味着任务的状态不会保存。

调度器的操作

1
2
3
4
5
6
7
8
9
10
11
scheduler.add_job()         
scheduler.modify_job()      
scheduler.remove_job()
scheduler.pause_job()
scheduler.resume_job()
scheduler.reschedule_job()
scheduler.start()
scheduler.shutdown()
scheduler.get_jobs()
scheduler.remove_all_jobs()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
添加job，所需要的属性
add_job(func, trigger=None, args=None, kwargs=None, id=None, name=None,
                misfire_grace_time=undefined, coalesce=undefined, max_instances=undefined,
                next_run_time=undefined, jobstore='default', executor='default',
                replace_existing=False, **trigger_args):
job_kwargs = {
    'trigger': self._create_trigger(trigger, trigger_args),
    'executor': executor,
    'func': func,
    'args': tuple(args) if args is not None else (),
    'kwargs': dict(kwargs) if kwargs is not None else {},
    'id': id,
    'name': name,
    'misfire_grace_time': misfire_grace_time,
    'coalesce': coalesce,
    'max_instances': max_instances,
    'next_run_time': next_run_time
}
job_kwargs = dict((key, value) for key, value in six.iteritems(job_kwargs) if
                  value is not undefined)
job = Job(self, **job_kwargs)

调度监听

1
2
3
4
5
6
7
def my_listener(event):
    if event.exception:
        print('The job() crashed :('.format(event.job_id))  # or logger.fatal('The job crashed :(')
    else:
        print('The job() worked :)'.format(event.job_id))
scheduler.add_listener(my_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)

夏令时问题

有些timezone时区可能会有夏令时的问题。这个可能导致令时切换时，任务不执行或任务执行两次。避免这个问题，可以使用UTC时间，或提前预知并规划好执行的问题。

1
2
# 在Europe/Helsinki时区, 在三月最后一个周一就不会触发；在十月最后一个周一会触发两次
sched.add_job(job_function, 'cron', hour=3, minute=30)

参考

发表于 | 分类于 | | 阅读次数:
| 字数统计:

一、Serverless简介

Serverless是一种无服务的架构，如AWS Lambda。Serverless与传统架构不同，由开发者实现的服务端逻辑运行在无状态的计算容器中，它是由事件触发，短暂的(可能只存在于一次请求过程中)，完全被第三方管理。

以亚马逊的AWS Lambda为案例，Lambda能让不用思考任何服务器，也就是说，不用你处理服务器上的部署、服务器容量和服务器的扩展和失败容错，还有服务器上选择什么OS操作系统，语言的更新，日志等等问题。你的应用程序只需要和多个第三方的API或服务打交道，也可以自我创建一个无服务器的API。

二、Amazon API Gateway

Amazon API Gateway 是一种完全托管的服务，可以帮助开发者轻松创建、发布、维护、监控和保护任意规模的 API。只需在 AWS 管理控制台中点击几下，您便可以创建可充当应用程序“前门”的 API，从后端服务访问数据、业务逻辑或功能，例如基于 Amazon Elastic Compute Cloud (Amazon EC2) 运行的工作负载、基于 AWS Lambda 运行的代码或任意 Web 应用。

Amazon API Gateway 负责管理所有任务，涉及接受和处理成千上万个并发 API 调用，包括流量管理、授权和访问控制、监控，以及 API 版本管理。Amazon API Gateway 没有最低费用和启动成本。您只需为收到的 API 调用以及传出的数据量付费。

三、使用NodeJs创建Serverless API

aws-serverless-express —— 基于Express的Serverless中间件

https://github.com/fingerchou/aws-serverless-express

Run serverless applications and REST APIs using your existing Node.js application framework, on top of AWS Lambda and Amazon API Gateway. The sample provided allows you to easily build serverless web applications/services and RESTful APIs using the Express framework.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
// lambda.js
'use strict';
const awsServerlessExpress = require('aws-serverless-express');
const app = require('./app');
// NOTE: If you get ERR_CONTENT_DECODING_FAILED in your browser, this is likely
// due to a compressed response (e.g. gzip) which has not been handled correctly
// by aws-serverless-express and/or API Gateway. Add the necessary MIME types to
// binaryMimeTypes below, then redeploy (`npm run package-deploy`)
const binaryMimeTypes = [
  'application/javascript',
  'application/json',
  'application/octet-stream',
  'application/xml',
  'font/eot',
  'font/opentype',
  'font/otf',
  'image/jpeg',
  'image/png',
  'image/svg+xml',
  'text/comma-separated-values',
  'text/css',
  'text/html',
  'text/javascript',
  'text/plain',
  'text/text',
  'text/xml'
];
const server = awsServerlessExpress.createServer(app, null, binaryMimeTypes);
exports.handler = (event, context) => awsServerlessExpress.proxy(server, event, context);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
// app.js
'use strict';
const express = require('express');
const bodyParser = require('body-parser');
const cors = require('cors');
const compression = require('compression');
const awsServerlessExpressMiddleware = require('aws-serverless-express/middleware'); 
const app = express();
const router = express.Router();
// gzip compression
app.use(compression());
// enable CORS - Cross Origin Resource Sharing
app.use(cors());
app.use(bodyParser.json());
app.use(bodyParser.urlencoded({ extended: true }));
app.use(awsServerlessExpressMiddleware.eventContext());
router.get('/hello', (req, res) => {
    res.json({"say": "Hello World"})
});
router.get('/users', (req, res) => {
    res.json(users);
});
router.get('/users/:userId', (req, res) => {
    const user = getUser(req.params.userId);
    if(!user) {
        return res.status(404).json({});
    }
    return res.json(user);
});
router.post('/users', (req, res) => {
    const user = {
        id: ++userIdCounter,
        name: req.body.name
    };
    users.push(user);
    res.status(201).json(user);
});
router.put('/users/:userId', (req, res) => {
    const user = getUser(req.params.userId);
    console.log(req.params.userId, user);
    if (!user) {
        return res.status(404).json({});
    }
    user.name = req.body.name;
    res.json(user);
});
router.delete('/users/:userId', (req, res) => {
    const userIndex = getUserIndex(req.params.userId);
    if (userIndex === -1) {
        return res.status(404).json({});
    }
    users.splice(userIndex, 1);
    res.json(users);
});
const users = [{
    id: 1,
    name: 'Joe'
}, {
    id: 2,
    name: 'Jame'
}];
let userIdCounter = users.length;
const getUser = (userId) => users.find(u => u.id === parseInt(userId));
const getUserIndex = (userId) => users.findIndex(u => u.id === parseInt(userId));
app.use('/', router);
module.exports = app;
1
2
3
4
5
6
7
// app.local.js
const app = require('./app');
const port = 5000;
app.listen(port);
console.log(`listening on http://localhost:${port}`);

Apex —— 发布程序到AWS Lambda

http://apex.run/

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// project.development.json
{
  "name": "finger_demo",
  "description": "this is finger demo",
  "memory": 128,
  "timeout": 300,
  "role": "xxx",
  "handler": "lambda.handler",
  "runtime": "nodejs8.10",
  "region": "us-east-1",
  "retainedVersions": 5,
  "hooks": {
    "build": "npm install"
  },
  "environment": {
  }
}
1
2
3
4
5
// 发布development环境
apex deploy -e development
// 使用development别名
apex alias development

四、创建AWS API-Gateway测试

新建API-Gateway，创建Proxy

proxy指向Lambda

发布API，别名为development

测试

发表于 | 分类于 | | 阅读次数:
| 字数统计:

原文地址：https://www.jiqizhixin.com/articles/2018-07-18-3

介绍

Pandas是基于Numpy构建的库，在数据处理方面可以把它理解为numpy加强版。不同于numpy的是，pandas拥有种数据结构：Series和DataFrame：

用代码生成一个简单的Series对象

1
2
3
4
5
6
7
8
9
>>> from pandas import Series, DataFrame
>>> import pandas as pd
>>> data = Series([1,2,3,4],index = ['a','b','c','d'])
>>> data
a    1
b    2
c    3
d    4
dtype: int64

Series是一种类似一维数组的数据结构，由一组数据和与之相关的index组成，这个结构一看似乎与dict字典差不多，我们知道字典是一种无序的数据结构，而pandas中的Series的数据结构不一样，它相当于定长有序的字典，并且它的index和value之间是独立的，两者的索引还是有区别的，Series的index是可变的，而dict字典的key值是不可变的。

用代码生成一个简单的DataFrame对象

1
2
3
4
5
6
7
>>> data = {'a':[1,2,3],'b':['we','you','they'],'c':['btc','eos','ae']}
>>> df = DataFrame(data)
>>> df
   a     b    c
0  1    we  btc
1  2   you  eos
2  3  they   ae

DataFrame这种数据结构我们可以把它看作是一张二维表，DataFrame长得跟我们平时使用的Excel表格差不多，DataFrame的横行称为columns，竖列和Series一样称为index，DataFrame每一列可以是不同类型的值集合，所以DataFrame你也可以把它视为不同数据类型同一index的Series集合。

Series

Series的两种生成方式
1
2
3
4
5
6
7
>>> data = Series([222,'btc',234,'eos'])
>>> data
0    222
1    btc
2    234
3    eos
dtype: object

虽然我们在生成的时候没有设置index值，但Series还是会自动帮我们生成index，这种方式生成的Series结构跟list列表差不多，可以把这种形式的Series理解为竖起来的list列表。

1
2
3
4
5
6
7
>>> data = Series([1,2,3,4],index = ['a','b','c','d'])
>>> data
a    1
b    2
c    3
d    4
dtype: int64

这种形式的Series可以理解为numpy的array外面披了一件index的马甲，所以array的相关操作，Series同样也是支持的。结构非常相似的dict字典同样也是可以转化为Series格式的：

1
2
3
4
5
6
7
>>> dic = {'a':1,'b':2,'c':'as'}
>>> dicSeries = Series(dic)
>>> dicSeries
a     1
b     2
c    as
dtype: object

查看Series的相关信息
1
2
3
4
5
6
7
>>> data.index
Index(['a', 'b', 'c', 'd'], dtype='object')
>>> data.values
array([1, 2, 3, 4], dtype=int64)
# in方法默认判断的是index值
>>> 'a' in data
True
Series的NaN生成
1
2
3
4
5
6
7
8
9
>>> index1 = [ 'a','b','c','d']
>>> dic = {'b':1,'c':1,'d':1}
>>> data2 = Series(dic,index=index1)
>>> data2
a    NaN
b    1.0
c    1.0
d    1.0
dtype: float64

从这里我们可以看出Series的生成依据的是index值，index‘a’在字典dic的key中并不存在，Series自然也找不到’a’的对应value值，这种情况下Pandas就会自动生成NaN(not a number)来填补缺失值，这里还有个有趣的现象，原本dtype是int类型，生成NaN后就变成了float类型了，因为NaN的官方定义就是float类型。

NaN的相关查询
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
>>> data2.isnull()
a     True
b    False
c    False
d    False
dtype: bool
>>> data2.notnull()
a    False
b     True
c     True
d     True
dtype: bool
# 嵌套查询NaN
>>> data2[data2.isnull()==True]
a   NaN
dtype: float64
# 统计非NaN个数
>>> data2.count()
3

切记切记，查询NaN值切记不要使用np.nan==np.nan这种形式来作为判断条件，结果永远是False，==是用作值判断的，而NaN并没有值，如果你不想使用上方的判断方法，你可以使用is作为判断方法，is是对象引用判断，np.nan is np.nan，结果就是你要的True。

Series的name属性
1
2
3
4
5
6
7
8
9
>>> data.index.name = 'abc'
>>> data.name = 'test'
>>> data
abc
a    1
b    2
c    3
d    4
Name: test, dtype: int64

Series对象本身及其索引index都有一个name属性，name属性主要发挥作用是在DataFrame中，当我们把一个Series对象放进DataFrame中，新的列将根据我们的name属性对该列进行命名，如果我们没有给Series命名，DataFrame则会自动帮我们命名为0。

DataFrame

1
2
3
4
5
6
7
>>> data = {'name': ['BTC', 'ETH', 'EOS'], 'price':[50000, 4000, 150]}
>>> data = DataFrame(data)
>>> data
  name  price
0  BTC  50000
1  ETH   4000
2  EOS    150

DataFrame的生成与Series差不多，你可以自己指定index，也可不指定，DataFrame会自动帮你补上。

查看DataFrame的相关信息
1
2
3
4
5
6
7
8
>>> data.index
RangeIndex(start=0, stop=3, step=1)
>>> data.values
array([['BTC', 50000],
       ['ETH', 4000],
       ['EOS', 150]], dtype=object)
>>> data.columns
Index(['name', 'price'], dtype='object')
DataFrame索引
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
>>> data.name
0    BTC
1    ETH
2    EOS
Name: name, dtype: object
>>> data['name']
0    BTC
1    ETH
2    EOS
Name: name, dtype: object
# loc['name']查询的是行标签
>>> data.iloc[1]
name      ETH
price    4000
Name: 1, dtype: object

其实行索引，除了iloc，loc还有个ix，ix既可以进行行标签索引，也可以进行行号索引，但这也大大增加了它的不确定性，有时会出现一些奇怪的问题，所以pandas在0.20.0版本的时候就把ix给弃用了。

DataFrame的常用操作
简单地增加行、列
1
2
3
4
5
6
7
8
9
10
11
12
13
14
>>> data['type'] = 'token'
>>> data
  name  price   type
0  BTC  50000  token
1  ETH   4000  token
2  EOS    150  token
>>> data.loc['3'] = ['ae',200,'token']
>>> data
  name  price   type
0  BTC  50000  token
1  ETH   4000  token
2  EOS    150  token
3   ae    200  token
删除行、列操作
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
>>> del data['type']
>>> data
  name  price
0  BTC  50000
1  ETH   4000
2  EOS    150
3   ae    200
>>> data.drop([2])
  name  price
0  BTC  50000
1  ETH   4000
3   ae    200
>>> data
  name  price
0  BTC  50000
1  ETH   4000
2  EOS    150
3   ae    200

这里需要注意的是，使用drop（）方法返回的是Copy而不是视图，要想真正在原数据里删除行，就要设置inplace=True

1
2
3
4
5
6
>>> data.drop([2],inplace=True)
>>> data
  name  price
0  BTC  50000
1  ETH   4000
3   ae    200

设置某一列为index
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
>>> data.set_index(['name'],inplace=True)
>>> data
      price
name
BTC   50000
ETH    4000
ae      200
# 将index返回回dataframe中
>>> data.reset_index(inplace=True)
>>> data
  name  price
0  BTC  50000
1  ETH   4000
2   ae    200
处理缺失值
1
2
data.dropna()    # 丢弃含有缺失值的行
data.fillna(0)    # 填充缺失值数据为0

还是需要注意：这些方法返回的是copy而不是视图，如果想在原数据上改变，别忘了inplace=True。

数据合并
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
>>> data
  name  price
0  BTC  50000
1  ETH   4000
2   ae    200
3  eos    eos
>>> data1
  name other
0  BTC     a
1  ETH     b
# 以name为key进行左连接
>>> pd.merge(data,data1,on='name',how='left')
  name  price other
0  BTC  50000     a
1  ETH   4000     b
2   ae    200   NaN
3  eos    eos   NaN

平时进行数据合并操作，更多的会出一种情况，那就是出现重复值，DataFrame也为我们提供了简便的方法：

1
data.drop_duplicates(inplace=True)
数据的简单保存与读取
1
2
3
4
5
6
7
>>> data.to_csv('test.csv')
>>> pd.read_csv('test.csv')
   Unnamed: 0 name  price
0           0  BTC  50000
1           1  ETH   4000
2           2   ae    200
3           3  eos    eos

为什么会出现这种情况呢，从头看到尾的同学可能就看出来了，增加第三行时，我用的是loc[‘3’]行标签来增加的，而read_csv方法是默认index是从0开始增长的，此时只需要我们设置下index参数就ok了

1
2
3
4
5
6
7
8
# 不保存行索引
>>> data.to_csv('test.csv',index=None)
>>> pd.read_csv('test.csv')
  name  price
0  BTC  50000
1  ETH   4000
2   ae    200
3  eos    eos

其他的还有header参数，这些参数都是我们在保存数据时需要注意的。

发表于 | 分类于 | | 阅读次数:
| 字数统计:

原文地址：https://www.jiqizhixin.com/articles/2018-07-18-2

介绍

numpy是专门为科学计算设计的一个python扩展包，为python提供高效率的多维数组，也被称为面向阵列计算（array oriented computing），同时numpy也是github上的一个开源项目：numpy，numpy是基于c语言开发，所以这使得numpy的运行速度很快，高效率运行就是numpy的一大优势。

用numpy生成一维数组

1
2
3
4
>>> import numpy as np
>>> a = np.array([[1,2,3]], dtype=np.int32)
>>> a
array([[1, 2, 3]])

numpy中定义的最重要的数据结构是称为ndarray的n维数组类型，这个结构引用了两个对象，一块用于保存数据的存储区域和一个用于描述元素类型的dtype对象：

性能

二维数组的生成在python中我们还可以用到list列表，如果用list来表示[1,2,3]，由于list中的元素可以是任何对象，所以list中保存的是对象的指针，如果要保存[1,2,3]就需要三个指针和三个整数对象，是比较浪费内存资源和cpu计算时间的，而ndarray是一种保存单一数据类型的多维数组结构，在数据处理上比list列表要快上很多。

使用

生成一个3x3的矩阵
1
2
3
4
5
6
7
8
9
10
11
12
13
>>> import numpy as np
>>> data = np.array([[1,2,3],[4,5,6],[7,8,9]])
>>> data
array([[1, 2, 3],
       [4, 5, 6],
       [7, 8, 9]])
# 等价于data=np.arange(1，10).reshape(3,3)
>>> data = np.arange(1,10).reshape(3,3)
>>> data
array([[1, 2, 3],
       [4, 5, 6],
       [7, 8, 9]])
查看矩阵信息
1
2
3
4
5
6
7
8
9
10
11
# 返回元组，表示n行n列
>>> data.shape
(3, 3)
# 返回数组数据类型
>>> data.dtype
dtype('int32')
# 返回是几维数组
>>> data.ndim
2
转换数据类型
1
2
3
4
5
6
7
8
# 拷贝一份新的数组
>>> a = data.astype(float)
>>> a
array([[1., 2., 3.],
       [4., 5., 6.],
       [7., 8., 9.]])
>>> a.dtype
dtype('float64')
数组之间的计算
1
2
3
4
5
6
7
8
>>> data + data
array([[ 2,  4,  6],
       [ 8, 10, 12],
       [14, 16, 18]])
>>> data * data
array([[ 1,  4,  9],
       [16, 25, 36],
       [49, 64, 81]])

可以看出相同规格的数组计算是直接作用在其元素级上的，那不同的规格的数组是否能进行运算呢，我们来看下这个例子：

1
2
3
4
5
6
# 生成一个2x2numpy数组
>>> data1 = np.array([[1,2],[1,2]])
>>> data + data1
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
ValueError: operands could not be broadcast together with shapes (3,3) (2,2)

我们可以看出不同规格的数组一起计算的话是会报出广播错误的，那是不是可以下结论了，别急我们再来看下方两个特殊例子：

1
2
3
4
5
6
7
8
9
10
>>> data2 = np.array([[1,2,3]])
>>> data + data2
array([[ 2,  4,  6],
       [ 5,  7,  9],
       [ 8, 10, 12]])
>>> data3 = np.array([[1], [2], [3]])
>>> data + data3
array([[ 2,  3,  4],
       [ 6,  7,  8],
       [10, 11, 12]])

data2数组的列数量与data数组相等，data3数组的行数量与data数组相等，这两个numpy数组虽然规格与data数组不一样，但却依然可以与data数组进行运算。

数据的切片
1
2
3
4
5
6
7
8
9
10
11
12
13
# 沿着行(axis=0)进行索引
>>> data[:2]
array([[1, 2, 3],
       [4, 5, 6]])
# 先沿着行(axis=0)进行索引，再沿着列(axis=1)进行索引
>>> data[:2, :2]
array([[1, 2],
       [4, 5]])
# 下标是从0开始
>>> data[1,0:2]
array([4, 5])

这里需要注意的是，切片操作是在原始数组上创建一个视图view，这只是访问数组数据的一种方式。 因此原始数组不会被复制到内存中，传递的是一个类似引用的东西，与上面的astype（）方法是两种不同的拷贝方式，这里我们来看一个例子：

1
2
3
4
5
6
7
8
>>> a = data[1]
>>> a
array([4, 5, 6])
>>> a[:] = 0
>>> data
array([[1, 2, 3],
       [0, 0, 0],
       [7, 8, 9]])

当切片对象a改变时，data的对应值也会跟着改变，这是在我们日常数据处理中有时会疏忽的一个点，最安全的复制方法是使用copy（）方法进行浅拷贝：

1
2
3
4
5
6
7
8
>>> a = data[1].copy()
>>> a
array([0, 0, 0])
>>> a[:] = 9
>>> data
array([[1, 2, 3],
       [0, 0, 0],
       [7, 8, 9]])

数组的布尔索引
1
2
3
4
5
6
7
8
9
10
11
>>> data
array([[1, 2, 3],
       [4, 5, 6],
       [7, 8, 9]])
>>> data > 3
array([[False, False, False],
       [ True,  True,  True],
       [ True,  True,  True]])
# 找出大于3的元素
>>> data[data > 3]
array([4, 5, 6, 7, 8, 9])
数组的逻辑表达处理
1
2
3
4
5
# 大于3的标记为1，小于等于3的标记为0
>>> np.where(data>3,1,0)
array([[0, 0, 0],
       [1, 1, 1],
       [1, 1, 1]])
数组的常用统计操作
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
# 沿着行(axis=0)进行索引，求出其平均值
>>> data.mean(axis=0)
array([4., 5., 6.])
# 求出全部元素的方差
>>> data.std()
2.581988897471611
# 统计数组中元素大于3的个数
>>> (data > 3).sum()
6
# 数组中是否存在一个或多个true
>>> data.any()
True
# 数组中是否全部数都是true
>>> data.all()
True
# 沿着行(axis=0)进行索引，进行累加
>>> data.cumsum(0)
array([[ 1,  2,  3],
       [ 5,  7,  9],
       [12, 15, 18]], dtype=int32)
# 沿着列(axis=1)进行索引，进行累乘
>>> data.cumprod(1)
array([[  1,   2,   6],
       [  4,  20, 120],
       [  7,  56, 504]], dtype=int32)
数组的排序操作
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
>>> data = np.random.randn(4,4)
>>> data
array([[-2.09150248, -0.07758111, -1.30427684, -0.79387749],
       [-0.10272387, -0.89728241, -0.54607326,  0.87435473],
       [ 1.83972061, -0.81256093, -1.21652558,  0.51171488],
       [-0.40912271,  0.21529186,  0.66593734, -0.54275447]])
# 沿着行(axis=0)进行索引，并进行升序排序
>>> data.sort(0)
>>> data
array([[-2.09150248, -0.89728241, -1.30427684, -0.79387749],
       [-0.40912271, -0.81256093, -1.21652558, -0.54275447],
       [-0.10272387, -0.07758111, -0.54607326,  0.51171488],
       [ 1.83972061,  0.21529186,  0.66593734,  0.87435473]])
# 降序操作
>>> data[::-1]
array([[ 1.83972061,  0.21529186,  0.66593734,  0.87435473],
       [-0.10272387, -0.07758111, -0.54607326,  0.51171488],
       [-0.40912271, -0.81256093, -1.21652558, -0.54275447],
       [-2.09150248, -0.89728241, -1.30427684, -0.79387749]])

注意：直接调用数组的方法的排序将直接改变数组而不会产生新的拷贝。

矩阵运算
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
>>> x=np.arange(9).reshape(3,3)
>>> x
array([[0, 1, 2],
       [3, 4, 5],
       [6, 7, 8]])
# 矩阵相乘
>>> np.dot(x, x)
array([[ 15,  18,  21],
       [ 42,  54,  66],
       [ 69,  90, 111]])
# 矩阵转置
>>> x.T
array([[0, 3, 6],
       [1, 4, 7],
       [2, 5, 8]])

发表于 | 分类于 | | 阅读次数:
| 字数统计:

问：假设 S由a, b, c三个数值组成, S的大小为2个字节(Byte)。已知S的值。求解a, b, c分别是多少？
其中a = 4 Bit, b = 4 Bit, c = 8 Bit。(a >= 0, b >= 0, c >= 0)

解：

如图所示：

  • a 为前4 Bit，则a = S >> 12
  • b 为a后4 Bit，则b = S >> 8 & 0b1111
  • c 为最后8 Bit, 则c = S & 0b11111111

假定S = 10086，则：

1
2
3
4
5
6
>>> S = 10086
>>> a = S >> 12
>>> b = S >> 8 & 0b1111
>>> c = S & 0b11111111
>>> print(a, b, c)
2 7 102

验证一下：

1
2
3
4
5
6
>>> a = 2
>>> b = 7
>>> c = 102
>>> S = (a << 12) + (b << 8) + c
>>> S
10086

发表于 | 分类于 | | 阅读次数:
| 字数统计:

一、斐波那契数列定义

斐波那契数列（Fibonacci sequence），又称黄金分割数列、因数学家列昂纳多·斐波那契（Leonardoda Fibonacci）以兔子繁殖为例子而引入，故又称为“兔子数列”，指的是这样一个数列：1、1、2、3、5、8、13、21、34、……

数学表达式：
F(1)=1，F(2)=1, F(n)=F(n-1)+F(n-2)（n>=2，n∈N*）

1
fib = lambda n: n if n <= 2 else fib(n - 1) + fib(n - 2)

二、异或

运算规则：如果a、b两个值不相同，则异或结果为1。如果a、b两个值相同，异或结果为0。
特性：同一变量与另一变量和其异或值异或等于另一个数，如（a ^ b）^ b = a。

1
2
3
4
5
>>> a = 2
>>> b = 3
>>> c = a ^ b
>>> c ^ b
2

三、加密解密逻辑

  • 1、定义斐波那契数列，用来做密码表。
  • 2、将秘钥字符串与斐波拉契数列做关联，进行加密计算得到新的密码表。
  • 3、加密，需要加密的原始字符串与加密计算后的新密码表进行异或，得到加密之后的字符串。
  • 4、解密，把加密后的字符串与加密计算后的新密码表再进行异或就得到原始字符串

四、代码实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
# [1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144...]
_fib = [1] * 64
# 初始化斐波拉契数列，本例定义数列大小为64
def __init_fib():
    for i in range(2, 64):
        _fib[i] = _fib[i-1] + _fib[i-2]
__init_fib()
def encrypt(source, secret):
    if not source or not secret:
        return source
    # 将字符串转成对应字符的ascii码数值
    src_list = list(map(ord, source))
    key_list = list(map(ord, secret))
    key_len = len(key_list)
    # 32 <= N <= 64
    N = max(sum(key_list) & 0x3F, 32)
    # 将秘钥字符串与斐波拉契数列做加密计算，生成新的列表
    # 保证计算后的数值小于255
    tmp = [(key_list[_fib[i] % key_len] + _fib[i]) & 0xFF for i in range(N)]
    # 将原始字符串与新生成的列表做异或
    dst_list = [src_list[i] ^ tmp[i % N] for i in range(len(src_list))]
    # chr 将范围[0-255]之间的数字转换成ascii码字符
    return ''.join(list(map(chr, dst_list)))
# 对称加密，加密即是解密
decrypt = encrypt
if __name__ == '__main__':
    secret_key = 'have a nice day'
    dst_string = encrypt('FingerChou', secret_key)
    print(dst_string)
    print(decrypt(dst_string, secret_key))

五、综述

上述只是利用斐波拉契数列进行了简单的加密计算。如果知道数列也就知道密码了。
虽然我们还增加了秘钥字符串。增加了复杂性。但是实际应用中，还是要把生成密码表的算法设计的更复杂一点。