如何写好一个单例

[TOC]

什么是单例模式？

• 如果你听说过设计模式，那么肯定知道单例模式，因为单例模式是设计模式中最简单的一种。顾名思义：单例模式就是一个类只有一个实例变量的一种设计模式，通过使用单例模式，可以节约系统的资源开销，避免共享资源的多重占用等优点。
• 什么时候会用：
  1. 对于那种经常实例化但是过一会儿就被销毁的对象适合使用单例模式。
  2. 对于创建对象需要消耗很多资源的对象。如：数据库连接池对象，线程池对象等
  3. 只需要一个对象保证全局的一致性的。如：Android中Application对象，网站的计数器等。

实现一个单例模式

• 如果你是一位对设计模式略有接触的新手，一定会毫不费力的就写出了以下单例代码（懒汉式单例：等到需要时再实例化）：

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

  /** * Created by forever on 2017/9/20. */ public class Singleton { private Singleton singleton; private Singleton() { } public Singleton getInstance() { if (singleton == null) { singleton = new Singleton(); } return singleton; } }

  其实上面的代码就已经涵盖了单例模式最重要的三个要素：

  1. 将构造方法私有化（保证外部不能直接构造）。
  2. 有一个静态属性指向实例
  3. 提供一个公有的静态方法向外面提供这个实例。

• 然而表面看似完美的代码，内部其实暗藏杀鸡：

  

  在单线程中看似是没有什么问题的，但如果放在多线程的环境中就会有问题了。加入有两个线程同时访问getInstance方法，如果期中一个线程刚进入if (singleton == null){}里面，这个时候另一个线程恰好也访问这个方法，并且完成创建了一个实例，那个刚刚挂起的那个线程继续运行的话就会再创建一个实例。那我们单例的理想不就破灭的了嘛。

  

基本方法的改进

• 既然了解了问题，那么我们如何才能防止两个线程同时实例化方法呢？有经验的同学或许就会立刻想到了Java的同步。通过synchronized关键字进行加锁。

  1 2 3 4 5 6

  public synchronized Singleton getInstance() { if (singleton == null) { singleton = new Singleton(); } return singleton; }

• 不过加锁的话对程序的性能性能会有很大影响，如果当某个线程正在访问该方法的时候其他线程就只能在锁池中等待该线程释放锁，我们稍加改进一下：

  1 2 3 4 5 6 7 8 9 10

  public Singleton getInstance() { if (singleton == null) { synchronized (Singleton.class){ if(singleton == null){ singleton = new Singleton(); } } } return singleton; }

  这样只在构造实例代码的时候加锁，对程序的性能影响就小多了。而且只要实例化完成之后，后面基本就不会进入这个同步代码块了。

• 看似已经很完美了，那么我们有什么其他办法不用加锁的方式也能避免多线程的问题呢？ok，当然有的，我们可以使用饿汉式的单例：

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

  /** * Created by forever on 2017/9/20. */ public class Singleton { public Singleton singleton = new Singleton(); private Singleton() { } public Singleton getInstance() { return singleton; } }

  上面代码与懒汉式加载最大的区别在于这里的single在开始就实例化了，也就是不管我们是否使用它，都会将其加载到内存中去。这个在获取的时候就直接返回就行了。如果不在意内存的话最好使用这个方法。

• 如果你说，我既不想要使用同步，但又十分在意内存资源怎么办，ok，说明你是一个很有追求的人，其实在也是有办法的（办法总比问题多）：

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

  public class Singleton { private Singleton(){ } public static Singleton getInstance(){ return Nested.singleton; } public static class Nested{ static { System.out.println("蛤蛤"); } private static Singleton singleton = new Singleton(); } }

  这个时候我们就需要一个内部类作为桥梁了，当我们getInstance()时，类加载器才会去加载Nested，然后实例化Singleton的实例，如果你对Java的类加载机制有了解的话一定很容易就理解了上述代码。

总结

• Java的单例模式看似简单，其实深究而言还是有很多值得深究的东西的，如果在面试的碰到了也可以和面试官多吹一会儿。最近一直在准备校招，作为菜鸟表示压力山大，也祝愿大家都能找到满意的工作。如果发现写的有什么问题欢迎指正，希望与大家共同进步。

经典算法——动态规划入门实例

[TOC]

神马是动态规划

• 专业定义：

  动态规划的本质，是对问题状态的定义和状态转移方程的定义。通过拆分问题，定义问题状态和状态之间的关系，使得问题能够以递推（或者说分治）的方式去解决。

  看完之后或许你一个字也没记住，这都是什么鬼？

  

• 关键点

  ok，让我们摒弃这些专业的条条框框，直奔主题。其实动态规划的核心就是拆解子问题——把一个大的问题拆解成逐步逐步的小问题，而这些小问题都可以直接由之前更小的问题得到。那么怎么拆解这些小问题呢？

  靠的就是状态的定义和状态转移方程的定义。

  当然，这样还是很抽象的。不如举个栗子：

  

求解最长公共子串

• 神马是最长公共子串？

  假定现在有两个字符串，最长公共子串就是它们之间公有的连续的最长子串，就像下图这样

  

  橙色部分就是最长的公有子串，记住这里一定要连续。那么问题来了？怎么求出任意两个字符串的最长公有子串呢？看似很简单，不过细思又感觉有点点复杂的感觉。

• 暴力求解

  简单粗暴的方法——先求出str1的所有子串，然后在求出str2的所有子串，然后再一个个比较，找到相同且最大的那个不就好了。ok，我们先来看看要多少次：

  str1的可能情况：

  1	n+(n-1)+(n-1)+.......+2+1 = ((n+1)*n)/2

  同理str2也是一样：

  1	m+(m-1)+(m-1)+.......+2+1 = ((m+1)*m)/2

  然后在通过两个for循环一个一个比较就搞定了嘛；

  1 2 3

  for i in range(((n+1)*n)/2): for i in range(((m + 1) * m) / 2): dosomthing()

  ok，这样不就搞定了，so easy，不过感觉有点细思极恐，这也太暴力了，万一字符串很大，这得算多久呀！心疼计算机3s。正所谓暴力膜法不可取，所以这时候我们的动态规划就登场啦。

