【编者的话】本文介绍了Serverless的演化过程，以及实施过程中可能存在的一些障碍，由此提出Funtainer（容器即函数）并简要阐述了其优势。

常言道：“Serverless并非没有服务器，而是不再需要关注服务器。”

在最近的Serverless会议上，我们看到了Serverless使用率的上升，并且这种上升令人印象深刻。

我很荣幸能够成为会议的一名主讲人。我所演讲的内容是开发者及DevOps世界中的剧烈变化，从容器管理和微服务部署到
函数和Serverless。

回首过去20年，上图向我们展示了演化历程。我们快速地从物理服务器管理迁移到虚拟机管理，而现在我们正在从容器运行转移到函数运行。

Docker的面世，使得Linux容器变得如此好用，以至于其迅速成为了主流。它们真正成为了帮助开发者构建、迁移和运行（应用）的有效工具。我们开始以一种舒服的方式开发微服务，并且认识到部署不再艰难。

2014年，AWS Lambda面世，这是一种允许用户运行代码而无需管理和关心服务器的服务。Lambda暗示着“遗忘容器，开始构建函数”。但如何联系两个概念是一个很大的问题。这两种技术变迁是否存在冲突，或者说能否以一种神奇的方式整合它们？

回答该问题前，让我们先从采用Serverless的障碍，容器的优点等方面开始讨论。

采用Serverless的障碍

1. 部署方法学的剧烈变迁

自古改变非易事。为了使用函数，我们需要从极小的构建开始，它甚至比微服务更小，本质上是微微服务（micro-microservice）。显然，我们将需要开发更多服务了。

2. 使用不同的接口

如果我们想要使用Spotinst Functions，Lambda或者Azure Functions，我们需要遵循特定的模板或签名。如下是我所说的一个例子：

Amazon的接口代码为白色，Google为蓝色，Azure为黄色。这意味着每个云对于代码的响应均有细微区别，使得形式的标准化变得更加困难。

3. 改变部署方式及其后的思想

我们已经习惯了VM、容器，蓝绿部署和健康检查或者按步就班地部署新版本的方式。函数需要我们处理一个些许不同的现实。再次，改变非易事。

4. 第三方依赖

在容器或者虚拟机上运行代码时，我们知道其所有的依赖将会在代码所运行的服务器上被找到。在Serverless中，我们需要将代码与其依赖一起打包到ZIP中，这样Serverless供应者才能够运行函数。此外，ZIP的大小被限制到50MB（包括依赖！），这使得在一个函数中运行复杂代码变得更加困难。

函数确实带来了一大堆优点，但它们被一些难以轻易踢开的绊脚石所阻碍。倘若我们能够将容器的优势应用到函数中，一切又会变得如何？

为了回答这个问题，让我们对AWS
Lambda背后的技术稍作深挖，一起来看看所谓的”Serverless计算”魔术是如何工作的？

其实一切都很简单：

1. 上传代码（包括依赖）
2. Function供应者（例如Lambda）拿到我们的代码，并将其注入到容器中
3. 容器在服务器上运行函数，Lambda返回结果

如果我们跳过第2步，让函数的输入即为容器，而函数供应商拿到容器后直接运行会怎样？

我们将这种“容器运行即函数”的处理称为“Funtainer”。这让我们能够同时享受运行容器和运行函数带来的好处，例如高可用性，无需运维，以及Serverless计算，所有的一切运行在近些年来我们所悉知的容器之上。

Funtainer，或者说容器即函数，正在以惊人的速度发展，3个大项目在这种大环境下获益匪浅：

1. IBM OpenWhisk
2. 我们自己的Spotinst Functions
3. OpenFaas

上述平台带来了难以置信的好处：

1. 不像其他FaaS产品，没有编程语言的限制
2. 无需担心依赖或者ZIP文件（所有的东西都被打包到容器）
3. 不再是供应商锁定，随时随地运行Funtainer
4. 无需关心不同的FaaS代码接口

所以，它是如何工作的？

1. 编写代码
2. 打包到容器
3. 编写Dockerfile，包含容器运行所需的依赖
4. 上传容器到Hub
5. 创建函数，将输入代码修改为DockerHub/ContainerName
6. 运行函数

换句话说，无论我们是否喜欢，剧变正在发生。真正的问题是，函数何时才会成为大多数公司运行代码的方式。

这种变化并不恐怖，不需要我们改变在容器上运行代码学到的方法学，转而学习新的（方法学）。运行容器即服务（Funtainer）让我们可以简单的搭上这波势头，享受容器和函数带来的优点。

原文链接：Funtainers: The Beauty of Running Containers as Functions（翻译：孙科）

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

　　实际项目中，pythoner更加关注的是Python的性能问题，之前也写过一篇文章《Python性能优化》介绍Python性能优化的一些方法。而本文，关注的是Python的内存优化，一般说来，如果不发生内存泄露，运行在服务端的Python代码不用太关心内存，但是如果运行在客户端（比如移动平台上），那还是有优化的必要。具体而言，本文主要针对的Cpython，而且不涉及C扩展。　　

　　我们知道，Python使用引用技术和垃圾回收来管理内存，底层也有各种类型的内存池，那我们怎么得知一段代码使用的内存情况呢？工欲善其事必先利其器，直接看windows下的任务管理器或者linux下的top肯定是不准的。

　　本文地址：www.cnblogs.com/xybaby/p/74…

Pytracemalloc

　　对于基本类型，可以通过sys.getsizeof()来查看对象占用的内存大小。以下是在64位Linux下的一些结果：

import sys

sys.getsizeof(1) 24 >>> sys.getsizeof([]) 72 >>> sys.getsizeof(()) 56 >>> sys.getsizeof({}) 280 >>> sys.getsizeof(True) 24

　　可以看到，即使是一个int类型(1)也需要占用24个字节，远远高于C语言中int的范围。因为Python中一切都是对象，int也不例外（事实上是PyIntObject），除了真正存储的数值，还需要保存引用计数信息、类型信息，更具体的可以参见《Python源码剖析》。

　　 而对于更复杂的组合类型，复杂的代码，使用getsizeof来查看就不准确了，因为在Python中变量仅仅指向一个对象，这个时候就需要更高级的工具，比如guppy，pysizer，pytracemalloc，
objgraph。在这里重点介绍pytracemalloc。

　　在Python3.4中，已经支持了pytracemalloc，如果使用python2.7版本，则需要对源码打补丁，然后重新编译。pytracemalloc在pep454中提出，主要有以下几个特点：

• Traceback where an object was allocated
• Statistics on allocated memory blocks per filename and per line number: total size, number and average size of allocated memory blocks
• Compute the differences between two snapshots to detect memory leaks

　　简单来说，pytracemalloc hook住了python申请和释放内存的接口，从而能够追踪对象的分配和回收情况。对内存分配的统计数据可以精确到每个文件、每一行代码，也可以按照调用栈做聚合分析。而且还支持快照（snapshot）功能，比较两个快照之间的差异可以发现潜在的内存泄露。

　　下面通过一个例子来简单介绍pytracemalloc的用法和接口，关于更详细用法和API，可以参考这份详尽的文档或者pytracemalloc的作者在pycon上的演讲ppt。

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

复制代码 1 import tracemalloc
 2 
 3 NUM_OF_ATTR =  10
 4 NUM_OF_INSTANCE = 100
 5  
 6 class Slots(object):
 7     __slots__ = ['attr%s'%i for i in range(NUM_OF_ATTR)]
 8     def __init__(self):
 9         value_lst = (1.0, True, [], {}, ())
10         for i in range(NUM_OF_ATTR):
11             setattr(self, 'attr%s'%i, value_lst[i % len(value_lst)])
12 
13 
14 class NoSlots(object):
15     def __init__(self):
16         value_lst = (1.0, True, [], {}, ())
17         for i in range(NUM_OF_ATTR):
18             setattr(self, 'attr%s'%i, value_lst[i % len(value_lst)])
19 
20 
21 
22 def generate_some_objs():
23     lst = []
24     for i in range(NUM_OF_INSTANCE):
25         o = Slots() if i % 2 else NoSlots()
26         lst.append(o)
27     return lst
28 
29 
30 if __name__ == '__main__':
31     tracemalloc.start(3)
32 
33     t = generate_some_objs() 
34 
35     snapshot = tracemalloc.take_snapshot()
36     top_stats = snapshot.statistics('lineno') # lineno filename traceback
37 
38     print(tracemalloc.get_traced_memory())
39     for stat in top_stats[:10]:
40         print(stat)

　　在上面的代码中，用到了pytracemalloc几个核心的API：

　　start(nframe: int=1)　　

　　　　pytracemalloc的一大好处就是可以随时启停，start函数即开始追踪内存分配，相应的stop会停止追踪。start函数有一个参数，nframes : 内存分配时记录的栈的深度，这个值越大，pytracemalloc本身消耗的内存越多，在计算cumulative数据的时候有用。

　　get_traced_memory()

　　　　返回值是拥有两个元素的tuple，第一个元素是当前分配的内存，第二个元素是自内存追踪启动以来的内存峰值。

　　take_snapshot()　　　　

　　　　返回当前内存分配快照，返回值是Snapshot对象，该对象可以按照单个文件、单行、单个调用栈统计内存分配情况　　　　

　　运行环境：windows 64位python3.4

(62280, 62920)

test_pytracemalloc_use_py3.4.py:10: size=16.8 KiB, count=144, average=120 B test_pytracemalloc_use_py3.4.py:17: size=16.7 KiB, count=142, average=120 B test_pytracemalloc_use_py3.4.py:19: size=9952 B, count=100, average=100 B
test_pytracemalloc_use_py3.4.py:26: size=9792 B, count=102, average=96 B test_pytracemalloc_use_py3.4.py:27: size=848 B, count=1, average=848 B test_pytracemalloc_use_py3.4.py:34: size=456 B, count=1, average=456 B
test_pytracemalloc_use_py3.4.py:36: size=448 B, count=1, average=448 B D:\Python3.4\lib\tracemalloc.py:474: size=64 B, count=1, average=64 B

