使用Etcd 提升系统健壮性

Posted 2022-01-31 胖虎是只mao

介绍

采用微服务架构的系统里包含许多服务，这些服务之间会互相调用。每个服务一般都有多个节点来提供服务，这些节点是动态变化的。当有节点新增或失效时，需要及时通知到服务调用方，否则调用方可能访问到失效节点，或者忽略掉新增节点。这种分布式系统下服务的可靠性，一般可通过服务注册与发现来达到。

使用 Etcd 来实现微服务架构系统的高可用性。目标是使得当各个后台服务节点有新增、删除或异常退出时，前台网站能够及时调整服务调用地址。

随着 CoreOS 和 Kubernetes 等项目在开源社区日益火热，它们项目中都用到的 etcd 组件作为一个高可用强一致性的服务发现存储仓库，渐渐为开发人员所关注。在云计算时代，如何让服务快速透明地接入到计算集群中，如何让共享配置信息快速被集群中的所有机器发现，更为重要的是，如何构建这样一套高可用、安全、易于部署以及响应快速的服务集群，已经成为了迫切需要解决的问题。etcd 为解决这类问题带来了福音，本文将从 etcd 的应用场景开始，深入解读 etcd 的实现方式，以供开发者们更为充分地享用 etcd 所带来的便利。

经典应用场景

要问 etcd 是什么？很多人第一反应可能是一个键值存储仓库，却没有重视官方定义的后半句，用于配置共享和服务发现。

A highly-available key value store for shared configuration and service discovery.

实际上，etcd 作为一个受到 ZooKeeper 与 doozer 启发而催生的项目，除了拥有与之类似的功能外，更专注于以下四点。

• 简单：基于 HTTP+JSON 的 API 让你用 curl 就可以轻松使用。
• 安全：可选 SSL 客户认证机制。
• 快速：每个实例每秒支持一千次写操作。
• 可信：使用 Raft 算法充分实现了分布式。

值得注意的是，分布式系统中的数据分为控制数据和应用数据。使用 etcd 的场景默认处理的数据都是控制数据，对于应用数据，只推荐数据量很小，但是更新访问频繁的情况。

目标

1. 当启动后台服务节点的时候，将该节点的访问地址注册到 Etcd 里该服务目录下
2. 当停止后台服务节点的时候，从服务目录中移除该节点地址
3. 在节点运行过程中，如果该节点异常退出，也需要及时从服务目录中移除该节点地址
4. 每个服务至少需要启动两个节点来验证服务高可用性，在同一台服务器启动的多个节点需要使用不同的监听端口，可在启动时通过环境变量来指定

提示语

1. 关于 Etcd 的介绍，可以参考这篇文章 etcd：从应用场景到实现原理的全方位解读
2. Etcd 安装比较简单，下载对应平台的已编译好的可执行程序压缩包，解压后执行里面的 etcd 程序即可启动服务。压缩包下载地址 ，实验环境可选择 etcd-vx.y.z-linux-amd64.tar.gz 。默认处理客户端请求的端口为 2379 。
3. 在 Python 里访问 Etcd 服务可使用 Python-Etcd
4. 在注册节点地址时设置过期时间，并不断地在到期之前刷新该节点地址
5. 为了防止服务地址注册和查询影响到主线程处理请求，可启动一个线程来完成这些后台工作

在 taobei/tbbuy/app.py 文件中添加如下代码：

import threading
from tblib.etcd import init_etcd_service


threading.Thread(target=init_etcd_service, args=(app, 'tbbuy')).start()

在 taobei/tbbuy/config.py 文件中添加如下代码：

import os

class BaseConfig(object):
    LISTENER = (os.getenv('APP_LISTEN_HOST', '0.0.0.0'),
                int(os.getenv('APP_LISTEN_PORT', '5030')))

    SQLALCHEMY_DATABASE_URI = 'mysql+mysqldb://root@localhost:3306/tbbuy?charset=utf8'
    SQLALCHEMY_TRACK_MODIFICATIONS = False

    ETCD_ADDR = 'localhost:2379'

    PAGINATION_PER_PAGE = 20

    CART_PRODUCT_LIMIT = 10

在 taobei/tbfile/app.py 文件中添加如下代码：