• 最开始我们说过，用动态规划算法时最重要的就是要拆解子问题，那么我们如何将这个问题拆解为更小的子问题呢？根据动态规划的思想，我们首先要分析出如何根据已有的结果算出我们需要的结果，举栗来说：

  

  假如我们已经求好了截至到str1[i-1]与str2[j-1]处两个字符串的最大公有子串，是否可以帮助我们求出截止到str1[i]与str[j]处两个字符串的最大公有子串呢？思考一下：

  为了方便，我们将截止到（这里的截止到要包含str1[i]与str2[j]）str[i]与str[j]处的最大字符串长度记作lcs(i,j)，假如：str1[i] == str2[j]，那么我们直接在lcs(i-1，j-1)后面加1不就是lcs（i，j）了嘛，如果str1[i] !=str2[j]，最大公有子串到这里就结束了，所以经过这两个点没有最大公有子串，直接记作0就好了。

  用数学公式就是：

  1 2	f(m,n)=0 str1[m] != str2[n] ; f(m,n)=f(m-1,n-1) + 1 str[m]==str2[n];

  转化为代码就是：

  1 2 3 4

  if str1[i]==str2[j]: lcs[i][j] = lcs[i-1][j-1] + 1 else: lcs[i][j] = 0

  用一张图片描述整体流程就是：

  

  ok，既然弄清楚了计算的流程，就可以写出代码，完整代码如下：

  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

  def lcs(mes1, mes2): if len(mes1) > len(mes2): mes1, mes2 = mes2, mes1 data = [[0 for i in range(len(mes2) + 1)] for i in range(len(mes1) + 1)] mes1 = "$" + mes1 mes2 = "$" + mes2 max_lenth = 0 end = 0 for i in range(1, len(mes1)): for j in range(1, len(mes2)): if mes1[i] == mes2[j]: data[i][j] = data[i - 1][j - 1] + 1 if data[i][j] > max_lenth: max_lenth = data[i][j] end = i if max != 0: result = mes1[end - max_lenth + 1:end + 1] print(result) if __name__ == '__main__': while True: try: mes1 = input() mes2 = input() lcs(mes1, mes2) except: break

这里申请的二维数组比两个字符串都要大1，原因为我们呢将i，j为0的边都赋值为了0（当然，这里为了方便直接把数组都初始化了为0），有点像为我们的动态规划棋盘撒了半圈水雷一样。参考上面的流程图。这是一道经典的动态规划题目，在华为的笔试题里面曾经出现过，上面代码就是之前根据题目写的，已经通过了测试。原题如下：

求解最大公有子序列

• 神马是最大公有子序列

  乍一看，和最大公共子串就两字之差，其实这两个差的还是挺远的。还是上面那两个字符串：

  

  橙色部分就是他们的最大公有子序列。和子串最大的区别就是这里的子串可以不用连续了，所以叫最大子序列了。这种算法有什么用呢？

  比如：有次毛概老师让你写一篇5000字的论文，可是要交的前一天晚上你才想起来怎么办？当然是网上随便找一篇，但是你又怕老师发现是抄袭的，所以就会在文章开头改点东西，文章中间改点，文章结尾改点，当你以为万事大吉的时候，如果老师刚好学过这个算法，那么可以通过对比最大子序列找到文章的相似度了，然后就。。。。

• 暴力求解

  。。。。。。。。。。。。这个子集比之前的高多了，有兴趣朋友可以计算试试。。。。。。。

• 其实仔细想想，这个和最大子串其实是一样一样的，就是递推公式有一丢丢不同，希望大家稍微思考下，应该很快就写出来了。完整代码如下，建议对比之前代码就会发现差别了：

  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

  def lcs(mes1, mes2): if len(mes1) > len(mes2): mes1, mes2 = mes2, mes1 data = [[0 for i in range(len(mes2) + 1)] for i in range(len(mes1) + 1)] mes1 = "$" + mes1 mes2 = "$" + mes2 max_lenth = 0 end = 0 for i in range(1, len(mes1)): for j in range(1, len(mes2)): if mes1[i] == mes2[j]: data[i][j] = data[i - 1][j - 1] + 1 else: data[i][j] = max(data[i - 1][j], data[i][j - 1]) i = len(mes1) - 1 j = len(mes2) - 1 str = "" while i != 0 and j != 0: if mes1[i] == mes2[j]: str += mes1[i] i -= 1 j -= 1 else: if data[i][j - 1] > data[i - 1][j]: j -= 1 else: i -= 1 print(str[::-1]) if __name__ == "__main__": lcs("abcdacgw", "acdacvgwc")

  其实就是拷贝的前面代码，改了几行代码和输出而已。

总结

• 算法是一门很理论通过也很实践的学问，建议先搞清流程，再写代码加深理解。本人算法菜鸟，如果错误，欢迎一起探讨。共同学习进步。

六大排序算法图解以及python实现

[TOC]

冒泡排序——名字好听的排序

• 大学最先接触的一种排序算法，听名字就很萌的感觉，也是比较简单的一种，算法图示如下：

  

  上图为第一次进行比较的图示：

  • 从第一个开始每次都比较相邻的两个数，如果发现顺序不对，就把两个数交换一下，直到最后一个。这个时候，最大的数自然而然就跑到最后一位上面去了。

  • 第二次的时候，也从第一个开始，不过只需要循环到n-2处就行了（因为n-1处经过第一次洗礼已经时最大了嘛。）

    。。。。。。依此类推，循环n次，整个数组就会变成有序的了。

• 实现代码如下：

  1 2 3 4 5

  def bubble(nums): for i in range(len(nums)): for j in range(len(nums)-i-1): if nums[j+1] > nums[j]: nums[j+1], nums[j] = nums[j], nums[j+1]

选择排序——简单粗暴的排序

• 选择排序大概是所有排序中最好理解，最简单粗暴的排序了。算法图示如下：

  

  如图所示：

  1. 在第一次遍历中，从最开始到最后。找到最大的数，并于最后一个交换。

  2. 在第二次中，从最开始到倒数第二个。找到最大的数，并于倒数第二个交换。

     。。。。以此类推，循环n次。完成没有任何技术含量，所以效率几乎也是所有排序中最低的，强烈不推荐。

• 算法实现

  1 2 3 4 5 6 7

  def select_sort(nums): for i in range(len(nums)): flag = i for j in range(i, len(nums)): if nums[j] > nums[flag]: flag = j nums[i], nums[flag] = nums[flag], nums[i]

插入排序——优雅不要污

• 也是很简单的一种算法，同时也是后面的希尔排序的基础。如下图流程所示：

  

  基本思想：

  1. 首先明确，第一个数肯定是有序的呀（一个数难道还能无序吗？），然后从第二个开始，从后往前扫描，把这个数插入到合适的位置，比第二个数大的一次往后面移动。
  2. 第二个数移动正确的位置之后，前两个数就是有序的，依次把后面的数按照刚才的方法插入到合适的位置，整个数组就变成有序的了。