　　如果将第36行的“lineno“改成“filename”，那么结果如下

(62136, 62764)

test_pytracemalloc_use_py3.4.py:0: size=54.5 KiB, count=491, average=114 B D:\Python3.4\lib\tracemalloc.py:0: size=64 B, count=1, average=64 B

　　有了Profile结果之后，可以看出来在哪个文件中有大量的内存分配。与性能优化相同，造成瓶颈的有两种情况：单个对象占用了大量的内存；同时大量存在的小对象。对于前者，优化的手段并不多，惰性初始化属性可能有一些帮助；而对于后者，当同样类型的对象大量存在时，可以使用slots进行优化。

Slots

　　默认情况下，自定义的对象都使用dict来存储属性（通过obj.__dict__查看），而python中的dict大小一般比实际存储的元素个数要大（以此降低hash冲突概率），因此会浪费一定的空间。在新式类中使用__slots__，就是告诉Python虚拟机，这种类型的对象只会用到这些属性，因此虚拟机预留足够的空间就行了，如果声明了__slots__，那么对象就不会再有__dict__属性。

　　使用slots到底能带来多少内存优化呢，首先看看这篇文章，对于一个只有三个属性的Image类，使用__slots__之后内存从25.5G下降到16.2G，节省了9G的空间！

　　)

　　到底能省多少，取决于类自身有多少属性、属性的类型，以及同时存在多少个类的实例。下面通过一段简单代码测试一下：

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

复制代码 1 # -*- coding: utf-8 -*-
 2 import sys
 3 import tracemalloc
 4 
 5 NUM_OF_ATTR =  3 #3 # 10 # 30 #90
 6 NUM_OF_INSTANCE = 10 # 10 # 100
 7  
 8 class Slots(object):
 9     __slots__ = ['attr%s'%i for i in range(NUM_OF_ATTR)]
10     def __init__(self):
11         value_lst = (1.0, True, [], {}, ())
12         for i in range(NUM_OF_ATTR):
13             setattr(self, 'attr%s'%i, value_lst[i % len(value_lst)])
14 
15 
16 class NoSlots(object):
17     def __init__(self):
18         value_lst = (1.0, True, [], {}, ())
19         for i in range(NUM_OF_ATTR):
20             setattr(self, 'attr%s'%i, value_lst[i % len(value_lst)])
21 
22 if __name__ == '__main__':
23     clz = Slots if len(sys.argv) > 1 else NoSlots
24     tracemalloc.start()
25     objs = [clz() for i in range(NUM_OF_INSTANCE)]
26     print(tracemalloc.get_traced_memory()[0])

　　上面的代码，主要是在每个实例的属性数目、并发存在的实例数目两个维度进行测试，并没有测试不同的属性类型。结果如下表：

　　百分比为内存优化百分比，计算公式为(b - a) / b， 其中b为没有使用__slots__时分配的内存， a为使用了__slots__时分配的内存。

注意事项

　　关于__slots__，Python文档有非常详尽的介绍，这里只强调几点注意事项

　　第一：基类和子类都必须__slots__，即使基类或者子类没有属性

class Base(object):

… pass … >>> class Derived(Base): … __slots__ = (‘a’, ) … >>> d.__slots__ (‘a’,) >>> getattr(d, ‘__dict__‘, ‘No Dict’)
{}

　　从上面的示例可以看到，子类的对象还是有__dict__属性，原因就在于基类没有声明__slots__。因此，可以通过看子类的实例有没有__dict__属性来判断slots的使用是否正确

　　第二：子类会继承基类的__slots__

　　更准确的说，如果访问属性的时候没有在子类的__slots__找到，会继续在基类的__slots__查找，因为Python使用descriptor在类这个层级实现__slots__的，具体可以参见《python属性查找 深入理解》一文

class Base(object):

… __slots__ = (‘a’,) … >>> class Derived(Base): … __slots__ = (‘b’, ) … >>> d = Derived()

d.__slots__ (‘b’,) >>> getattr(d, ‘__dict__‘, ‘No Dict’) ‘No Dict’ >>> d.a = 1 >>> d.c = 0 Traceback (most recent call last):
File ““, line 1, in AttributeError: ‘Derived’ object has no attribute ‘c’

objgraph

　　在大型工程中，怎么排查有哪些大量存在的对象呢，毕竟同一个类型存在的对象越多，优化越有效果。除了直接看代码，最好使的就是使用objgraph.py的show_most_common_types(N)函数，该函数返回Python gc管理的所有对象中，数目前N多的对象，在排除掉python builtin对象之后，剩下的就是可优化的对象。比如在最上面的代码中：在最后加上这么两句：

import objgraph

objgraph.show_most_common_types(25)

　　输出如下：
　　

再论Python dict

　　前面介绍slots的时候，就提到Python自定义的对象中通过dict来管理属性。这种机制极大的提高了Python的灵活性 – 可以随时给对象增加属性，但是其实现机制也带来了内存上的浪费。不管是python源码，还是Python程序，都大量使用了dict，因此这部分内存浪费不容小视。

　　python中的dict使用的是散列表（类似C++中的std::unordered_map），当计算出的hash值冲突的时候，采用开放地址法解决冲突（另一种常见的冲突解决算法是链表法）。为了降低冲突概率，当装填因子（实际存储的元素与散列表长度的比值）超过2/3的时候就会对散列表进行扩容，因此散列表中一定会存在一些未使用的槽。

　　下面简单看看PyDictObject的数据结构（python2.7.3 dictobject.h）

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

复制代码 1 #define PyDict_MINSIZE 8
 2 
 3 typedef struct {
 4     /* Cached hash code of me_key.  Note that hash codes are C longs.
 5      * We have to use Py_ssize_t instead because dict_popitem() abuses
 6      * me_hash to hold a search finger.
 7      */
 8     Py_ssize_t me_hash;
 9     PyObject *me_key;
10     PyObject *me_value;
11 } PyDictEntry;
12 
13 
14 typedef struct _dictobject PyDictObject;
15 struct _dictobject {
16     PyObject_HEAD
17     Py_ssize_t ma_fill;  /* # Active + # Dummy */
18     Py_ssize_t ma_used;  /* # Active */
19 
20     /* The table contains ma_mask + 1 slots, and that's a power of 2.
21      * We store the mask instead of the size because the mask is more
22      * frequently needed.
23      */
24     Py_ssize_t ma_mask;
25 
26     /* ma_table points to ma_smalltable for small tables, else to
27      * additional malloc'ed memory.  ma_table is never NULL!  This rule
28      * saves repeated runtime null-tests in the workhorse getitem and
29      * setitem calls.
30      */
31     PyDictEntry *ma_table;
32     PyDictEntry *(*ma_lookup)(PyDictObject *mp, PyObject *key, long hash);
33     PyDictEntry ma_smalltable[PyDict_MINSIZE];
34 };

　　从定义可以看出，除了固定的部分（几个Py_ssize_t），PyDictObject中主要是PyDictEntry对象，PyDictEntrty包含一个Py_ssize_t（int）和两个指针。上面源码中的注释（第26行）指出，当dict的元素比较少时，ma_table指向ma_smalltable，当元素增多时，ma_table会指向新申请的空间。ma_smalltable的作用在于Python（不管是源码还是代码）都大量使用dict，一般来说，存储的元素也不会太多，因此Python就先开辟好PyDict_MINSIZE(默认为8)个空间。

　　为什么说PyDictObject存在浪费呢，PyDictEntry在32位下也有12个字节，那么即使在ma_smalltable（ma_table）中大量的位置没有被使用时，也要占用这么多字节。用这篇文章中的例子：

　　假设有这么一个dict： 　　d = {‘timmy’: ‘red’, ‘barry’: ‘green’, ‘guido’: ‘blue’}

　　在Python源码中的视图就是这样的：

　下面的entries就是ma_smalltable

1 2 3 4 5 6 7 8 9 10

> 复制代码entries = [['--', '--', '--'], > [-8522787127447073495, 'barry', 'green'], > ['--', '--', '--'], > ['--', '--', '--'], > ['--', '--', '--'], > [-9092791511155847987, 'timmy', 'red'], > ['--', '--', '--'], > [-6480567542315338377, 'guido', 'blue']] > >

　　然而，完全可以这么存储：

indices = [None, 1, None, None, None, 0, None, 2]

1 2 3 4 5

> 复制代码entries = [[-9092791511155847987, 'timmy', 'red'], > [-8522787127447073495, 'barry', 'green'], > [-6480567542315338377, 'guido', 'blue']] > >

　　indices的作用类似ma_smalltable，但只存储一个数组的索引值，数组只存储实际存在的元素（PyDictEntry），当dict中的元素越稀疏，相比上一种存储方式使用的内存越少。而且，这种实现， dict就是有序的（按插入时间排序）

　　这就是python3.6中新的dict实现，Compact dict! Stackoverflow上也有相关讨论。

总结

　　本文中介绍了Python内存优化的Profile工具，最有效的优化方法：使用slots，也介绍了在python3.6中新的dict实现。

　　当然，还有一些良好的编码习惯。比如尽量使用immutable而不是mutable对象：使用tuple而不是list，使用frozenset而不是set；另外，就是尽量使用迭代器，比如python2.7中，使用xrange而不是range，dict的iterxx版本。

references

pytracemalloc

pep454： Add a new tracemalloc module to trace Python memory allocations　

save-ram-with-python-slots/

Python源码分析-PyDictObject

More compact dictionaries with faster iteration

本文版权归作者xybaby（博文地址：http://www.cnblogs.com/xybaby/）所有，欢迎转载和商用，请在文章页面明显位置给出原文链接并保留此段声明，否则保留追究法律责任的权利，其他事项，可留言咨询。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