import threading
from tblib.etcd import init_etcd_service

threading.Thread(target=init_etcd_service, args=(app, 'tbfile')).start()

在 taobei/tbfile/config.py 文件中添加如下代码：

import os


class BaseConfig(object):
    LISTENER = (os.getenv('APP_LISTEN_HOST', '0.0.0.0'),
                int(os.getenv('APP_LISTEN_PORT', '5040')))

    MONGO_URI = 'mongodb://localhost:27017/tbfile'

    ETCD_ADDR = 'localhost:2379'

在 taobei/tblib/etcd.py 文件中添加如下代码：

import time

import etcd


def init_etcd_service(app, name):
    host, port = app.config['ETCD_ADDR'].split(':')
    port = int(port)
    client = etcd.Client(host=host, port=port)

    key = '/taobei/services/'.format(name)
    value = 'http://:'.format(
        app.config['LISTENER'][0], app.config['LISTENER'][1])
    while True:
        client.write(key, value, append=True, ttl=5)
        time.sleep(4)


def init_etcd_client(app):
    host, port = app.config['ETCD_ADDR'].split(':')
    port = int(port)
    client = etcd.Client(host=host, port=port)

    key = '/taobei/services'
    while True:
        time.sleep(1)

        try:
            client.read(key, recursive=True, wait=True)
            r = client.read(key, recursive=True, sorted=True)
        except Exception as e:
            print(e)
            continue

        d = 
        for child in r.children:
            if child.value is None:
                continue

            name = child.key.split('/')[-2].upper()
            if d.get(name) is None:
                d[name] = []

            if child.value not in d[name]:
                d[name].append(child.value)
        print('Current service addresses:')
        print(d)

        for name, addresses in d.items():
            app.config['SERVICE_'.format(name)]['addresses'] = addresses

在 taobei/tbmall/app.py 文件中添加如下代码：

import threading
from tblib.etcd import init_etcd_service

threading.Thread(target=init_etcd_service, args=(app, 'tbmall')).start()

在 taobei/tbmall/config.py 文件中添加如下代码：

import os


class BaseConfig(object):
    SECRET_KEY = '4bOoOz6GFmF5vVEPd0SvyOOt7m2b16l6'

    LISTENER = (os.getenv('APP_LISTEN_HOST', '0.0.0.0'),
                int(os.getenv('APP_LISTEN_PORT', '5020')))

    SQLALCHEMY_DATABASE_URI = 'mysql+mysqldb://root@localhost:3306/tbmall?charset=utf8'
    SQLALCHEMY_TRACK_MODIFICATIONS = False

    ETCD_ADDR = 'localhost:2379'

    PAGINATION_PER_PAGE = 20

在 taobei/tbweb/app.py 文件中添加如下代码：

import threading
from tblib.etcd import init_etcd_service

threading.Thread(target=init_etcd_client, args=(app, )).start()

在 taobei/tbweb/config.py 文件中添加如下代码：

import os


class BaseConfig(object):
    LISTENER = (os.getenv('APP_LISTEN_HOST', '0.0.0.0'),
                int(os.getenv('APP_LISTEN_PORT', '5050')))
    SECRET_KEY = '4bOoOz6GFmF5vVEPd0SvyOOt7m2b16l6'

    SQLALCHEMY_DATABASE_URI = 'mysql+mysqldb://root@localhost:3306/tbweb?charset=utf8'
    SQLALCHEMY_TRACK_MODIFICATIONS = False

    ETCD_ADDR = 'localhost:2379'

    SITE_NAME = '淘贝网'
    PAGINATION_PER_PAGE = 20

    SERVICE_TBBUY = 
        'addresses': ['http://localhost:5030'],
    
    SERVICE_TBFILE = 
        'addresses': ['http://localhost:5040'],
    
    SERVICE_TBMALL = 
        'addresses': ['http://localhost:5020'],
    
    SERVICE_TBUSER = 
        'addresses': ['http://localhost:5010'],
    

在 taobei/tbweb/templates/macros.html 添加如下代码：

% macro file_url(filename) %
 filename if filename.startswith('http') else '/'.format(config['SERVICE_TBFILE']['addresses'] | random, filename) 
% endmacro %