• 代码实现：

  1 2 3 4 5 6 7 8 9 10 11

  def insert_sort(nums): for i in range(1, len(nums)): temp = nums[i] last = 0 for j in range(i-1, -1, -1): if nums[j] > temp: nums[j + 1] = nums[j] else: last = j + 1 break nums[last] = temp

  ​

希尔排序——插入排序的升级版

• 希尔排序是插入排序的优化版，虽然表面上看起来并没有什么优化，相对于插入排序引入了步长的概念

  

  如图所示：上图就表示step为4的场景。这时候整数数组的数据按照步长分为如上四组，对每一组都分别进行插入排序。

  • 当完成第一轮插入排序之后，step的步长减半，进行第二轮插入排序。

    。。。。以此类推，直到步长step=1的那一轮排序完成，整个数组就可以变为有序了。

  不过问题来了：这最后一轮不和插入排序一毛一样么？这能提高效率：

  

  回答是：当然可以！其实每一次我们排序的时候整体都在逐渐趋于有序，都在不断的降低最后一步时的时间复杂度。在step合适的情况下，时间复杂度可以达到O（n^1.3），虽然我也不知道怎么算出来的，估计得问问数学家了。

快速排序——名字最牛逼的排序

• 大名鼎鼎的快排，听名字就无意中透漏着我是最屌的感觉。

  

• 快排的基本思想就是先找到一个数作为patition，将数组中比patition小的数放在其左边，比patition大的数放在其右边，然后对其每一边再分别进行相同的操作（递归实现）。

  

  上图是具体实现的思路：

  1. 我们一般已第一个数为partition，然后由两个标兵分别指向开始和结尾。
  2. 由最右边的橙色箭头开始往前扫描，直到扫描到比partition小的数字停下来，然后再由蓝色箭头向后扫描，直到遇见比partition大的数字停下来。然后交换停下来的两个数字。
  3. 就这样按照顺序，一直扫描。知道橙蓝两个箭头相遇，将相遇时的数字与partition交换就行了。

• 编程实现：

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

  def quick_sort(nums, start, end): if end < start: return 0 flag = nums[start] left = start right = end while start < end: while nums[end] > flag and end > start: end -= 1 while nums[start] <= flag and end > start: start += 1 nums[end], nums[start] = nums[start], nums[end] nums[left], nums[end] = nums[end], nums[left] quick_sort(nums, left, end - 1) quick_sort(nums, end + 1, right) def main(): nums = [6, 3, 10, 5, 1, 9, 2, 7, 23, 4, 14] quick_sort(nums, 0, len(nums) - 1) print(nums)

归并排序——分治法的完美践行者

• 归并排序是一种利用分治思想实现的排序，也是一种效率很高的排序。其排序流程具体如下：

  

  如图所示：

  • 在归并排序的时候，我们将数组不断的拆分为两半，直到数据只剩一个的时候，然后再按照大小顺序把他们拼装起来。

• 编码实现：

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

def merge_sort(nums):
    if len(nums) == 1:
        return nums
    mid = len(nums) // 2
    left = merge_sort(nums[:mid])
    right = merge_sort(nums[mid:])
    return merge(left, right)
def merge(left, right):
    temp = []
    left_p, right_p = 0, 0
    while left_p < len(left) and right_p < len(right):
        if left[left_p] < right[right_p]:
            temp.append(left[left_p])
            left_p += 1
        else:
            temp.append(right[right_p])
            right_p += 1
    temp += (left[left_p:])
    temp += (right[right_p:])
    return temp
def main():
    numbers = [6, 3, 10, 5, 45, 9, 2, 7, 23, 4, 14]
    numbers = merge_sort(numbers)
    print(numbers)

代码好像比之前的都要多一点，不过思路还是很简单的。主要是多了一个merge的过程：

• 对于分别有序的两个数组，通过每次取出一个数据比较，将较小的那个放到新的数组中，较大的那个等待下次继续比较，就这样一直取，比较，取，比较，直到两个数组的数都放在了新的数组中。返回新的数组。
• 大体的流程还是通过递归实现的，其实大部分分治的算法都是通过递归搞定的，方便理解。
  • 就效率而言，归并排序是六种排序中时间复杂度最好的，即使是最差情况，都是O(nlogn)，不过由于归并排序需要额外的空间，所以也是一种拿空间换时间的策略。

六大算法的效率比较

• 排序

  | 排序法 | 平均时间复杂度 | 最差情形 | 稳定度 | 额外空间 | 备注 |
  | ——- | ——– | ———– | —- | ——– | ——— |
  | 冒泡排序 | O(n2) | O(n2) | 稳定 | O(1) | n小时较好 |
  | 选择排序 | O(n2) | O(n2) | 不稳定 | O(1) | n小时较好 |
  | 插入排序 | O(n2) | O(n2) | 稳定 | O(1) | 大部分已排序时较好 |
  | Shell排序 | O(nlogn) | O(ns) 1<s<2 | 不稳定 | O(1) | s是所选分组 |
  | 快速排序 | O(nlogn) | O(n2) | 不稳定 | O(nlogn) | n大时较好 |
  | 归并排序 | O(nlogn) | O(nlogn) | 稳定 | O(1) | n大时较好 |

Hbase的架构简单解析

[TOC]

Hbase内部的基本组成

• Hbase的整体主要由zookeeper，Hmaster，HRegionServer，Hdfs文件系统组成。由这四部分共同完成数据的读取与写入。

• 我们知道，hbase有个很重要的特性，可以通过分布式集群存取海量数据，当数据量越来越多的时候通过增加集群主机的数量就可以将数据分散到不同的主机上，这是传统的数据库很难实现的，那么hbase是如何做到的呢？下面我们就来解析下hbase的基本架构原理。

Hbase数据的存储方式

• 对hbase有一点了解的童鞋都应该知道，hbase不同与传统的数据库，它是一种列式存储的数据库，对于里面的每一张表，每一行的数据都有一个rowkey，在数据库中的数据都是按照rowkey顺序排列的。所以可以将每张表的数据按照rowkey进行划分：

  不同的范围的数据划分到不同的地方（hbase中的这个地方被称为HRegion），不同的HRegion被放在不同的主机上，当查询数据的时候，只要根据rowkey先找到数据在那个范围的HRegion中，就可以直接到那个HRegion中找到数据，所以查询效率会比传统的数据库快很多。

  

  基本就像上图这样，随着数据不断增多，Region也会不断的增多（对表的划分增多），不同的Region由多个不同的RegionServer管理着。