在学习 Java 集合时， 最先学习的便是 List 中的 ArrayList 和 LinkedList， 学习集合很关键的是学习其源码， 了解底层实现方式， 那么今天就讲讲 ArrayList 实现的一个接口 RandomAccess。

好奇心的产生

查看 ArrayList 的源码， 发现它实现了 RandomAccess 这个接口， 出于好奇点进去看看， 结果发现这接口是空的， 这当然引发了更大的好奇心：这空架子到底有何用？

深入探究

JDK 官方文档是不可少的工具， 先看看它是怎么说的：RandomAccess 是 List 实现所使用的标记接口，用来表明其支持快速（通常是固定时间）随机访问。此接口的主要目的是允许一般的算法更改其行为，从而在将其应用到随机或连续访问列表时能提供良好的性能。

标记接口（Marker）：这就说明了 RandomAccess 为空的原因，这个接口的功能仅仅起到标记的作用。

这不是与序列化接口 Serializable 差不多吗？ 只要你认真观察， 其实不只这一个标记接口， 实际上 ArrayList 还实现了另外两个这样的空接口：

Cloneable 接口 ：实现了 Cloneable 接口，以指示 Object.clone() 方法可以合法地对该类实例进行按字段复制。 如果在没有实现 Cloneable 接口的实例上调用 Object 的 clone 方法，则会导致抛出 CloneNotSupportedException 异常。

Serializable 接口： 类通过实现 java.io.Serializable 接口以启用其序列化功能。未实现此接口的类将无法使其任何状态序列化或反序列化。

继续探讨

标记接口都有什么作用呢？ 继续讨论 RandomAccess 的作用，其他两个在此不作讨论。

如果 List 子类实现了 RandomAccess 接口，那表示它能快速随机访问存储的元素， 这时候你想到的可能是数组， 通过下标 index 访问， 实现了该接口的 ArrayList 底层实现就是数组， 同样是通过下标访问， 只是我们需要用 get() 方法的形式 ， ArrayList 底层仍然是数组的访问形式。

同时你应该想到链表， LinkedList 底层实现是链表， LinkedList 没有实现 RandomAccess 接口，发现这一点就是突破问题的关键点。

数组支持随机访问， 查询速度快， 增删元素慢； 链表支持顺序访问， 查询速度慢， 增删元素快。所以对应的 ArrayList 查询速度快，LinkedList 查询速度慢， RandomAccess 这个标记接口就是标记能够随机访问元素的集合， 简单来说就是底层是数组实现的集合。

为了提升性能，在遍历集合前，我们便可以通过 instanceof 做判断， 选择合适的集合遍历方式，当数据量很大时， 就能大大提升性能。

随机访问列表使用循环遍历，顺序访问列表使用迭代器遍历。

先看看 RandomAccess 的使用方式

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

复制代码import java.util.*;
public class RandomAccessTest {
    public static void traverse(List list){

        if (list instanceof RandomAccess){
            System.out.println("实现了RandomAccess接口，不使用迭代器");

            for (int i = 0;i < list.size();i++){
                System.out.println(list.get(i));
            }

        }else{
            System.out.println("没实现RandomAccess接口，使用迭代器");

            Iterator it = list.iterator();
            while(it.hasNext()){
                System.out.println(it.next());
            }

        }
    }
    public static void main(String[] args) {
        List<String> arrayList = new ArrayList<>();
        arrayList.add("a");
        arrayList.add("b");
        traverse(arrayList);

        List<String> linkedList = new LinkedList<>();
        linkedList.add("c");
        linkedList.add("d");
        traverse(linkedList);
    }
}

下面我们加入大量数据进行性能测试：

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

复制代码import java.util.*;
public class RandomAccessTimeTest {

    //使用for循环遍历
    public static long traverseByLoop(List list){
        long startTime = System.currentTimeMillis();
        for (int i = 0;i < list.size();i++){
            list.get(i);
        }
        long endTime = System.currentTimeMillis();
        return endTime-startTime;
    }

    //使用迭代器遍历
    public static long traverseByIterator(List list){
        Iterator iterator = list.iterator();
        long startTime = System.currentTimeMillis();
        while (iterator.hasNext()){
            iterator.next();
        }
        long endTime = System.currentTimeMillis();
        return endTime-startTime;
    }

    public static void main(String[] args) {
        //加入数据
        List<String> arrayList = new ArrayList<>();
        for (int i = 0;i < 30000;i++){
            arrayList.add("" + i);
        }
        long loopTime = RandomAccessTimeTest.traverseByLoop(arrayList);
        long iteratorTime = RandomAccessTimeTest.traverseByIterator(arrayList);
        System.out.println("ArrayList:");
        System.out.println("for循环遍历时间:" + loopTime);
        System.out.println("迭代器遍历时间:" + iteratorTime);

        List<String> linkedList = new LinkedList<>();
        //加入数据
        for (int i = 0;i < 30000;i++){
            linkedList.add("" + i);
        }
        loopTime = RandomAccessTimeTest.traverseByLoop(linkedList);
        iteratorTime = RandomAccessTimeTest.traverseByIterator(linkedList);
        System.out.println("LinkedList:");
        System.out.println("for循环遍历时间:" + loopTime);
        System.out.println("迭代器遍历时间:" + iteratorTime);
    }
}

结果：

ArrayList:
for 循环遍历时间: 3
迭代器遍历时间: 7

LinkedList:
for 循环遍历时间: 2435
迭代器遍历时间: 3

结论

根据结果我们可以得出结论：
ArrayList 使用 for 循环遍历优于迭代器遍历
LinkedList 使用 迭代器遍历优于 for 循环遍历

根据以上结论便可利用 RandomAccess 在遍历前进行判断，根据 List 的不同子类选择不同的遍历方式， 提升算法性能。

学习阅读源码， 发现底层实现的精妙之处， 改变自己的思维， 从每一个小细节提升代码的性能。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

CI/CD平台的作用我就不再啰嗦了。现在又比较流行docker容器化(快速，环境无缝切换等作用)，所以世面上还没有一套成型的，固定的套件来拱企业快速搭建自己的CI/CD平台(花钱买服务的除外),docker云也更是五花八门，DaoCloud,Swarm等。作者有幸参加了公司的CI/CD平台搭建，从无到有的做了一套持续集成持续交付平台，算上大致接近尾声了。这里分享一下，阅读此博文需要以下知识储备与要求:

• Jenkins
• docker，docker云(Rancher或者其他云)
• 着手做过CI平台

文章结构

1. 基于Jenkins,Shell脚本的传统CI平台
2. 基于Jenkins,docker云的容器化CI/CD平台

1. 基于Jenkins,Shell脚本的传统CI平台

作者在上家公司着手做了一套基于Shell脚本的CI平台，这里不再啰嗦，只简单放一张图:

2. 基于Jenkins,docker云的容器化CI/CD平台

重头戏来了。基于Shell脚本的部署其实还是能解决一般企业的CI需求，然而，对于微服务架构的多工程快速部署，平台无关性，就需要容器化部署了。
同理，上一张图:

下面开始一步一步的讲解:

1. 代码提交

2. 根据webhook触发Jenkins构建

说白了，就是代码提到Gitlab，在Gitlab配置一个webhook像Jenkins发送一个请求，Jenkins收到这个请求后就会开始执行构建。当然了，Jenkins里是依赖的Gitlab plugin插件来支持这个功能的。参考博文:Gitlab plugin使用

3. 进行构建

这一步是最繁琐的，也是要重点讲的，这里涉及到很多配置。首先需要说一下的是，我这里的Jenkins是物理部署的。这一步Jenkins的配置我会在最后贴出来。Jenkins在这一步里执行构建(clean install)后打包好工程后自动根据插件

1
2
3

复制代码                <groupId>com.spotify</groupId>
                <artifactId>docker-maven-plugin</artifactId>
                <version>0.4.13</version>

依赖一个事先准备好的Dockerfile构建出docker镜像并推送到远程Habor.
这一步需要注意如下配置:

1. docker环境的配置。
2. docker-maven-plugin配置
   这两个配置都不是那么好配置的。

• 先说第一个，docker环境的安装好后，需要设置Docker的Insecure registries里加上自己的私服Harbor地址，这是因为docker在进行推送镜像时是https协议，而Harbor是http协议的，所以要设置信任这个地址，否则在进行推送时会报错的。参考博文:docker配置信任地址，还是给大家截图一下我的配置：
  
• 再说第二个，这个插件的配置就更难了，先上我自己的配置。

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

