update_repo.sh用于识别新的提交并通知监听器。它首先记录当前所在的提交ID，然后拉取最新的代码，接着检查最新的提交ID。如果当前的版本ID与最新的匹配，说明代码没有变动，所以监听器不会作出任何响应。但是，如果提交ID间存在不同，就意味着有新了新的提交。这时，update_repo.sh将创建一个名为.commit_id的文件来记录最新的提交ID。

update_repo.sh的细分步骤如下：首先，我们的脚本会执行run_or_fail.sh文件，该文件提供了一些基于shell脚本的辅助函数。通过这些函数我们可以运行指定的脚本并可以在运行出错时输出错误信息。

!/bin/bash

source run_or_fail.sh

接下来，我们的脚本会试图删除.commit_id文件。 因为repo_observer.py会再无限循环中不断的调用updaterepo.sh ，如果在上一次的调用中产生了.commit_id 文件，并且其中储存的版本ID我们在上一次轮询中已经完成了测试，就会造成混乱。所以我们在每次都会先删除上一次的.commit_id文件，以免产生混乱。

bash rm -f .commit_id

在删除了文件之后（在文件已经存在的情况下），脚本会检查我们监听的代码库是否存在，再把.commit_id更新到最近的一次提交，保证.commit_id文件与代码库提交ID之间的同步。

run_or_fail "Repository folder not found!" pushd $1 1> /dev/null run_or_fail "Could not reset git" git reset --hard HEAD

再之后，读取git的日志，将其中最后一次的提交ID解析出来。

COMMIT=$(run_or_fail "Could not call 'git log' on repository" git log -n1) if [ $? != 0 ]; then echo "Could not call 'git log' on repository" exit 1 fi COMMIT_ID=echo $COMMIT | awk '{ print $2 }'

接下来，拉取代码库，获取最近所有的更改，并得到最新的提交ID。

run_or_fail "Could not pull from repository" git pull COMMIT=$(run_or_fail "Could not call 'git log' on repository" git log -n1) if [ $? != 0 ]; then echo "Could not call 'git log' on repository" exit 1 fi NEW_COMMIT_ID=echo $COMMIT | awk '{ print $2 }'

最后，如果新得到的提交ID与上一次的ID不匹配，我们就知道在两次轮询间发生了新的提交，所以我们的脚本应该将新的提交ID储存在.commit_id文件中。

if the id changed, then write it to a file

if [ $NEW_COMMIT_ID != $COMMIT_ID ]; then popd 1> /dev/null echo $NEW_COMMIT_ID > .commit_id fi

当repo_observer.py运行update_repo.sh完之后，监听器会检查.commit_id是否存在。如果文件存在，我们就知道在上一次的轮询后又发生了新的提交，我们需要通知测试样例调度器来开始测试。监听器会通过连接并发送一个'status'请求来检查调度器服务的运行状态，以保证它处在可以正常接受指令的状态正常工作状态。

      if os.path.isfile(".commit\_id"):
          try:
              response = helpers.communicate(dispatcher\_host,
                                             int(dispatcher\_port),
                                             "status")
          except socket.error as e:
              raise Exception("Could not communicate with dispatcher server: %s" % e)

如果它以“OK”响应，则监听器会打开.commit_id文件，读取最新的提交ID并使用dispatch:请求将该ID发送给调度程序，监听器会每个5秒发送一次指令。如果发生任何错误，监听器同样会每隔5s进行一次重试。

          if response == "OK":
              commit = ""
              with open(".commit\_id", "r") as f:
                  commit = f.readline()
              response = helpers.communicate(dispatcher\_host,
                                             int(dispatcher\_port),
                                             "dispatch:%s" % commit)
              if response != "OK":
                  raise Exception("Could not dispatch the test: %s" %
                  response)
              print "dispatched!"
          else:
              raise Exception("Could not dispatch the test: %s" %
              response)
      time.sleep(5)