Hbase读写数据的基本流程

• 写数据

  

  1. 首先要找到管理相应Region的RegionServer（但是如何找到RegionServer的呢？这个会在读取数据哪里解释），然后发送写数据的请求。
  2. 当RegionServer收到请求后，首先会将请求写入到HLog中（防止数据意外丢失，很多系统都是利用的这种机制）。
  3. 在Hlog写入完成之后，在找到相应的Region将数据写入。

  那么问题来了？在Region内部又是如何将数据写入到hdfs中的呢？这里就涉及到Region内部的数据持久化机制了：

  

  在Region内部，当数据操作请求由RegionServer传到Region的时候，Region中包含了很多的Store（每个Store对应了一个Table在这个Region中的一个Column Family，即每个Column Family就是一个集中的存储单元），写入数据根据具体在那个Column Family，将数据传到具体的Store中。数据传输流程如下：

  1. RegionServer首先会将数据写入到MemStore内存中，这时返回客户端通知说已经写入成功了。
  2. 当MemStore中的数据达到一定大小的时候，会有一次flush的过程，在此过程中，数据会被写入StoreFile文件中。然后StoreFile又会被序列化为HFile文件。（备注：在写入MemStore的时候会被插入到指定的顺序中，所以是有序的）
  3. HFile之后通过Hdfs的api上传到hdfs中。

• 读数据流程

  首先我们需要思考的是，要想读到数据，首先客户端肯定需要知道数据在哪吧。（写数据同样也是要先知道数据在哪），我们已经知道，hbase的数据都划分到了region中，那么我们如何找到相应的region呢？整体流程如下图：

  

  1. 首先我们要通过zookeeper找找到-root-的位置，这个root本质上也是一个region，不过里面存放的是不同范围的rowkey应该去那个RegionServer的.meta.中去查找
  2. 通过请求.meta.我们知道了我们需要的region在哪个RegionServer上，然后就可以去那个regionServer上去读数据了。

注1：这里其实有个问题，为啥不直接把数据的region所在的信息写在-root-中，这样不就少一个流程了。加快了访问速度吗？主要是为了保存的数据量考虑，由于Region的数量可能会有很多，而-root-只能由一个，所以防止-root-的内存被撑爆，所以多加了一层。

注2：不过一般公司的数据真的有这么多吗？应该是没有的吧，因而在HBase 0.96以后去掉了-ROOT- Table，只剩下这个特殊的目录表叫做Meta Table(hbase:meta)，它存储了集群中所有用户HRegion的位置信息，而ZooKeeper的节点中(/hbase/meta-region-server)存储的则直接是这个Meta Table的位置

Hmaster是干嘛的？

• 在上面的读写流程中，我们已经提到了HRegionServer，hdfs，zookeeper的作用了，但是一直都没有提到Hmaster，这个听上去很像老大的东西到底是干嘛的呢？难道只是打酱油的？Ofcource Not！

  

  之前我们虽然已经知道了数据是如何存入的，但是有一个问题一直没说，那就是hbase为啥可以管理海量数据呢？当数据越来越多的时候，是谁来负责将数据进行划分为更小的区域的呢？那就是一直在默默奉献额老大——HMaster。

• Hmaster是如何管理数据的呢？

  1. 清理冗余的数据

     当同一region的数据块达到四块的时候，Hmaster会对其进行一次数据合并，清理期中的冗余数据。合并之后的数据会小于256M。

  2. HRegion Split

     • 最初，一个Table只有一个HRegion，随着数据写入增加，如果一个HRegion到达一定的大小，就需要Split成两个HRegion，这个大小由hbase.hregion.max.filesize指定。这个Split过程就是由Hmaster完成的。
     • 当split时，两个新的HRegion会在同一个HRegionServer中创建，它们各自包含父HRegion一半的数据，当Split完成后，父HRegion会下线，而新的两个子HRegion会向HMaster注册上线，处于负载均衡的考虑，这两个新的HRegion可能会被HMaster分配到其他的HRegionServer中。

  3. 当HRegionServer宕机之后，hmaster可以读取hlog，将数据在发送给其他RegionServer进行数据恢复。并修改.meta.

  注：RegionServer不管理Hbase的元数据，元数据由Zookeeper管理。

总结一下

• Hbase中各个主要部分的功能：

  

  1. Hmater

     • 在Region Split后，负责新Region的分配；
     • 新机器加入时，管理HRegion Server的负载均衡，调整Region分布
     • 在HRegion Server宕机后，负责失效HRegion Server 上的Regions迁移。

  2. Region Server

     • Region server维护Master分配给它的region，处理对这些region的IO请求
     • HRegion Server管理了很多table的分区，也就是region。

  3. zookeeper

     • ZooKeeper为HBase集群提供协调服务，它管理着HMaster和HRegionServer的状态(available/alive等)，并且会在它们宕机时通知给HMaster
     • zookeeper中管理着hbase的元数据，例如-root-的位置所在。

  4. hdfs

     • 数据文件的存放处。由于其本身的分布式存储机制，所以数据文件很安全。
     • hadoop的datanode最好和region在同一主机上，方便读取数据。尽量避免网络数据传输。

Hbase入门

[TOC]

Hbase是干嘛的？

• HBASE是一个高可靠性、高性能、面向列、可伸缩的分布式存储系统。其主要作用是通过分布式的集群系统进行海量数据的存储与处理。其主要的优点有：
  1. 线性扩展，随着数据量增多可以通过节点扩展进行支撑
  2. 数据存在hdfs上面，备份机制健全
  3. 通过zookeeper协调查找数据，读取速度快。
  4. 良好的数据备份与恢复机制，数据安全性由保障。

Hbase的安装与配置

1. 下载安装包并解压，然后修改环境变量

2. 修改hbase-env.sh配置文件

   1 2 3 4 5 6 7 8

   # The java implementation to use. Java 1.7+ required. export JAVA_HOME=/usr/local/soft/jdk1.8.0_144 # Extra Java CLASSPATH elements. Optional. export HBASE_CLASSPATH=/usr/local/soft/hbase/conf export JAVA_CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar export HBASE_OPTS="-XX:+UseConcMarkSweepGC" # 告诉habse用我们自己配置的zookeeper export HBASE_MANAGES_ZK=false