复制代码<build>
        <resources>
            <resource> //必须要转换一下，因为Dockerfile里有内置变量需要转换
                <directory>src/main/docker</directory>
                <filtering>true</filtering>
                <includes>
                    <include>**/Dockerfile</include>
                </includes>
                <targetPath>../docker</targetPath>
            </resource>
            <!--注意这里一定要把resources目录下的配置文件引进去,并且不要指定targetPath-->
            <resource>
                <directory>src/main/resources</directory>
                <includes>
                    <include>**/*.yml</include>
                </includes>
                <filtering>false</filtering>
            </resource>
        </resources>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
            <plugin>
                <groupId>com.spotify</groupId>
                <artifactId>docker-maven-plugin</artifactId>
                <version>0.4.13</version>
                <executions>
                    <execution>
                        <phase>install</phase>
                        <goals>
                            <goal>build</goal>
                        </goals>
                    </execution>
                </executions>
                <configuration>
                    <serverId>docker-private-registry</serverId>
                    <registryUrl>${docker.repostory}</registryUrl>
                    <pushImage>true</pushImage>
                    <imageName>
                      ${docker.repostory}/${docker.registry.name}/${project.artifactId}
                    </imageName>
                    <dockerDirectory>
                        ${project.build.directory}/docker
                    </dockerDirectory>
                    <resources>
                        <rescource>
                            <targetPath>/</targetPath>
                            <directory>${project.build.directory}</directory>
                            <include>${project.build.finalName}.jar</include>
                        </rescource>
                    </resources>
                </configuration>
            </plugin>
        </plugins>
    </build>

这里需要指出的是，docker-private-registry这一行，一定要在自己maven的setting文件里配置上自己Harbor和账号密码。

1
2
3
4
5
6
7
8

复制代码<server>
<id>docker-private-registry</id>
<username>zdy</username>
<password>****</password>
<configuration>
<email>something@qq.com</email>
</configuration>
</server>

给个参考博文(建议大家用我的配置，参考博文真的只能参考，很多错误):spotify插件

然后给下自己的srm/main/docker目录下的Dockerfile

1
2
3
4
5
6
7
8
9
10
11

复制代码FROM anapsix/alpine-java:8

ENV APP_NAME @project.build.finalName@.@project.packaging@
# 后期更改为日志目录,将日志挂载出来
VOLUME /tmp
#默认工作目录/
ADD $APP_NAME /app.jar

EXPOSE 8070

ENTRYPOINT ["java","-jar","/app.jar"]

上面的具体配置我就不一一讲了，因为很复杂，像dockerfile里的内置变量转换不出来什么的，大家就多参考参考我的配置，配置都是整理了很长时间才出来的，还是那句老话，只讲思路，具体配置大家多参考其他文章，思路出来了，慢慢的就可以摸索出来。

4. 构建完成后发送CURL请求触发Rancher的WebHook

按照上面的配置后,jenkins构建算是完成了，我们新构建的docker镜像也推送到了远程的HarBor私服镜像库。这时就要执行CURL指令来触发Rancher拉取镜像部署了。
下面给一下我jenkins的截图。

至于CURL后面如何写的，Rancher官网webhook，需要事先在Rancher的webhook配置，生成一些参数，然后curl.我就不多讲了，大家自行看官网。
Rancher WebHook

5. Rancher根据WebHook触发后拉取镜像并且部署。

讲的比较草率，可能很多人很蒙蔽，其实这篇博文的精华就是开头那一张图，按照图中的思路摸索摸索就大致可以出来，为什么作者不愿意详讲呢，因为说实在的配置这个东西太麻烦了，很多配置很复杂，都需要自己摸索。我也只能提一提思路。

效果我就不演示，提交完代码后，在Jenkins构建好镜像自动推送到Harbor，然后Curl Ranche的WebHook,触发Rancher拉取最新镜像并部署。
Over, Have a good day .

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

2017源创会年终盛典，8折限时优惠！ >>>

Sharding-JDBC 2.0.0，在经过几个月的开发，和3个里程碑的迭代之后正式发布。Sharding-JDBC集分库分表、读写分离、分布式主键、柔性事务和数据治理与一身，提供一站式的解决分布式关系型数据库的解决方案。

从2.x版本开始，Sharding-JDBC正式将包名、Maven坐标、码云仓库、GitHub仓库和官方网站统一为io.shardingjdbc。这意味着除了当当的无私奉献，我们也乐于采纳第三方公司的代码贡献。本次2.0.0的版本，由当当与数人云共同开发，感谢当当与数人云开发人员的辛苦工作。

Sharding-JDBC 1.x关注SQL兼容性、分库分表、读写分离、分布式主键、柔性事务等分片功能；Sharding-JDBC 2.x提供了全新的Orchestration模块，关注数据库和数据库访问层应用的治理。2.0.0在治理方面的主要更新是：

1. 配置动态化。可以通过zookeeper或etcd作为注册中心动态修改数据源以及分片规则。
2. 数据治理。提供熔断数据库访问程序对数据库的访问和禁用从库的访问的能力。
3. 跟踪系统支持。可以通过sky-walking等基于Opentracing协议的APM系统中查看sharding-jdbc的调用链，并提供sky-walking的自动探针。
4. 提供sharding-jdbc的spring-boot-starter。

通过2.x提供的数据治理能力，sharding-jdbc的架构图更新为：

数人云不但贡献了sharding-jdbc的核心代码，还提供了hawk的统一配置中心平台，也会于近期开源。通过对sharding-jdbc注册中心的读写，提供了对配置的图形化界面支持。

著名的apm开源软件sky-walking也将于近期采用sharding-jdbc作为其底层存储追踪日志的存储引擎。

sharding-jdbc将与配置中心hawk，apm的sky-walking一起打造分布式服务的生态圈。

欢迎访问Sharding-JDBC的官网：shardingjdbc.io/。感谢开源中国对Sharding-JDBC的大力支持。目前2017年最受欢迎的中国开源软件也在评选中，欢迎为Sharding-JDBC投上您宝贵的一票。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

娇弱的 PHP [ php-fpm.conf & php.ini 安全优化实践 ]

2017-11-23
phpsec

0x01 关于 php

0x02 演示环境

0x03 下载 php-5.6.32.tar.gz,并安装好所需各种依赖库

0x04 开始编译安装php5.6.32,要带的编译参数比较多,大家下去以后,可以仔细了解下这些参数都是干什么用的,其实,都是一些php内置功能模块,这里默认就已经启用了一些比较常用的模块,如,pdo,mysqli,关于下面的模块,并不用全部都装,根据你自己实际的开发业务需求,用什么装什么即可,切记不要一上来不管用不用就先装一大堆,你不用,很可能就会被别人利用

0x05 编辑,配置并优化php-fpm.conf,即 fastcgi 的服务端,如下

| 123456 | # 让php进程以一个系统伪用户的身份起来,在能满足实际业务需求的情况下,最大限度降低php进程权限# useradd -s /sbin/nologin -M phpfpm # cp /usr/local/php/etc/php-fpm.conf.default /usr/local/php/etc/php-fpm.conf# vi /usr/local/php/etc/php-fpm.conf# mkdir /usr/local/php/logs# egrep -v "^$|;" /usr/local/php/etc/php-fpm.conf 简化php-fpm配置文件 |
| — | — |

| 12345 | # /usr/local/php/sbin/php-fpm 启动php-fpm# ps -le | grep "php-fpm"# netstat -tulnp | grep ":9000"# echo "/usr/local/php/sbin/php-fpm" >> /etc/rc.local && cat /etc/rc.local# killall php-fpm 如果想关闭php-fpm,直接把它进程kill掉即可 |
| — | — |

尝试让php与mysql,nginx 进行联动,看看php能不能被正常解析

0x06 最后,我们就来好好关注下php解析器自身的安全,php解析器的设置全部依靠php.ini文件来实现,所以,下面就来详细说明一下针对php.ini的安全配置

将 register_globals 项设为Off,本身的意思就是注册为全局变量,也就是说,设置为On的时候,从客户端传过来的参数值会被直接注册到php全局变量中后端直接可以拿到该变量进行使用,如果为Off,则表示只能到特定的全局数组中才能取到该数据,建议关闭,容易造成变量覆盖问题,不过在php高版本[如,> 5.6.x]中,已经去除对此项的设置,官方给的说明是这样的本特性已自 PHP 5.3.0 起废弃并将自 PHP 5.4.0 起移除,如果你用的还是低版本的php就需要注意把此项关闭

将 cgi.fix_pathinfo的值设为 0,默认cgi.fix_pathinfo 项是开启的,即值为1,它会对文件路径自动进行修正,我们要把它改成0,不要让php自动修正文件路径,防止入侵者利用此特性构造解析漏洞来配合上传webshell

建议同时关闭以下两项,如果实在有业务需求,请在代码中严格限制检查用户传过来的数据

禁用各种高危函数,尽可能让各种 webshell [ 一句话,大马 ] 无法再靠php内置函数来执行各种系统命令,下面是一些比较常见的命令和代码执行函数,如果你还发现有其它的一些不常用的高危函数,也可以一并加进来,防止被入侵者率先发现并利用,此项默认为空

转义开关,主要用来转义各种特殊字符,如,单引号,双引号,反斜线和空字符...个人建议,在这里先把它关闭,因为它并不能很好的防住sql注入,或者,基本是防不住的,比如,利用宽字节 说到这儿,顺便再补充一句,对付宽字节的最好办法就是全站统一使用 utf-8,这里还是建议大家采用sql语句预编译和绑定变量的方式来预防sql注入,这也是目前为止比较切实有效的预防手段,对于从客户端过来的各种其它数据,可以单独写个检查类,如果你想安全就不要对这些开关寄予太大的希望,可能php官方也发现,这个开关实质就是个摆设,所以给出了这样的说明本特性已自 PHP 5.3.0 起废弃并将自 PHP 5.4.0 起移除

关闭php自身的各种错误回显,反正只要记得,项目上线后,所有的程序错误一律接收到我们自己事先准备好的地方,一旦被入侵者在前端看到,极易造成敏感信息泄露,高版本的php,默认这些危险项就是处于关闭状态的

隐藏php的详细版本号,即X-Powered-By中显示的内容,不得不再次强调,有些漏洞只能针对特定的类型版本,在实际渗透过程中,如果让入侵者看到详细的版本号,他很可能就会直接去尝试利用该版本所具有的一些漏洞特性再配合着其它漏洞一起使用

限制php对本地文件系统的访问,即把所有的文件操作都限制指定的目录下,让php其实就是限制了像fopen()这类函数的访问范围,一般主要用来防止旁站跨目录,把webshell死死控制在当前站点目录下,此项默认为空,不建议直接写到php.ini中,可以参考前面nginx安全部署中的,直接在每个站点目录下新建一个.user.ini然后再把下面的配置写进去即可,比较灵活

关于对服务端session的一些处理

隐藏后端使用的真正脚本类型,扰乱入侵者的渗透思路,另外,切记不要把敏感数据直接明文存在session中,有泄露风险

修改session文件存放路径,最好不要直接放在默认的/tmp目录下,实际中可能是一台单独的session服务器,比如,memcached

安全模式可根据实际业务需求选择性开启,安全模式的意思就是操作文件的函数只能操作与php进程UID相同的文件,但php进程的uid并不一定就是web服务用户的uid,这也就造成了麻烦,也就是说,你想避免这种麻烦,可能就需要在最开始配置时就让php进程和web服务使用同一个系统用户身份,但这又正好跟我前面说的相背了,我们在前面说过,最好把php进程用户和web服务用户分开,这样更容易进行权限控制,另外,高版本的php[ > php5.4 ]已不再支持安全模式,因为官方可能也觉得它并没什么卵用,而且低版本php的安全模式,还可被绕过,所以,如果你用的是低版本的php,请根据自身实际业务做取舍

限制php单脚本执行时长,防止服务器资源被长期滥用而产生拒绝服务的效果

关于上传,如果实际的业务根本不涉及到上传,直接把上传功能关掉即可,如果需要上传,再根据需求做出调整即可,对防入侵来讲,这里对我们意义并不是非常大

0x07 利用 chattr 锁定一些不需要经常改动的重要配置文件,如,php-fpm.conf,php.ini,my.cnf…,为了防止chattr工具被别人滥用,你可以把它改名隐藏到系统的某个角落里,用的时候再拿出来

锁定

解锁

0x08 务必勤于关注php官方的高危补丁发布及说明,和其它工具不同,php 自身bug多,因为关注的人多,搞的人更多,所以暴露出来的各种安全问题也就更多更多

0x09 最后,告诉大家一个怎么把曾经yum下来的包保留着的办法,在系统断网的情况下也许用的着,只需到yum配置里面去调整下即可,保留的包的路径也在yum配置中定义好了

小结:
至此为止,关于整套LNMP架构的安全部署及优化,也就差不多完成了,我们关注的点,更多可能还是集中在防御入侵上,捎带了一点性能优化,这里所给的参数选项基本全部都可直接用于实战部署,但并不是所有的都是必须的,还需要你好好根据自己实际的业务需求做出适当取舍,或者在此基础进行定制改进,有些地方都是基于自己平时实战经验的考量来的,并不一定完全是对的,如果有性能更好,更安全的方案,也非常欢迎大家一起来私信交流
^_^

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

从这一篇开始，大概会花四五篇的内容篇幅，归纳整理一下之前学过的SQL数据库，一来可以为接下来数据分析工作提前巩固基础，二来把以前学的SQL内容系统化、结构化。

今天这一篇仅涉及MySQL与本地文本文件的导入导出操作，暂不涉及主要查询语言以及MySQL与R语言和Python的交互。

平台使用Navicat Premium（当然你也可以使用MySQL自带的workbench或者MySQL Conmand line）。

以下仅涉及MySQL中使用命令行语句导入/导出本地磁盘的文本文件（csv\txt文件）。

文件导入（csv）：

在导入本地文件之前，请确保你的MySQL设置有本地文件导入导出权限。

在导入MySQL之前，需要在指定数据库中先建立空表，以备之后导入。

1
2
3
4
5
6
7
8
9

复制代码USE db1;
CREATE TABLE subway (
    ID      INTEGER(5)  NOT NULL AUTO_INCREMENT,   -- ID是主键，格式为整数，非空值、自增列
    address VARCHAR(10) NOT NULL,
    lon     FLOAT(10,6) NOT NULL,                  -- 浮点
    lat     FLOAT(10,6) NOT NULL,                  -- 浮点
    Type    VARCHAR(10) NOT NULL,                  -- 字符型（10位）
    PRIMARY KEY (ID)                            -- 设定主键
    );

创建空表的语句格式如上：

1
2
3
4
5

复制代码CREATE TABLE 表名 (
  column1 类型(字符位数)  是否允许为空值 自增列（可选） 默认值（可选）,
  column2 类型(字符位数)  是否允许为空值 自增列（可选） 默认值（可选）,
  PRIMARY KEY (column1)
  );

此时在MySQL中生成一个空表（仅有字段名称）。

以下是导入语句：

1
2
3
4
5
6
7
8
9

复制代码load data local infile 'D:/subway.txt' 
into table db1.subway          -- 导入本地文件语句
character set gbk                -- 设置导入文件编码     
fields terminated by '\t'   -- 指定txt文件内的字段分隔符
optionally enclosed by '"'  -- 指定字符闭合符(可选参数，有些格式txt会设置字符使用双引号/单引号包括等格式)
escaped by '"'                 -- 指定转义符（字符内含符号与闭合符冲突，使用何种符号进行包括并转义，使其保留原意）
lines terminated by '\r\n'  -- 指定换行符
ignore 1 lines                 -- 指定从文件第几行开始导入（如果本地文件有行名，需要略过一行）
(address,lon,lat,Type);             -- 最后一行指定要导入的列名(次内列名需与之前新建的空表列名严格匹配)

主键可以设定为导入列中的某一列(保证无缺失值无重复值即可)，并不是必须设置的。

做简单的表格信息概览：

1

复制代码desc db1.subway

文件导出（TXT）：

将刚才导入的subway文件导出到本地。

1
2
3
4
5
6
7
8

复制代码select "ID","address","lon","lat","Type"   -- 为要导出的字典命名
union select * from db1.subway       -- 指定要从目标表中导出的字段（与第一句指定的字段严格对应）
into outfile 'D:/SUBWAY.txt'            -- 导出目录及文件名
character set gbk                         -- 设置输出编码
FIELDS TERMINATED BY '\t'            -- 字段分隔符
-- OPTIONALLY ENCLOSED BY '"'                  -- 文本包括符号(可选，这里注释掉了)            
-- escaped by '"'                              -- 冲突转义符（可选，这里注释掉了）
LINES TERMINATED BY '\r\n';          -- 换行符

文件导入（csv）:

仍然是导入之前先新建空表：

1
2
3
4
5
6
7
8
9
10

复制代码CREATE TABLE President (

    ID         INTEGER(5)  NOT NULL AUTO_INCREMENT,   -- ID是主键，格式为整数，非空值、自增列
    STATE_NAME VARCHAR(15) NOT NULL,                  -- 字符型（15位）
    STATE_ABBR VARCHAR(5)  NOT NULL,                  -- 浮点
    Count      INTEGER(5)  NOT NULL,                  -- 整数型（5位）
    Clinton    FLOAT(8,4)  NOT NULL,                  -- 浮点型（8位，保留四位小数）
    Trump      FLOAT(8,4)  NOT NULL,
    Results    VARCHAR(5)  NOT NULL,    PRIMARY KEY (ID)                             
    );

导入本地CSV文件：

1
2
3
4
5
6
7
8
9

复制代码load data local infile 'D:/President.csv' 
into table db1.president       -- 导入本地文件语句
-- character set gbk                  -- 设置导入文件编码 (因为原始文件就是utf-8编码的，这里无需指定，如果不是需要单独指定) 
fields terminated by ','    -- 指定txt文件内的字段分隔符
optionally enclosed by '"'  -- 指定字符闭合符(可选参数，有些格式txt会设置字符使用双引号/单引号包括等格式)
escaped by '"'                 -- 指定转义符（字符内含符号与闭合符冲突，使用何种符号进行包括并转义，使其保留原意）
lines terminated by '\r\n'  -- 指定换行符
ignore 1 lines                 -- 指定从文件第几行开始导入（如果本地文件有行名，需要略过一行）
(STATE_NAME,STATE_ABBR,Count,Clinton,Trump,Results);

DESC db1.president

文件导出（csv）:

将刚才导入的President文件导出到本地csv文件。

1
2
3
4
5
6
7
8

复制代码select "ID","STATE_NAME","STATE_ABBR","Count","Clinton","Trump","Results"  -- 为要导出的字典命名
union select * from db1.president        -- 指定要从目标表中导出的字段（与第一句指定的字段严格对应）
into outfile 'D:/President1.csv'            -- 导出目录及文件名
character set gbk                             -- 设置输出编码
FIELDS TERMINATED BY ','                 -- 字段分隔符
OPTIONALLY ENCLOSED BY '"'               -- 文本包括符号(可选，这里注释掉了)
escaped by '"'                              -- 冲突转义符（可选，这里注释掉了）
LINES TERMINATED BY '\r\n';              -- 换行符

除此之外，还有几个基础的增删命令需要掌握：

插入命令：

1
2
3
4
5
6
7
8
9

复制代码INSERT INTO president
   (LastName, Address)          -- 插入列名称
VALUES 
  ('Wilson', 'Champs-Elysees'); -- 具体值
select count(*) from db1.president
INSERT INTO db1.president (STATE_NAME,STATE_ABBR,`Count`,Clinton,Trump,Results) 
VALUES ('Wilson', 'WL',10,0.4555,0.2344,'T');

select count(*) from db1.president

关于删除表：

1
2
3
4

复制代码truncate db1.president;       -- 删除表内所有记录（保留空表）
drop table db1.president;  -- 彻底删除表（数据库中该表将不存在）
drop table db1.subway;     -- 彻底删除表（数据库中该表将不存在）
select count(*) from db1.president

本文小结：

文件导入（txt\csv）

数据导出（TXT\csv）

表创建

表删除

记录插入

在线课程请点击文末原文链接：

Hellobi Live | R语言可视化在商务场景中的应用
往期案例数据请移步本人GitHub：
github.com/ljtyduyu/Da…

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

RPC

RPC即远程过程调用，它的提出旨在消除通信细节、屏蔽繁杂且易错的底层网络通信操作，像调用本地服务一般地调用远程服务，让业务开发者更多关注业务开发而不必考虑网络、硬件、系统的异构复杂环境。

RPC过程

先看看集群中RPC的整个通信过程，假设从节点node1开始一个RPC调用，

1. 先将待传递的数据放到NIO集群通信框架中；
2. 由于使用的是NIO模式，线程无需阻塞直接返回；
3. 由于与集群其他节点通信需要花销若干时间，为了提高CPU使用率当前线程应该放弃CPU的使用权进行等待操作；
4. NIO集群通信框架接收到node2节点的响应消息，并将消息封装成Response对象保存至响应数组；
5. 通信框架接收到node4节点的响应消息，由于是使用了并行通信，所以node4可能比node3先返回消息，并将消息封装成Response对象保存至响应数组；
6. 通信框架最后接收到node3节点的响应消息，并将消息封装成Response对象保存至响应数组；
7. 现在所有节点的响应都已经收集完毕，是时候通知刚刚被阻塞的那条线程了，原来的线程被notify醒后拿到所有节点的响应Response[]进行处理，至此完成了整个集群RPC过程。

多线程

上面整个过程是在只有一条线程的情况下，一切看起来没什么问题，但如果有多条线程并发调用则会导致一个问题：线程与响应的对应关系将被打乱，无法确定哪个线程对应哪几个响应。

因为NIO通信框架不会每个线程都独自使用一个socket通道，为提高性能一般都是使用长连接，所有线程共用一个socket通道，这时就算线程一比线程二先放入通信框架也不能保证响应一比响应二先接收到，所以接收到响应一后不知道该通知线程一还是线程二。只有解决了这个问题才能保证RPC调用的正确性。

怎么解决多线程

要解决线程与响应对应的问题就需要维护一个线程响应关系列表，响应从关系列表中就能查找对应的线程，如图，在发送之前生成一个UUID标识，此标识要保证同socket中唯一，再把UUID与线程对象关系对应起来，可使用Map数据结构实现，UUID的值作为key，线程对应的锁对象为value。

接着制定一个协议报文，UUID作为报文的其中一部分，报文发往另一个节点node2后将响应信息message放入报文中并返回，node1对接收到的报文进行解包根据UUID去查找并唤起对应的线程，告诉它“你要的消息已经收到，往下处理吧”。但在集群环境下，我们更希望是集群中所有节点的消息都接收到了才往下处理，如图下半部分，一个UUID1的请求报文会发往node2、node3和node4三个节点，这时假如只接收到一个响应则不唤起线程，直到node2、node3对应UUID1的响应报文都接收到后才唤起对应线程往下执行。同样地，UUID2、UUID3的报文消息都是如此处理，最后集群中对应的响应都能正确回到各自线程上。

来个例子

用简单代码实现一个RPC例子，自行选择一个集群通信框架负责底层通信，接着往下：

1. 定义一个RPC接口，这些方法是预留提供给上层具体逻辑处理的入口，replyRequest方法用于处理响应逻辑，leftOver方法用于残留请求的逻辑处理。

1
2
3
4

复制代码public interface RpcCallback {  
    public Serializable replyRequest(Serializable msg, Member sender);  
    public void leftOver(Serializable msg, Member sender);  
}

2. 定义通信消息协议，实现Externalizable接口自定义序列化和反序列化，message用于存放响应消息，uuid标识用于关联线程，rpcId用于标识RPC实例，reply表示是否回复。

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

复制代码public class RpcMessage implements Externalizable {
  protected Serializable message;
  protected byte[] uuid;
  protected byte[] rpcId;
  protected boolean reply = false;

  public RpcMessage() {}

  public RpcMessage(byte[] rpcId, byte[] uuid, Serializable message) {
    this.rpcId = rpcId;
    this.uuid = uuid;
    this.message = message;
  }

  @Override
  public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
    reply = in.readBoolean();
    int length = in.readInt();
    uuid = new byte[length];
    in.readFully(uuid);
    length = in.readInt();
    rpcId = new byte[length];
    in.readFully(rpcId);
    message = (Serializable) in.readObject();
  }

  @Override
  public void writeExternal(ObjectOutput out) throws IOException {
    out.writeBoolean(reply);
    out.writeInt(uuid.length);
    out.write(uuid, 0, uuid.length);
    out.writeInt(rpcId.length);
    out.write(rpcId, 0, rpcId.length);
    out.writeObject(message);
  }
}

3. 响应类型，提供多种唤起线程的条件，一共四种类型，分别表示接收到第一个响应就唤起线程、接收到集群中大多数节点的响应就唤起线程、接收到集群中所有节点的响应才唤起线程、无需等待响应的无响应模式。

1
2
3
4
5
6

复制代码public class RpcResponseType {  
  public static final int FIRST_REPLY = 1;  
  public static final int MAJORITY_REPLY = 2;  
  public static final int ALL_REPLY = 3;  
  public static final int NO_REPLY = 4;  
}

4. 响应对象，用于封装接收到的消息，Member在通信框架是节点的抽象，这里用来表示来源节点。

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

复制代码public class RpcResponse {
  private Member source;
  private Serializable message;

  public RpcResponse() {}

  public RpcResponse(Member source, Serializable message) {
    this.source = source;
    this.message = message;
  }

  public void setSource(Member source) {
    this.source = source;
  }

  public void setMessage(Serializable message) {
    this.message = message;
  }

  public Member getSource() {
    return source;
  }

  public Serializable getMessage() {
    return message;
  }
}

5. RPC响应集，用于存放同个UUID的所有响应。

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

复制代码public class RpcCollector {  
    public ArrayList<RpcResponse> responses = new ArrayList<RpcResponse>();   
    public byte[] key;  
    public int options;  
    public int destcnt;  
    public RpcCollector(byte[] key, int options, int destcnt) {  
        this.key = key;  
        this.options = options;  
        this.destcnt = destcnt;  
    }  
    public void addResponse(Serializable message, Member sender){  
        RpcResponse resp = new RpcResponse(sender,message);  
        responses.add(resp);  
    }  
    public boolean isComplete() {  
        if ( destcnt <= 0 ) return true;  
        switch (options) {  
            case RpcResponseType.ALL_REPLY:  
                return destcnt == responses.size();  
            case RpcResponseType.MAJORITY_REPLY:  
            {  
                float perc = ((float)responses.size()) / ((float)destcnt);  
                return perc >= 0.50f;  
            }  
            case RpcResponseType.FIRST_REPLY:  
                return responses.size()>0;  
            default:  
                return false;  
        }  
    }  
    public RpcResponse[] getResponses() {  
        return responses.toArray(new RpcResponse[responses.size()]);  
    }  
}

6. RPC核心类，是整个RPC的抽象，它实现通信框架的ChannelListener接口，实现了该接口就能在messageReceived方法中处理接收到的消息。因为所有的消息都会通过此方法，所以它必须要根据key去处理对应的线程，同时它也要负责调用RpcCallback接口定义的相关的方法，例如响应请求的replyRequest方法和处理残留的响应leftOver方法，残留响应是指有时我们在接收到第一个响应后就唤起线程。

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

复制代码public class RpcChannel implements ChannelListener {
  private Channel channel;
  private RpcCallback callback;
  private byte[] rpcId;
  private int replyMessageOptions = 0;
  private HashMap<byte[], RpcCollector> responseMap = new HashMap<byte[], RpcCollector>();

  public RpcChannel(byte[] rpcId, Channel channel, RpcCallback callback) {
    this.rpcId = rpcId;
    this.channel = channel;
    this.callback = callback;
    channel.addChannelListener(this);
  }

  public RpcResponse[] send(Member[] destination, Serializable message, int rpcOptions,
      int channelOptions, long timeout) throws ChannelException {
    int sendOptions = channelOptions & ~Channel.SEND_OPTIONS_SYNCHRONIZED_ACK;
    byte[] key = UUIDGenerator.randomUUID(false);
    RpcCollector collector = new RpcCollector(key, rpcOptions, destination.length);
    try {
      synchronized (collector) {
        if (rpcOptions != RpcResponseType.NO_REPLY) responseMap.put(key, collector);
        RpcMessage rmsg = new RpcMessage(rpcId, key, message);
        channel.send(destination, rmsg, sendOptions);
        if (rpcOptions != RpcResponseType.NO_REPLY) collector.wait(timeout);
      }
    } catch (InterruptedException ix) {
      Thread.currentThread().interrupt();
    } finally {
      responseMap.remove(key);
    }
    return collector.getResponses();
  }

  @Override
  public void messageReceived(Serializable msg, Member sender) {
    RpcMessage rmsg = (RpcMessage) msg;
    byte[] key = rmsg.uuid;
    if (rmsg.reply) {
      RpcCollector collector = responseMap.get(key);
      if (collector == null) {
        callback.leftOver(rmsg.message, sender);
      } else {

        synchronized (collector) {
          if (responseMap.containsKey(key)) {
            collector.addResponse(rmsg.message, sender);
            if (collector.isComplete()) collector.notifyAll();
          } else {
            callback.leftOver(rmsg.message, sender);
          }
        }
      }
    } else {
      Serializable reply = callback.replyRequest(rmsg.message, sender);
      rmsg.reply = true;
      rmsg.message = reply;
      try {
        channel.send(new Member[] {sender}, rmsg,
            replyMessageOptions & ~Channel.SEND_OPTIONS_SYNCHRONIZED_ACK);
      } catch (Exception x) {}
    }
  }

  @Override
  public boolean accept(Serializable msg, Member sender) {
    if (msg instanceof RpcMessage) {
      RpcMessage rmsg = (RpcMessage) msg;
      return Arrays.equals(rmsg.rpcId, rpcId);
    } else
      return false;
  }
}

7. 自定义一个RPC，它要实现RpcCallback接口，分别对请求处理和残留响应处理，这里请求处理仅仅是简单返回“hello,response for you!”作为响应消息，残留响应处理则是简单输出“receive a leftover message!”。假如整个集群有五个节点，由于接收模式设置成了FIRST_REPLY，所以每个只会接受一个响应消息，其他的响应都被当做残留响应处理。

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

复制代码public class MyRPC implements RpcCallback {
  @Override
  public Serializable replyRequest(Serializable msg, Member sender) {
    RpcMessage mapmsg = (RpcMessage) msg;
    mapmsg.message = "hello,response for you!";
    return mapmsg;
  }

  @Override
  public void leftOver(Serializable msg, Member sender) {
    System.out.println("receive a leftover message!");
  }

  public static void main(String[] args) {
    MyRPC myRPC = new MyRPC();
    byte[] rpcId = new byte[] {1, 1, 1, 1};
    byte[] key = new byte[] {0, 0, 0, 0};
    String message = "hello";
    int sendOptions = Channel.SEND_OPTIONS_SYNCHRONIZED_ACK | Channel.SEND_OPTIONS_USE_ACK;
    RpcMessage msg = new RpcMessage(rpcId, key, (Serializable) message);
    RpcChannel rpcChannel = new RpcChannel(rpcId, channel, myRPC);
    RpcResponse[] resp =
        rpcChannel.send(channel.getMembers(), msg, RpcResponseType.FIRST_REPLY, sendOptions, 3000);
    while (true)
      Thread.currentThread().sleep(1000);
  }
}

可以看到通过上面的RPC封装后，上层可以把更多的精力关注到消息逻辑处理上面了，而不必关注具体的网络IO如何实现，屏蔽了繁杂重复的网络传输操作，为上层提供了很大的方便。

=============广告时间===============

公众号的菜单已分为“分布式”、“机器学习”、“深度学习”、“NLP”、“Java深度”、“Java并发核心”、“JDK源码”、“Tomcat内核”等，可能有一款适合你的胃口。

鄙人的新书《Tomcat内核设计剖析》已经在京东销售了，有需要的朋友可以购买。感谢各位朋友。

为什么写《Tomcat内核设计剖析》

=========================

欢迎关注：

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

作为PHP开发人员，我们并不经常需要担心内存管理。PHP 引擎在我们背后做了很好的清理工作，短期执行上下文的 Web 服务器模型意味着即使是最潦草的代码也不会造成持久的影响。

很少情况下我们可能需要走出这个舒适的地方 —— 比如当我们试图在一个大型项目上运行 Composer 来创建我们可以创建的最小的 VPS 时，或者当我们需要在一个同样小的服务器上读取大文件时。

后面的问题就是我们将在本教程中深入探讨的。

在 GitHub**上可以找到本教程的源码。

Tocy 翻译于 5天前 0人顶 顶 翻译得不错哦！
衡量成功的标准

确保我们对代码进行任何改进的唯一方法是测试一个不好的情况，然后将我们修复之后的测量与另一个进行比较。换句话说，除非我们知道“解决方案”对我们有多大的帮助（如果有的话），否则我们不知道它是否真的是一个解决方案。
这里有两个我们可以关系的衡量标准。首先是CPU使用率。我们要处理的进程有多快或多慢？第二是内存使用情况。脚本执行时需要多少内存？这两个通常是成反比的 - 这意味着我们可以以CPU使用率为代价来降低内存使用，反之亦然。
在一个异步执行模型（如多进程或多线程的PHP应用程序）中，CPU和内存的使用率是很重要的考量因素。在传统的PHP架构中，当任何一个值达到服务器的极限时，这些通常都会成为问题。
测量PHP内的CPU使用率是不切实际的。如果这是你要关注的领域，请考虑在Ubuntu或MacOS上使用类似top的工具。对于Windows，请考虑使用Linux子系统，以便在Ubuntu中使用top。

Tocy 翻译于 5天前 0人顶 顶 翻译得不错哦！
为了本教程的目的，我们将测量内存使用情况。我们将看看在“传统”的脚本中使用了多少内存。我们将执行一些优化策略并对其进行度量。最后，我希望你能够做出一个有经验的选择。
我们查看内存使用多少的方法是：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

复制代码// formatBytes is taken from the php.net documentation

memory_get_peak_usage();

function formatBytes($bytes, $precision = 2) {
    $units = array("b ", "kb ", "mb ", "gb
                                ", "tb ");

    $bytes = max($bytes, 0);
    $pow = floor(($bytes ? log($bytes) : 0) / log(1024));
    $pow = min($pow, count($units) - 1);

    $bytes /= (1 << (10 * $pow));

    return round($bytes, $precision) . "  " . $units[$pow];
}

我们将在脚本的最后使用这些函数，以便我们能够看到哪个脚本一次使用最大的内存。

Tocy 翻译于 5天前 0人顶 顶 翻译得不错哦！

我们的选择是什么？

这里有很多方法可以有效地读取文件。但是也有两种我们可能使用它们的情况。我们想要同时读取和处理所有数据，输出处理过的数据或根据我们所读取的内容执行其他操作。我们也可能想要转换一个数据流，而不需要真正访问的数据。
让我们设想一下，对于第一种情况，我们希望读取一个文件，并且每10,000行创建一个独立排队的处理作业。我们需要在内存中保留至少10000行，并将它们传递给排队的工作管理器（无论采取何种形式）。
对于第二种情况，我们假设我们想要压缩一个特别大的API响应的内容。我们不在乎它的内容是什么，但我们需要确保它是以压缩形式备份的。
在这两种情况下，如果我们需要读取大文件，首先，我们需要知道数据是什么。第二，我们并不在乎数据是什么。让我们来探索这些选择吧…

Tocy 翻译于 5天前 0人顶 顶 翻译得不错哦！
逐行读取文件

有许多操作文件的函数，我们把部分结合到一个简单的文件阅读器中(封装为一个方法)：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

复制代码// from memory.php

function formatBytes($bytes, $precision = 2) {
    $units = array("b ", "kb ", "mb ", "gb
                                ", "tb ");

    $bytes = max($bytes, 0);
    $pow = floor(($bytes ? log($bytes) : 0) / log(1024));
    $pow = min($pow, count($units) - 1);

    $bytes /= (1 << (10 * $pow));

    return round($bytes, $precision) . "  " . $units[$pow];
}

print formatBytes(memory_get_peak_usage());

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

复制代码// from reading-files-line-by-line-1.php

function readTheFile($path) {
    $lines = [];
    $handle = fopen($path, "r ");

    while(!feof($handle)) {
        $lines[] = trim(fgets($handle));
    }

    fclose($handle);
    return $lines;
}

readTheFile("shakespeare.txt ");

require "memory.php ";

我们读取一个文本文件为莎士比亚全集。文件大小为5.5MB，内存占用峰值为12.8MB。现在让我们用一个生成器来读取每一行：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

复制代码// from reading-files-line-by-line-2.php

function readTheFile($path) {
    $handle = fopen($path, "r ");

    while(!feof($handle)) {
        yield trim(fgets($handle));
    }

    fclose($handle);
}

readTheFile("shakespeare.txt ");

require "memory.php ";

文本文件大小不变，但内存使用峰值只是393KB。即使我们能把读取到的数据做一些事情也并不意味着什么。也许我们可以在看到两条空白时把文档分割成块，像这样：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

复制代码// from reading-files-line-by-line-3.php

$iterator = readTheFile("shakespeare.txt ");

$buffer = " ";

foreach ($iterator as $iteration) {
    preg_match("/\n{3}/ ", $buffer, $matches);

    if (count($matches)) {
        print ". ";
        $buffer = " ";
    } else {
        $buffer .= $iteration . PHP_EOL;
    }
}

require "memory.php ";

猜到我们使用了多少内存吗？我们把文档分割为1216块，仍然只使用了459KB的内存，这是否让你惊讶？考虑到生成器的性质，我们使用的最多内存是使用在迭代中我们需要存储的最大文本块。在本例中，最大的块为101985字符。

我已经撰写了使用生成器提示性能和Nikita Popov的迭代器库，如果你感兴趣就去看看吧！

生成器还有其它用途，但是最明显的好处就是高性能读取大文件。如果我们需要处理这些数据，生成器可能是最好的方法。

Tot_ziens 翻译于 5天前 0人顶 顶 翻译得不错哦！

管道间的文件

在我们不需要处理数据的情况下，我们可以把文件数据传递到另一个文件。通常被称为管道（大概是因为我们看不到除了两端的管子里面，当然，它也是不透明的），我们可以通过使用流方法实现。让我们先写一个脚本从一个文件传到另一个文件。这样我们可以测量内存的占用情况：

1
2
3
4
5
6
7

复制代码// from piping-files-1.php

file_put_contents(
    "piping-files-1.txt ", file_get_contents("shakespeare.txt ")
);

require "memory.php ";

不出所料，这个脚本使用更多的内存来进行文本文件复制。这是因为它读取(和保留)文件内容在内存中，直到它被写到新文件中。对于小文件这种方法也许没问题。当为更大的文件时，就捉襟见肘了…

让我们尝试用流(管道)来传送一个文件到另一个：

1
2
3
4
5
6
7
8
9
10
11

复制代码// from piping-files-2.php

$handle1 = fopen("shakespeare.txt ", "r ");
$handle2 = fopen("piping-files-2.txt ", "w ");

stream_copy_to_stream($handle1, $handle2);

fclose($handle1);
fclose($handle2);

require "memory.php ";

这段代码稍微有点陌生。我们打开了两文件的句柄，第一个是只读模式，第二个是只写模式，然后我们从第一个复制到第二个中。最后我们关闭了它，也许使你惊讶，内存只占用了393KB

Tot_ziens 翻译于 4天前 0人顶 顶 翻译得不错哦！
这似乎很熟悉。像代码生成器在存储它读到的每一行代码？那是因为第二个参数fgets规定了每行读多少个字节（默认值是-1或者直到下一行为止）。

第三个参数stream_copy_to_stream和第二个参数是同一类参数（默认值相同），stream_copy_to_stream一次从一个数据流里读一行，同时写到另一个数据流里。它跳过生成器只有一个值的部分（因为我们不需要这个值）。

这篇文章对于我们来说可能是没用的，所以让我们想一些我们可能会用到的例子。假设我们想从我们的CDN中输出一张图片，作为一种重定向的路由应用程序。我们可以参照下边的代码来实现它：

1
2
3
4
5
6
7
8
9
10
11

复制代码// from piping-files-3.php

file_put_contents(
    "piping-files-3.jpeg ", file_get_contents(
        "https://github.com/assertchris/uploads/raw/master/rick.jpg "
    )
);

// ...or write this straight to stdout, if we don't need the memory info

require "memory.php ";

南宫冰郁 翻译于 5天前 0人顶 顶 翻译得不错哦！
设想一下，一个路由应用程序让我们看到这段代码。但是，我们想从CDN获取一个文件，而不是从本地的文件系统获取。我们可以用一些其他的东西来更好的替换file_get_contents（就像Guzzle），即使在引擎内部它们几乎是一样的。

图片的内存大概有581K。现在，让我们来试试这个

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

复制代码// from piping-files-4.php

$handle1 = fopen(
    "https://github.com/assertchris/uploads/raw/master/rick.jpg ", "r "
);

$handle2 = fopen(
    "piping-files-4.jpeg ", "w "
);

// ...or write this straight to stdout, if we don't need the memory info

stream_copy_to_stream($handle1, $handle2);

fclose($handle1);
fclose($handle2);

require "memory.php ";

内存使用明显变少（大概400K），但是结果是一样的。如果我们不关注内存信息，我们依旧可以用标准模式输出。实际上，PHP提供了一个简单的方式来完成：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

复制代码$handle1 = fopen(
    "https://github.com/assertchris/uploads/raw/master/rick.jpg ", "r "
);

$handle2 = fopen(
    "php://stdout ", "w "
);

stream_copy_to_stream($handle1, $handle2);

fclose($handle1);
fclose($handle2);

// require "memory.php ";

南宫冰郁 翻译于 5天前 0人顶 顶 翻译得不错哦！

其它流

还有其它一些流，我们可以通过管道来写入和读取（或只读取/只写入）：

• php://stdin (只读)

• php://stderr (只写, 如php://stdout)

• php://input (只读) 这使我们能够访问原始请求体

• php://output (只写) 让我们写入输出缓冲区

• php://memory 和 php://temp (读-写) 是我们可以临时存储数据的地方。 不同之处在于一旦它变得足够大 php://temp 会将数据存储在文件系统中，而 php://memory 将一直持存储在内存中直到资源耗尽。

  Tony 翻译于 5天前 0人顶 顶 翻译得不错哦！
  过滤器

还有一个我们可以在stream上使用的技巧，称为过滤器。它们是一种中间的步骤，提供对stream数据的一些控制，但不把他们暴露给我们。想象一下，我们会使用Zip扩展名来压缩我们的shakespeare.txt文件。

1
2
3
4
5
6
7
8
9
10

复制代码// from filters-1.php

$zip = new ZipArchive();
$filename = "filters-1.zip ";

$zip->open($filename, ZipArchive::CREATE);
$zip->addFromString("shakespeare.txt ", file_get_contents("shakespeare.txt "));
$zip->close();

require "memory.php ";

这是一小段整洁的代码，但它测量内存占用在10.75MB左右。使用过滤器的话，我们可以减少内存：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

复制代码// from filters-2.php

$handle1 = fopen(
    "php://filter/zlib.deflate/resource=shakespeare.txt ", "r "
);

$handle2 = fopen(
    "filters-2.deflated ", "w "
);

stream_copy_to_stream($handle1, $handle2);

fclose($handle1);
fclose($handle2);

require "memory.php ";

此处，我们可以看到名为php://filter/zlib.deflate的过滤器，它读取并压缩资源的内容。我们可以在之后将压缩数据导出到另一个文件中。这仅使用了896KB.

我知道这是不一样的格式，或者制作zip存档是有好处的。你不得不怀疑：如果你可以选择不同的格式并节省约12倍的内存，为什么不选呢？

为了解压此数据，我们可以通过执行另一个zlib filter将压缩后的数据还原：

1
2
3
4
5

复制代码// from filters-2.php

file_get_contents(
    "php://filter/zlib.inflate/resource=filters-2.deflated "
);

Streams have been extensively covered in Stream在“理解PHP中的流”和“U高效使用PHP中的流”中已经被全面介绍了。如果你喜欢一个完全不同的视角，可以阅读一下。

Tocy 翻译于 4天前 0人顶 顶 翻译得不错哦！

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

1、写在最前面

近期在处理一个批量表单上传的优化需求时，遇到一个innoDB的加锁问题，导致多线程在批量处理表单并插入数据库数据的时候出现LockTimeOutException等待锁超时异常。错误异常如下

1
2
3
4

复制代码org.springframework.dao.CannotAcquireLockException: could not
execute statement; SQL [n/a]; nested exception is
org.hibernate.exception.LockTimeoutException: could not execute
statement...

网上找了很多资料也比较零碎，通过自己对Mysql以及innodb的理解，结合Mysql官方文档对Mysql的锁以及加锁机制做一次简单的介绍。

2、事务的隔离级别

先来回顾一下事务的隔离级别：

• Read-Uncommitted:未提交读，可以读取到其他未提交的事务修改的数据。
• Read-Committed:提交读，只允许读取已提交事务的数据，可能导致不可重复读、幻读问题。
• RePeatable-Read:可重复读，同一个事务内多次相同的查询结果是一致的，可以解决重复读问题。
• serializable:串行化，最高隔离级别，每次都要获取表级锁，并发性能很差。

表注：不同的隔离级别可能导致的读取问题

查看数据库的隔离级别
mysql> select @@global.tx_isolation, @@session.tx_isolation, @@tx_isolation;

可看出Mysql的默认隔离级别是Repeatable-Read

3、InnoDB Locking

• 共享锁和排它锁 (shared locks、exclusive (X) locks)
• 意向锁 (Intention Locks)
• 记录锁 (Record Locks)
• 间隙锁 (Gap Locks)
• 自增锁 (AUTO-INC Locks)

3.1 共享锁和排它锁

InnoDB实现了两种标准的行级锁，共享锁(shared (S) locks) 和排他锁 (exclusive (X) locks).

• 共享锁允许一个占有锁的事务去读取一行数据
• 排它锁则允许事务对某一行记录进行写操作

如果一个事务持有了一个共享锁，其他事务仍然可以获取这行记录的共享锁，但不能获取到这行记录的排它锁。当一个事务获取到了某一行的排它锁，则其他事务将无法再获取这行记录的共享锁和排它锁。

分析一下上面出现错误的原因，当T1执行 update user set name='loy' where id = 1;时T1先获取到X锁由于T1没有执行commit; 所以他一直持有
这把锁当T2也执行更新操作的时候就会一直等待X导致获取锁超时。

3.2 意向锁

意向锁是一种表级锁，和基本锁一样也分为共享锁和排他锁。

• 意向共享锁 (Intention shared (IS)) 将要去获取某一行的共享锁。
• 意向排它锁 (Intention exclusive (IX)) 将要去获取某一行的排它锁。

事务在获取共享锁之前必须先获取意向共享锁，同理获取排他锁之前必须先获取意向排它锁，意向锁不会阻塞其他任何对表的操作，他只是告诉其他事务他将要去获取某一行的共享锁或者排他锁。

注：基本锁和意向锁兼容矩阵

3.3 记录锁

记录是是作用在索引上的一种锁，他锁住的是某一条记录的索引而非记录本身，如果当前表没有索引那么 InnoDB将会为其创建一个隐藏的聚合索引，而Record Locks将会锁住这个隐藏的聚合索引。

3.4 间隙锁

间隙锁和记录锁一样也是作用在索引上。不同的是记录锁只作用于一条索引记录而间隙锁可以锁住一个范围内的索引。举个例子:

SELECT id from t_user where id BETWEEN 5 AND 10 FOR UPDATE;
间隙锁将会在(4,5)和(10,11)之间加锁，他将阻止其他事务在这个范围内插入数据。

3.5 自增锁

自增锁是一种特殊的表级锁，他只作用在包含自增列的插入操作时。当一个事务正在插入一条数据时，其他的任何事务都必须等待整个事务完成插入操作，在取获取锁来执行插入操作。
官方文档是这样描述的：
An AUTO-INC lock is a special table-level lock taken by transactions inserting into tables with AUTO_INCREMENT columns. In the simplest case, if one transaction is inserting values into the table, any other transactions must wait to do their own inserts into that table, so that rows inserted by the first transaction receive consecutive primary key values.

看到这里我们在回到文章开头的地方，重新思考异常出现的原因。其实就很明了了。在需要保存数据的表中就包含有自动增长的列 id当多线在同时插入数据时就会导致 AUTO-INC Locks 锁竞争激烈，当线程等待锁的时间过长就会有上诉的异常抛出。

4、死锁分析

并发可能导致的死锁分析

数据库建表

1
2
3
4
5
6

复制代码CREATE TABLE `user` (
  `id` int(20) unsigned NOT NULL AUTO_INCREMENT,
  `name` varchar(11) DEFAULT NULL,
  `age` int(11) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;

看一下在T1 roolback之前各事务锁的占用情况

此时T1获取到的id=2这一行的X排它锁，T2和T3获取到的则是S共享锁，当我们把T1 rollback 以后T1将释放以获取到的X锁，此时的T2和T3都持有S锁，回顾S、X锁的兼容矩阵，因为S锁和X锁是冲突的，T2和T3都在等待对方释放S锁，导致死锁。

5、总结

尽管在日常开发中对InnoDB锁没有太多的接触，但是了解InnoDB加锁机制在遇到相关问题的时候会让你处理起来更得心应手。InnoDB还有其他一些比较有意思的锁感兴趣的同学可以移步Mysql官方文档学习。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有