如果你不使用KeyboardInterrupt （Ctrl + c）终止监听器发送进程或发送终止信号，监听器会永远重复这一操作。

测试样例分配器（ dispatcher.py ）

测试样例分配器是一个用来为测试运行器分配测试任务的独立进程。他会在一个指定端口监听来自代码库监听器及测试运行器的请求。分配器允许测试运行器主动注册，当监听器发送一个提交ID时，它会将测试工作分配给一个已经注册的测试运行器。同时，它还可以平稳的处理测试运行器遇到的各种问题，当一个运行器发生故障，它可以立即将该运行器运行测试的提交ID重新分配给一个新的测试运行器。

dispatch.py 脚本以serve函数为入口。首先，它会解析你设定的分配器的地址及端口：

def serve(): parser = argparse.ArgumentParser() parser.add_argument("--host", help="dispatcher's host, by default it uses localhost", default="localhost", action="store") parser.add_argument("--port", help="dispatcher's port, by default it uses 8888", default=8888, action="store") args = parser.parse_args()

这里我们会开启分配器进程以及一个runner_checker函数进程，和一个redistribute函数进程。

  server = ThreadingTCPServer((args.host, int(args.port)), DispatcherHandler)
  print `serving on %s:%s` % (args.host, int(args.port))
  ...
  runner\_heartbeat = threading.Thread(target=runner\_checker, args=(server,))
  redistributor = threading.Thread(target=redistribute, args=(server,))
  try:
      runner\_heartbeat.start()
      redistributor.start()
      # Activate the server; this will keep running until you
      # interrupt the program with Ctrl\+C or Cmd\+C
      server.serve\_forever()
  except (KeyboardInterrupt, Exception):
      # if any exception occurs, kill the thread
      server.dead = True
      runner\_heartbeat.join()
      redistributor.join()

runner_checker函数会定期的ping每一个注册在位的运行器，来确保他们都处于正常工作的状态。。如果有运行器没有响应，该函数就会将其从注册的运行器池中删除，并且之前分配给他的提交ID会被重新分配给一个新的可用的运行器。函数会在pending_commits变量中记录运行受到运行器失去响应影响的提交ID

  def runner\_checker(server):
      def manage\_commit\_lists(runner):
          for commit, assigned\_runner in server.dispatched\_commits.iteritems():
              if assigned\_runner == runner:
                  del server.dispatched\_commits[commit]
                  server.pending\_commits.append(commit)
                  break
          server.runners.remove(runner)
      while not server.dead:
          time.sleep(1)
          for runner in server.runners:
              s = socket.socket(socket.AF\_INET, socket.SOCK\_STREAM)
              try:
                  response = helpers.communicate(runner["host"],
                                                 int(runner["port"]),
                                                 "ping")
                  if response != "pong":
                      print "removing runner %s" % runner
                      manage\_commit\_lists(runner)
              except socket.error as e:
                  manage\_commit\_lists(runner)

redistribute函数用来将pending_commits中记录的提交ID进行分发。当redistribute运行时，它会不断的检查pending_commits中是否有提交ID。一旦发现pending_commits中存在提交ID，函数会调用dispatch_tests方法来将提交ID分配出去。

  def redistribute(server):
      while not server.dead:
          for commit in server.pending\_commits:
              print "running redistribute"
              print server.pending\_commits
              dispatch\_tests(server, commit)
              time.sleep(5)

dispatch_tests函数用来从已注册的运行器池中返回一个可用的运行器。如果得到了一个可用的运行器，函数会发送一个带有提交ID的运行测试指令。如果当前没有可用的运行器，函数会在2s的休眠之后重复上述过程。如果分配成功了，函数会在dispatched_commits变量中记录提交ID及该提交ID的测试正在由哪一个运行器运行。如果提交ID在pending_commits变量中，dispatch_tests函数会在重新分配后将提交ID从pending_commits中删除。