3. 修改hbase-site.xml

   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

   <configuration> <property> <name>hbase.master</name> <value>mini1:60000</value> </property> <property> <name>hbase.master.maxclockskew</name> <value>180000</value> </property> <property> <name>hbase.rootdir</name> <value>hdfs://mini1:9000/hbase</value> </property> <property> <name>hbase.zookeeper.property.clientPort</name> <value>2181</value> <description>Property from ZooKeeper's config zoo.cfg.</description> </property> <!--是否采用分布式策略--> <property> <name>hbase.cluster.distributed</name> <value>true</value> </property> <!--zookeeper所在位置--> <property> <name>hbase.zookeeper.quorum</name> <value>mini5,mini6,mini7</value> </property> <property> <name>hbase.zookeeper.property.dataDir</name> <value>/home/hadoop/hbase/tmp/zookeeper</value> </property> </configuration>

4. 启动hbase

   1	start-hbase.sh

5. 打开shell交互式命令行

   1 2	hbase shell # web管理打开地址 http://mini1:16030

常见错误

1. 发现ERROR: org.apache.hadoop.hbase.ipc.ServerNotRunningYetException: Server is not running yet，一般是hdfs正处于安全模式，可以直接关闭（感觉不太好）:

   1	hadoop dfsadmin -safemode leave

   ​

基本数据模型

1. Row Key

   key是用来检索记录的主键,，通过row key来定位数据。是最重要的数据模型，当插入数据的时候，数据会按照row key进行排序，插入到合适的位置。

2. Columns Family

   HBASE表中的每个列，都归属于某个列族。列族是表的schema的一部 分(而列不是)，必须在使用表之前定义。列名都以列族作为前缀。例如 courses:history，courses:math都属于courses 这个列族。

3. Cell

   由{row key, columnFamily, version} 唯一确定的单元。cell中 的数据是没有类型的，全部是字节码形式存贮。

4. TimeStamp

   HBASE中通过rowkey和columns确定的为一个存贮单元称为cell。每个 cell都保存 着同一份数据的多个版本。版本通过时间戳来索引。时间戳的类型是 64位整型。时间戳可以由HBASE(在数据写入时自动 )赋值。

Hbase的常用命令

基于netty的rpc框架

[TOC]

如果你已经对以下东东有所了解，那么你就可以完成一个rpc框架了

• Java的反射技术
• java的动态代理机制
• 基于nio的框架netty
• 全世界最好的框架-spring
• java的序列化

神马是rpc？

• 在这个大数据时代，很多公司的服务器都是以集群的方式存在的。在我们传统的mvc后台开发中，我们就需要把不同层的服务部署到不同的服务器上面，这个每个服务器的的压力就会比较小了。

  但是这样也会带来一个问题——我在这台机器上如何才能调用到另一台机器的代码呢？这是个问题。

  我们先来举个栗子：

  

  比如我们在一个传统mvc项目中，我们有一个UserController处理用户的请求，假如它长的这个样子：

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

  @RestController @RequestMapping(value = "/user") public class UserController { @Autowired UserService userService; @RequestMapping(value = "/current") public AjaxResponse login(String name,String password) { return AjaxResponse.success(userService.login(name,password)); } }

  正常情况下，这个UserService的实现肯定是在同一个项目或者是本地的，早就已经被加入到spring容器中了，不过加入我们为了减少服务器的压力，我们将UserService的实现放到另一台服务器上，加入我们有一个膜法，可以在本地的Controller像调用本地方法一样调用另一台的userServiceImpl就好了，Rpc就是这样一种技术。

实现思路

• 表面上看这是一个很难完成的任务，本机怎么可能可以调用到远程的方法呢？不过如果我们这个任务拆分开来，就会发现只要一步一步来，其实还是挺简单的。

  我们可以换一种思路，既然直接调用不行，我们可以曲线救国呀，我们只要把调用方法的对象的名称，方法的名字，方法的参数与方法的类型都通过网络发送到另一台机器上，另一条机器接收到之后根据请求信息调用该对象的方法，然后在把执行结果通过网路直接返回回来不就ok了。其实，rpc框架的大体思路就是如此。

• 大体实现流程如下：

  1. 通过java的动态代理机制为我们UserService创建代理对象，在代理对象执行方法的时候实际上已经被我们定制的方法拦截。
  2. 在拦截的逻辑里面，我们在获取到调用的方法的所有接口，方法名，参数集合，参数类型集合后封装到一个JavaBean——request中去，然后我们将这个对象序列化之后通过网络传输到另一台机器上。
  3. 另一台机器接受到这个网络请求后，将数据反序列化为Request对象，从而了解我们请求的是具体是什么对象的什么方法，然后服务器通过反射的方式调用，并将执行结果通过另一个JavaBean——Response返回。
  4. 本机收到服务端的返回。整个rpc调用就完成了。

• 如下图所示，由于画图水平有限，不过大致就是这个意思：

  

代码具体实现

1. 首先我们需要为我们的网络请求分装两个JavaBean，分别为Request与Response.。

   1 2 3 4 5 6 7 8 9 10

   //在Request中应有的属性 private String requestId; private String className; private String methodName; private Class<?>[] parameterTypes; private Object[] parameters; //在response应该有的属性 private String requestId; private Throwable error; private Object result;