def dispatch_tests(server, commit_id):

  # NOTE: usually we don't run this forever
  while True:
      print "trying to dispatch to runners"
      for runner in server.runners:
          response = helpers.communicate(runner["host"],
                                         int(runner["port"]),
                                         "runtest:%s" % commit\_id)
          if response == "OK":
              print "adding id %s" % commit\_id
              server.dispatched\_commits[commit\_id] = runner
              if commit\_id in server.pending\_commits:
                  server.pending\_commits.remove(commit\_id)
              return
      time.sleep(2)

分配器服务用到了标准库中的一个叫SocketServer模块，它是一个非常简单的服务器。SocketServer模块中有四种基本的服务器类型： TCP ， UDP ， UnixStreamServer和UnixDatagramServer 。为了保证我们的数据传输连续稳定，我们使用基于TCP协议的套接字（UPD并不能保证数据的稳定和连续）。

SocketServer中提供的默认的TCPServer同时只能处理一个请求，因此当分配器与一个运行器建立会话后，就无法再同时与监听器建立连接了。此时来自监听器的会话只能等待第一个会话完成并断开连接才能建立与分配器的连接。这对于我们的项目而言并不是非常理想，在我们预想中，分配器应该直接而迅速的同时与所有运行器及监听器进行通信。

为了使我们的分配器可以同时维护多个连接，我们使用了一个自定义的类ThreadingTCPServer来为默认的SocketServer 类增加多线程运行的功能。也就是说无论何时分配器接收到连接请求，他都会新建一个进程来处理这个会话。这就使分配器同时维护多个连接成为了可能。

class ThreadingTCPServer(SocketServer.ThreadingMixIn, SocketServer.TCPServer): runners = [] # Keeps track of test runner pool dead = False # Indicate to other threads that we are no longer running dispatched_commits = {} # Keeps track of commits we dispatched pending_commits = [] # Keeps track of commits we have yet to dispatch

分配器会为每一个请求定义了处理函数。我们通过DispatcherHandler类定义来实现这种功能。DispatcherHandler继承自SocketServer的BaseRequestHandler 基类。这个基类只需要我们定义句柄函数，只要请求连接它就会被调用。我们将这个函数的自定义内容写在DispatcherHandler中，并且确保在每一次请求出现是，这个函数能够被调用。这个函数会不断地监听发来的请求（ self.request中保存着请求信息），并解析请求中的指令。

class DispatcherHandler(SocketServer.BaseRequestHandler): """ The RequestHandler class for our dispatcher. This will dispatch test runners against the incoming commit and handle their requests and test results """ command_re = re.compile(r"(\w+)(:.+)*") BUF_SIZE = 1024 def handle(self): self.data = self.request.recv(self.BUF_SIZE).strip() command_groups = self.command_re.match(self.data) if not command_groups: self.request.sendall("Invalid command") return command = command_groups.group(1)

这个函数可以处理如下指令： status ， register ， dispatch和results 。其中 status函数用来检测分配器服务是否处于运行状态。

      if command == "status":
          print "in status"
          self.request.sendall("OK")

为了让分配器的功能生效，我们需要注册至少一个运行器。当注册器被调用时，为了确保在需要发送提交ID触发测试时能准确的找到对应的运行器，会在一个列表中保存下运行器的“地址：端口”数据（运行器的数据会被保存在一个叫ThreadingTCPServer的对象中）。

      elif command == "register":
          # Add this test runner to our pool
          print "register"
          address = command\_groups.group(2)
          host, port = re.findall(r":(\w\*)", address)
          runner = {"host": host, "port":port}
          self.server.runners.append(runner)
          self.request.sendall("OK")

dispatch指令用于代码库监听器下发测试的提交ID给测试运行器。此命令的用法为dispatch: 。分配器从此消息中解析出提交ID并将其发送给测试运行器。

      elif command == "dispatch":
          print "going to dispatch"
          commit\_id = command\_groups.group(2)[1:]
          if not self.server.runners:
              self.request.sendall("No runners are registered")
          else:
              # The coordinator can trust us to dispatch the test
              self.request.sendall("OK")
              dispatch\_tests(self.server, commit\_id)

results指令会由测试运行器在上报测试结果是调用。此命令的用法为results::: 。用于标识测试报告对应的提交ID。用于计算结果数据使用需要多大的缓冲区。最后， 中是实际报告信息。

      elif command == "results":
          print "got test results"
          results = command\_groups.group(2)[1:]
          results = results.split(":")
          commit\_id = results[0]
          length\_msg = int(results[1])
          # 3 is the number of ":" in the sent command
          remaining\_buffer = self.BUF\_SIZE \- \
              (len(command) \+ len(commit\_id) \+ len(results[1]) \+ 3)
          if length\_msg > remaining\_buffer:
              self.data \+= self.request.recv(length\_msg \- remaining\_buffer).strip()
          del self.server.dispatched\_commits[commit\_id]
          if not os.path.exists("test\_results"):
              os.makedirs("test\_results")
          with open("test\_results/%s" % commit\_id, "w") as f:
              data = self.data.split(":")[3:]
              data = "\n".join(data)
              f.write(data)
          self.request.sendall("OK")

测试运行器（ test_runner.py ）

测试运行器负责对给定的提交ID运行测试，并上报测试结果。它仅会与分配器通信，分配器负责为其提供需要运行测试的提交ID，并且会接收测试结果报告。

test_runner.py文件会以启动测试运行器服务的serve函数为入口，并启动一个线程来运行dispatcher_checker函数。由于此启动过程与repo_observer.py和dispatcher.py的启动过程非常相似，因此我们在这里就不再赘述。

dispatcher_checker函数每五秒对分配器执行一次ping操作，以确保它仍然在正常运行。这个操作主要是出于资源管理上的考虑。如果对应的分配器挂了，就需要将测试运行器也关闭。 否则测试运行器就只能空跑，及接收不到新的任务也无法提交之前任务产生报告。

  def dispatcher\_checker(server):
      while not server.dead:
          time.sleep(5)
          if (time.time() \- server.last\_communication) > 10:
              try:
                  response = helpers.communicate(
                                     server.dispatcher\_server["host"],
                                     int(server.dispatcher\_server["port"]),
                                     "status")
                  if response != "OK":
                      print "Dispatcher is no longer functional"
                      server.shutdown()
                      return
              except socket.error as e:
                  print "Can't communicate with dispatcher: %s" % e
                  server.shutdown()
                  return

测试运行器的服务于分配器相同都是ThreadingTCPServer ，它需要多线程运行的是因为分配器既会向它下发提交ID，也可能在测试运行的期间的ping它。

class ThreadingTCPServer(SocketServer.ThreadingMixIn, SocketServer.TCPServer): dispatcher_server = None # Holds the dispatcher server host/port information last_communication = None # Keeps track of last communication from dispatcher busy = False # Status flag dead = False # Status flag

整个通信流是从分配器向测试运行器发送需要运行测试的提交ID开始的。如果测试运行器的状态可以运行测试，它会发送确认消息回分配器，然后关闭第一个连接。为了使测试运行器在跑测试的同时还能接受来自分配器的其他请求，它会单独启动一个进程来运行测试。

这样，当分配器在测试运行器正在运行测试的时候发来一个请求（比如一个ping请求）， 测试运行器的测试跑在另一个进程上，运行器服务本身仍然可以在作出响应。这样测试运行器就可支持同时运行多个任务了。还有一种替代多线程运行的设计是在分配器与测试运行器间建立一个长连接。 但这样会在分配器端消耗大量的内存来维持连接，另外这种方式还对网络有强依赖。 如果网络一旦产生波动（比如突然的断线）就会对系统造成破坏。

测试运行器会从分配器接收到两种消息。 第一种是ping消息 ，分配器用这个消息来验证测试运行器是否仍处于活跃状态。

class TestHandler(SocketServer.BaseRequestHandler): …

  def handle(self):
      ....
      if command == "ping":
          print "pinged"
          self.server.last\_communication = time.time()
          self.request.sendall("pong")