2. 创建RpcClient，封装我们的网络请求流程。其中最重要的是这个方法：

   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

   public Response send(Request request) throws InterruptedException { ClientBootstrap bootstrap = new ClientBootstrap(); ExecutorService boss = Executors.newCachedThreadPool(); ExecutorService work = Executors.newCachedThreadPool(); bootstrap.setFactory(new NioClientSocketChannelFactory(boss,work)); bootstrap.setPipelineFactory(new ChannelPipelineFactory() { @Override public ChannelPipeline getPipeline() throws Exception { ChannelPipeline pipeline = Channels.pipeline(); pipeline.addLast("decoder",new ResponseDecoder()); pipeline.addLast("encoder",new RequestEncoder()); pipeline.addLast("handler",RpcClient.this); return pipeline; } }); ChannelFuture connect = bootstrap.connect(new InetSocketAddress(address, port)).sync(); connect.getChannel().write(request).sync(); //阻塞线程直到完成请求或者请求失败 synchronized (obj){ obj.wait(); } connect.getChannel().close().sync(); return this.response; }

   这里用netty3进行的网咯请求，这里ResponseDecoder与RequestEncoder是对Response与Request进行的序列化与反序列化，采用的谷歌的Protostuff序列化框架实现（为啥不用java自带的序列化工具呢？因为java自定的序列化附带了很多其他信息，序列化的字节长度比谷歌的长好几倍，所以是为了节约带宽，同时Protostuff的序列化支持多种编程语言）

3. 创建代理的工具类，返回代理对象。

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

   public <T>T proxy(Class<?> clazz){ return (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class<?>[] { clazz }, new InvocationHandler() { public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { Request request = new Request(); request.setClassName(method.getDeclaringClass().getName()); request.setMethodName(method.getName()); request.setParameters(args); request.setRequestId(UUID.randomUUID().toString()); request.setParameterTypes(method.getParameterTypes()); RpcClient client =new RpcClient(address,port); //通过封装的网络框架进行网络请求 Response response = client.send(request); if (response.getError()!=null){ throw response.getError(); } else{ return response; } } }); }

4. 服务端在开启服务的时候就需要通过spring扫描所有的service实现类，将其装进spring的容器中。

   1 2 3 4 5 6 7 8 9

   public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { Map<String, Object> beansWithAnnotation = applicationContext.getBeansWithAnnotation(RPCService.class); for(Map.Entry<String,Object> entry :beansWithAnnotation.entrySet()){ String interfaceName = entry.getValue().getClass() .getAnnotation(RPCService.class).value().getName(); serviceMap.put(interfaceName,entry.getValue()); } startServer(); }

   需要发布的服务类都需要使用@RPCService注解，这是一个自定义的注解。

5. 在服务端收到客户端的网络请求之后，我们就需要从spring容器中找到请求的服务类完成调用并返回执行结果。

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

   @Override public void messageReceived(ChannelHandlerContext ctx, MessageEvent event) throws Exception { Request request = (Request) event.getMessage(); Response response = new Response(); //调用请求类的请求方法执行并返回执行结果 Object invoke = null; try { Object requestBean = serviceMap.get(request.getClassName()); Class<?> requestClazz = Class.forName(request.getClassName()); Method method = requestClazz.getMethod(request.getMethodName(), request.getParameterTypes()); invoke = method.invoke(requestBean, request.getParameters()); response.setRequestId(UUID.randomUUID().toString()); response.setResult(invoke); } catch (Exception e) { response.setError(e); response.setRequestId(UUID.randomUUID().toString()); } System.out.println(request+""+response); //返回执行结果 ctx.getChannel().write(response); }

总结

• 整体的流程还是比较简单的，就是具体实现的时候会有一些细节问题需要好好处理。虽然是第一次写这种轮子程序，不过感觉还是不错的。完整代码已上传到我的GitHub仓库里面，有兴趣的小伙伴可以去看看。

通过zookeeper完成动态感知分布式服务器上下线的功能

[TOC]

业务描述

• 某分布式系统中，主节点可以有多台，可以动态上下线，任意一台客户端都能实时感知到主节点服务器的上下线。当主节点下线的时候，客户端会收到通知，并更新目前在还在线的主节点host信息，可以防止客户端向已经挂掉的节点进行请求。

服务器端的实现

• 要想完成上述功能，在我们可以想到通过zookeeper的短暂态的节点完成，在服务器启动的时候，我们连接到zookeeper并向其注册一个短暂态的节点，当服务器因为某种意外宕机的时候，这个节点也会被删除，这样客户端访问所有的注册节点的信息，就是仍然在正常工作的主节点。

• 服务器端的代码如下：

  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

  import java.io.IOException; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooDefs.Ids; import org.apache.zookeeper.ZooKeeper; public class ClusterServer { //zookeeper的ip与端口，这里是zookeeper是一个集群 private String ZK_SERVER_LIST = "mini1:2181,mini2:2181,mini3:2181"; //设置超时时间，当主节点2s后没反应就认该节点已经挂了 private static final int sessionTimeout = 2000; //所有的注册节点信息当道/servers下面，方便管理 private String SERVER_DIR = "/servers"; private ZooKeeper zk; public void connect() throws IOException { zk = new ZooKeeper(ZK_SERVER_LIST, sessionTimeout, new Watcher() { //当zookeeper服务器集群有断线时会调用 @Override public void process(WatchedEvent event) { // TODO Auto-generated method stub System.out.println(event.toString()); } }); } public void register(String hontname) throws IOException, KeeperException, InterruptedException { //创建瞬时态且带序号的节点，这样在节点下线后该节点就会被删除 String result = zk.create(SERVER_DIR+"/server", hontname.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); System.out.println("创建结果："+result); } public void diy(){ System.out.println("just do it"); try { Thread.sleep(Integer.MAX_VALUE); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } public static void main(String[] args) throws IOException, KeeperException, InterruptedException { ClusterServer server =new ClusterServer(); server.connect(); server.register(args[0]); server.diy(); } }

• 程序启动的时候就向服务器把当前主节点的host信息注册到zookeeper中。

• 这里的hostname本来可以通过java api进行动态获取的，不过为了方便实验就省略了。

客户端的实现

• 客户端所需要的工作是获取正常工作的主节点并且当主节点发生变化的时候可以收到信息，获取最新的正常工作的主节点。所以我们可以对servers下面的子节点信息进行监听，

• 代码实现

  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

  import java.io.IOException; import java.util.ArrayList; import java.util.List; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooKeeper; public class ClusterClient { private String ZK_SERVER_LIST = "mini1:2181,mini2:2181,mini3:2181"; private static final int sessionTimeout = 2000; private String SERVER_DIR = "/servers"; private ZooKeeper zk; private List<String> servers=new ArrayList<>(); public void connect() throws IOException { zk = new ZooKeeper(ZK_SERVER_LIST, sessionTimeout, new Watcher() { @Override public void process(WatchedEvent event) { System.out.println(event.toString()); try { //收到通知后再次对服务器设置监听。 getServerList(); System.out.println(servers); } catch (KeeperException | InterruptedException e) { } } }); } //获取服务器列表并设置监听 public void getServerList() throws KeeperException, InterruptedException{ List<String> nodeList = zk.getChildren(SERVER_DIR,true); servers.clear(); for(String node:nodeList){ byte []data = zk.getData(SERVER_DIR+"/"+node, false, null); String hostName = new String(data); servers.add(hostName); } } //做自己爱做的事同时完成守护进程的功能 public void diy(){ new Thread(new Runnable() { @Override public void run() { // TODO Auto-generated method stub while (true){ } } }).start();; } public static void main(String[] args) throws IOException, KeeperException, InterruptedException { ClusterClient client =new ClusterClient(); client.connect(); client.getServerList(); client.diy(); } }

测试

• 我们可以在多个终端分别运行服务器程序，用来模拟多个主节点，直接关闭终端模仿主节点的宕机。

• 实验结果如下（当我们动态增加节点的结果）：

  

• 所有需要的jar包在zookeeper的解压包里面都有。或者直接在我的github里面有代码以及jar包。

  ​

zookeeper入门

[TOC]

zookeeper的简介

顾名思义，zookeeper就是动物园管理员的意思，Zookeeper是一个分布式协调服务；就是为用户的分布式应用程序提供协调服务。其主要具有一下特点：

1. zookeeper是为别的分布式程序服务的
2. Zookeeper本身就是一个分布式程序（只要有半数以上节点存活，zk就能正常服务）
3. Zookeeper所提供的服务涵盖：主从协调、服务器节点动态上下线、统一配置管理、分布式共享锁、统一名称等服务。

虽然说可以提供各种服务，但是zookeeper在底层其实只提供了两个功能：

1. 管理(存储，读取)用户程序提交的元数据（不是用户程序的业务数据）；
2. 并为用户程序提供数据节点监听服务；（监听因故宕机的程序，并通知其他应用程序接替工作）。

zookeeper的特性

1. Zookeeper：一个leader，多个follower组成的集群

2. 全局数据一致：每个server保存一份相同的数据副本，client无论连接到哪个server，数据都是一致的

3. 分布式读写，更新请求转发，由leader实施

4. 更新请求顺序进行，来自同一个client的更新请求按其发送顺序依次执行

5. 数据更新原子性，一次数据更新要么成功，要么失败

6. 实时性，在一定时间范围内，client能读到最新数据

zookeeper环境搭建

1. 拷贝安装包到对应服务器上。

2. 解压缩安装包

   1 2	tar -zvxf zookeeper-3.4.5.tar.gz mv zookeeper-3.4.5 zookeeper #修改文件夹名称

3. 配置环境变量

   1 2 3 4 5 6

   vim /etc/profile #加入以下信息 export ZOOKEEPER_HOME=/opt/zookeeper export PATH=$PATH:$JAVA_HOME/bin:$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$HIVE_HOME/bin:$ZOOKEEPER_HOME/bin #添加之后 source /etc/profile

4. 修改conf文件夹下的配置文件（复制zoo_sample.xml为zoo.xml），添加如下内容

   1 2 3 4 5

   dataDir=/root/zookeeper/data dataLogDir=/root/zookeeper/log server.1=mini1:2888:3888 #其余节点的主机名，以及通信的端口和检测连接的端口 server.2=mini2:2888:3888 server.3=mini3:2888:3888

5. 在前面定义的dataDir目录下新建muid文件，加入内容1，之后配置的节点依次递增。

6. 启动服务（其余节点需要关闭防火墙）

   1 2	zkServer.sh start #启动服务 zkServer.sh status #查看服务状态（主从节点信息）

   ​

Zookeeper的数据管理功能

• 在zookeeper中默认通过树的结构保存用户的元数据

  ​

  

  每一个节点都可以存放数据，类似与文件系统的树形存储目录。

• 通过命令行连接到zookeeper服务器。

  通过解压目录的bin/zkCli.sh可以直接启动命令行的客户端。

• 在zookeeper中每个节点的数据可以分为两类：

  1. 瞬时节点（ephemeral）（当连接进程与zookeeper断开连接的一定时间后（心跳时间）就会删除节点）
  2. 持久节点（persistent）（即时连接进程与zookeeper断开也不会删除该节点）

  而且每种节点又分为两种，有序节点与无序节点（有序节点在创建时会默认在其后添加一个递增的序号）。所以总共在zookeeper中包括了四种节点。

  • PERSISTENT
  • PERSISTENT_SEQUENTIAL（持久序列/test0000000019 ）
  • EPHEMERAL
  • EPHEMERAL_SEQUENTIAL

通过客户端对元数据进行管理

• 通过bin/zkCli.sh我们通过客户端连接到zookeeper，我们可以在命令行对节点进行增删改查操作。

• 通过help命令我们可以对查看命令及其使用方法。

  

  值得注意的是在ls与get命令后面我们可以对某个数据增加监控，当某个数据发生变化的时候连接进程后收到信号。只能监听一次，完成后后面的事件不会再触发监听。

  1 2 3 4

  get /test/test1 true #对/test/test1增加监听，当其数据发生变化的时候会收到信息 # WatchedEvent state:SyncConnected type:NodeDataChanged path:/test/test1 ls /test true #对/test增加监听，子目录下内容发生变化的时候会收到信息。 # WatchedEvent state:SyncConnected type:NodeDeleted path:/test/test1

  ​

python搭建简单的静态web服务器

[TOC]

储备知识

• 一丢丢的python（io和多线程的知识）
• 一丢丢的http协议
• 一丢丢的tcp/ip协议（当然不了解也没关系）
• 一丢丢的正则表达式知识

web服务器基本原理

• 当在浏览器的地址栏输入一个ip与端口之后，浏览器就会通过tcp/ip协议与相应的主机端口进程建立联系。经历过三次握手之后就会将http请求发送到相应的服务器进程去，之前我们了解的http协议在服务进程收到的其实就是一串有特殊格式的字符串。

  当我们浏览器输入localhost:9876 后服务进程实际收到的如下：

  

大致流程

1. 在服务端建立tcp服务进程，为了保证服务端可以同时处理多个请求，我们需要在每接受一个请求后为其单独使用一个线程（或者进程）为其进行服务。

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

   server = socket(AF_INET, SOCK_STREAM) server.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) address = ('', 9876) server.bind(address) server.listen(10) try: while True: print("-------等待接受服务----------") client, client_address = server.accept() print("-------接受服务成功----------") # 如果使用进程服务，可以在在后面把client关闭。 p = Thread(target=deal_socket, args=(client,)) p.start() except Exception as e: print(e) finally: server.close() print("-------服务结束----------")

   这里的deal_socket函数就是我们为一个请求服务的函数，在一个单独的线程中运行。

2. 在处理url请求的函数中，我们需要读取出客户端的http请求。

   1 2 3 4 5 6 7 8 9 10 11

   def deal_socket(client): print("-------开启新的线程----------") try: data = client.recv(1024) if len(data) > 0: fileName = get_request_name_from_http(data.decode("utf-8")) writeHtml(client, fileName) finally: client.close() print("-------关闭新的线程----------")

   • 在这里的data就是我们读取到的http服务请求，当其长度等于0时代表客户端已经关闭了tcp连接。
   • 这里的get_request_name_from_http()需要我们解析出请求的静态资源
   • 这里的writeHtml()将静态文本写回到客户端。

3. 在get_request_name_from_http函数中我们需要从原始的url请求中解析出http请求中我们需要的请求资源部分，这里我们可以通过正则表达式完成简单的完成解析。

   1 2 3 4 5 6 7

   def get_request_name_from_http(http): # 注意这里通过非贪婪模式匹配 r = re.search(r"GET /(.+?) ", http) fileName = "" if r != None: fileName = r.group(1) return fileName

   • 请求的http大概格式是这样（当我们访问http://localhost:9876/html/index.html时）

     1 2 3 4 5 6

     GET /html/index.html HTTP/1.1 Host: localhost:9876 Connection: keep-alive Cache-Control: max-age=0 Upgrade-Insecure-Requests: 1 User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/59.0.3071.109 Safari/537.36

     第一行就是我们请求的静态资源，我们将其通过非贪婪模式的正则表达式扣出来。

4. 在解析到请求的静态地址后就是简单的读取请求的文件，然后已http协议的格式返回回去就行了。

   1 2 3 4 5 6 7 8 9 10 11

   def writeHtml(client, fileName): rspHead = None rspBody = None if not os.path.exists(fileName): rspHead = "HTTP/1.1 404 error\r\nServer: foreverServer\r\n\r\n" rspBody = "file not found" else: rspHead = "HTTP/1.1 200 OK\r\nServer: foreverServer\r\n\r\n" html = open(fileName, 'r', encoding='UTF-8') rspBody = html.read() client.send((rspHead + rspBody).encode("utf-8"))

   • 当请求的静态文件不存在时，将返回给客户端文件不存在。

   • 上面的相应格式是根据http相应报文的格式而定的，否则浏览器会不识别：

     ​

     

     在Windows中\r\n分别代表回车和换行，而现在在unix系统中\n就代表了回车换行。

     ​

     完整代码

• 下面是完整的服务代码，不到60行就可以完成一个简单的静态web服务器，这就是python的魅力：

  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

  from socket import * from threading import Thread import os import re def get_request_name_from_http(http): r = re.search(r"GET /(.+?) ", http) fileName = "" if r != None: fileName = r.group(1) return fileName def writeHtml(client, fileName): rspHead = None rspBody = None if not os.path.exists(fileName): rspHead = "HTTP/1.1 404 error\r\nServer: foreverServer\r\n\r\n" rspBody = "file not found" else: rspHead = "HTTP/1.1 200 OK\r\nServer: foreverServer\r\n\r\n" html = open(fileName, 'r', encoding='UTF-8') rspBody = html.read() client.send((rspHead + rspBody).encode("utf-8")) def deal_socket(client): print("-------开启新的线程----------") try: data = client.recv(1024) if len(data) > 0: fileName = get_request_name_from_http(data.decode("utf-8")) writeHtml(client, fileName) finally: client.close() print("-------关闭新的线程----------") def main(): server = socket(AF_INET, SOCK_STREAM) server.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) address = ('', 9876) server.bind(address) server.listen(10) try: while True: print("-------等待接受服务----------") client, client_address = server.accept() print("-------接受服务成功----------") # 就这里和多线程不同，并且千万不能把client关掉 p = Thread(target=deal_socket, args=(client,)) p.start() except Exception as e: print(e) finally: server.close() print("-------服务结束----------") if __name__ == "__main__": main()

• 到此就用python构建了史上最挫的静态wen服务器了，直接在浏览其输入静态html请求就可以显示网页了：

  

  虽然很挫，不过web服务器的基本原理就是如此，牛逼的服务器也只是在这之上做了很多完善，下一篇我们将采用python提供的WSGI标准完成一个动态的web框架。

java运行时内存管理

• Java运行时的数据区
  

• 各个部分的数据简介

  1. 程序计数器（线程私有）

     是一块较小的内存空间，它可以看作是当前线程所执行的字节码的行号指示器。字节码解释器工作时就是通过改变这个计数器的值来选取下一条需要执行的字节码指令，分支、循环、跳转、异常处理、线程恢复等基础功能都需要依赖这个计数器来完成。

  2. Java虚拟机栈（线程私有，生命周期与线程相同）

     由一系列栈帧构成，用于存储局部变量表、操作数栈、动态链接、方法出口等信息。每一次的方法调用创建一个帧，并压栈。

  3. 本地方法栈

     本地方法栈（Native Method Stack）与虚拟机栈所发挥的作用是非常相似的。虚拟机栈为虚拟机执行Java方法（也就是字节码）服务，而本地方法栈则为虚拟机使用到的Native方法服务。

  4. Java堆（与程序开发最密切）

     Java堆（Java Heap）是Java虚拟机所管理的内存中最大的一块。Java堆是被所有线程共享的一块内存区域，在虚拟机启动时创建。此内存区域的唯一目的就是存放对象实例

  5. 方法区（保存装载的类信息。）

     方法区（Method Area）与Java堆一样，是各个线程共享的内存区域，它用于存储已被虚拟机加载的类信息、常量、静态变量、即时编译器编译后的代码等数据。

• Java对象的创建过程

  虚拟机遇到一条new指令时，首先将去检查这个指令的参数是否能在常量池中定位到一个类的符号引用，并且检查这个符号引用代表的类是否已被加载、解析和初始化过。如果没有，那必须先执行相应的类加载过程，接下来虚拟机将为新生对象分配内存。

  举个栗子：下图是对下面执行代码的图解

  

  1 2 3 4 5 6 7 8 9

  public class AppMain { // 运行时, jvm 把appmain的信息都放入方法区 public static void main(String[] args){ //main 方法本身放入方法区。 Sample test1 = new Sample( " 测试1 " ); //test1是引用，所以放到栈区里， Sample是自定义对象应该放到堆里面 test1.printName(); } }

  ​

• Jvm的内存模型（重点）

  –每一个线程有一个工作内存和主存独立

  –工作内存存放主存中变量的值的拷贝

  

  • 当数据从主内存复制到工作存储时，必须出现两个动作：第一，由主内存执行的读（read）操作；第二，由工作内存执行的相应的load操作；
  • 当数据从工作内存拷贝到主内存时，也出现两个操作：第一个，由工作内存执行的存储（store）操作；第二，由主内存执行的相应的写（write）操作。

  每一个操作都具有原子性，不会中断。由于工作内存的存在，所以在另一个线程中的变量值被修改其他线程使用该变量时并不会得到最新的变量值，可以可以通过Volatile关键字使得在其他线程中立即可见。