另一个是 runtest ，它的格式是runtest: 。这条指令用于分配器下发需要测试的提交ID。当接收到runtest时，测试运行器将检查当前是否有正在运行的测试。如果有，它会给分配器返回BUSY的响应。如果没有，它会返回OK，将其状态设置为busy并运行其run_tests函数。

      elif command == "runtest":
          print "got runtest command: am I busy? %s" % self.server.busy
          if self.server.busy:
              self.request.sendall("BUSY")
          else:
              self.request.sendall("OK")
              print "running"
              commit\_id = command\_groups.group(2)[1:]
              self.server.busy = True
              self.run\_tests(commit\_id,
                             self.server.repo\_folder)
              self.server.busy = False

这个函数会调用一个叫test_runner_script.sh 的shell脚本，该脚本会将代码更新为给定的提交ID。脚本返回后，如果代码库已经被成功的更新，运行器会使用unittest运行测试并将结果收集到一个文件中。测试运行完毕后，测试运行器将读入结果报告文件，并将报告发送给调度程序。

  def run\_tests(self, commit\_id, repo\_folder):
      # update repo
      output = subprocess.check\_output(["./test\_runner\_script.sh",
                                      repo\_folder, commit\_id])
      print output
      # run the tests
      test\_folder = os.path.join(repo\_folder, "tests")
      suite = unittest.TestLoader().discover(test\_folder)
      result\_file = open("results", "w")
      unittest.TextTestRunner(result\_file).run(suite)
      result\_file.close()
      result\_file = open("results", "r")
      # give the dispatcher the results
      output = result\_file.read()
      helpers.communicate(self.server.dispatcher\_server["host"],
                          int(self.server.dispatcher\_server["port"]),
                          "results:%s:%s:%s" % (commit\_id, len(output), output))

这是test_runner_script.sh的内容 ：

!/bin/bash

REPO=$1 COMMIT=$2 source run_or_fail.sh run_or_fail "Repository folder not found" pushd "$REPO" 1> /dev/null run_or_fail "Could not clean repository" git clean -d -f -x run_or_fail "Could not call git pull" git pull run_or_fail "Could not update to given commit hash" git reset --hard "$COMMIT"

要运行test_runner.py ，必须将其指向存储库的副本。你可以使用我们先前创建的/path/to/test_repo test_repo_clone_runner 副本作为启动参数。默认情况下， test_runner.py将在localhost的8900-9000端口上启动，并尝试连接到localhost:8888上的调度程序服务器。你通过可以一些可选参数来更改这些值。--host和--port参数用于指定运行测试运行器服务器地址和端口， --dispatcher-server参数指定调度程序的地址。

控制流程图

图2.1是该系统的概述图。图中假设所有三个文件（ repo_observer.py ， dispatcher.py和test_runner.py ）都已在运行，并描述了每个进程在新的提交发生时所采取的操作。

图2.1 - 控制流程

运行代码

我们可以在本地运行这个简单的CI系统，为每个进程使用不同的终端shell。我们首先启动分配器，它默认运行在端口8888上：

$ python dispatcher.py

开一个新的的shell，我们启动测试运行器（这样它就可以在分配器中注册了）：

$ python test_runner.py

测试运行器将自动为自己分配端口，范围为8900-9000。你可以根据需求尽可能多起几个测试运行器。

最后，在另一个新shell中，让我们启动代码库监听器：

$ python repo_observer.py --dispatcher-server=localhost:8888

现在万事俱备，让我们触发一些测试吧玩一下吧！根据设计我们需要创建一个新的提交来触发测试。切换到你的主代码仓库中， 随便改点什么：

$ cd /path/to/test_repo $ touch new_file $ git add new_file $ git commit -m"new file" new_file

然后repo_observer.py识别到有一个新的提交产生了，之后通知分配器。你可以在它们各自的shell窗口中查看它们的运行日志。当分配器收到测试结果，它就会将它们存储在此代码库中的test_results/文件夹中，并使用提交ID作为文件名。

错误处理

该CI系统中包括一些简单的错误处理。

如果你讲test_runner.py进程杀掉， dispatcher.py将确定该运行器将会识别出这个节点已经不再活跃，并将其从运行器池中移除。

你也可以模拟网络或系统故障，在测试运行器执行测试的时候将它杀死。这时，分配器会识别到运行器已经挂了，它会将挂掉的运行器从运行器池中移除，并将这个运行器之前在执行的任务分配给池中其他的运行器。

如果你杀掉分配器，那么监听器会直接报错。测试运行器也会发现分配器不再运行，并自动关闭。

结论

通过逐个分析各个进程中的不同功能，我们对构建了一个分布式的持续集成系统的有了一些基本的认识。通过套接字请求实现进程间的通信，我们的CI系统可以分布式的运行在不同的机器上，这增强了我们的系统可靠性和可扩展性。

这套CI系统现在的功能仍然非常简单，你还可以发挥自己的才能对它进行各种扩展以实现更多功能。以下是一些改进建议：

对每次提交自动运行测试

当前系统将定期检查是否有新的提交并对最近的一次提交运行测试。这个设计可以改为每次提交都触发测试。你可以修改定期检查程序，获取在两次轮询中发生的所有提交来实现这个功能。

更智能的运行器

如果测试运行器检测到分配器没有响应，则它将停止运行。当测试运行器正在运行测试时，也会立即关闭！如果测试运行器可以有一段时的等待期或者长期运行（如果你并不在乎它对资源的占用）来等待分配器恢复可能会更好一些。这样当分配器恢复时，运行器既可以将之前执行的测试的报告重新发回分配器。这样可以避免因分配器故障而引起的重复任务，在对每一个提交都执行测试时，这将很大程度上节约运行器资源。

报告展示

在真正的CI系统中，测试报告一般会发送到一个单独的报告服务。在报告系统中，人们可以查看报告详情，或是设置一些通知规则，在遇到故障或其他一些特殊的情况下通知相关人员。你可以为我的CI系统创建一个独立的报告进程，替换掉分配器的报告收集功能。这个新的进程可以是一个Web服务（或链接到一个Web服务上）， 这样我们就可以在网页上直接在线查看测试报告，甚至可以用一个邮件服务器来实现测试失败时的提醒。

测试运行器管理器

在当前的系统中，我们必须手动运行test_runner.py文件来启动测试运行器。你可以创建一个测试运行器管理器进程，通过这个进程来管理查看所有运行器上的负载和来自分配器的请求，对运行器的数量进行相应的调整。这个进程会接受所有的测试任务，根据任务启动测试运行器，并在任务少的时候减少运行器的实例。

遵循这些建议，你可以使这个简单的CI系统更加健壮并且容错率更高，并且具有与其他系统（比如一个网页版的报告查看器）集成的能力。

如果你希望了解现在的持续集成系统可以实现到什么样的灵活性，我建议你去看看Jenkins ，这是一个用Java编写的非常强大的开源CI系统。它提供了一个基本的CI系统，同时也允许使用插件进行扩展。你可以通过GitHub访问其源代码。另一个推荐的项目是Travis CI ，它是用Ruby编写的，其源代码也可以通过GitHub 得到。

这是了解CI系统如何工作以及如何自己构建CI系统的尝试。现在你应该对制作一个可靠的分布式系统所需的内容有了更深入的了解，希望你可以利用这些知识开发更复杂的解决方案。

使用Bash是因为我们需要 检查文件存在，创建文件和使用Git，而shell脚本是实现这一目标的最直接，最简单的方法。当然，你也可以使用一些跨平台的Python包;例如，用的Python内置的os模块访问文件系统，用GitPython访问Git，但是用这些包实现起来稍微有一点不直观。↩

      </path></path></commit></results></length></commit></results></length></commit></commit></commit></code>