基于Python的微服务Demo：使用Flask框架与Consul实现服务注册、发现及健康检查

我来给你一个基于Flask和Consul的微服务Demo实现。

# service_provider.py - 服务提供者
from flask import Flask, jsonify
import consul
import socket
import threading
import time

app = Flask(__name__)

# 初始化Consul客户端
consul_client = consul.Consul(host='localhost', port=8500)

def register_service(service_name, port):
    """注册服务到Consul"""
    service_id = f"{service_name}-{socket.gethostname()}-{port}"
    
    # 获取本机IP
    hostname = socket.gethostname()
    ip_address = socket.gethostbyname(hostname)
    
    # 注册服务
    consul_client.agent.service.register(
        name=service_name,
        service_id=service_id,
        address=ip_address,
        port=port,
        check={
            "name": f"{service_name} Health Check",
            "tcp": f"{ip_address}:{port}",
            "interval": "10s",
            "timeout": "5s"
        }
    )
    print(f"服务 {service_name} 已注册到Consul")

def health_check_thread():
    """健康检查线程"""
    while True:
        try:
            # 这里可以添加自定义的健康检查逻辑
            time.sleep(30)
        except Exception as e:
            print(f"健康检查异常: {e}")

@app.route('/api/hello', methods=['GET'])
def hello():
    return jsonify({"message": "Hello from Flask Service!", "status": "healthy"})

@app.route('/health', methods=['GET'])
def health():
    """健康检查端点"""
    return jsonify({"status": "healthy"}), 200

if __name__ == '__main__':
    # 注册服务
    service_port = 5000
    register_service('flask-service', service_port)
    
    # 启动健康检查线程
    health_thread = threading.Thread(target=health_check_thread, daemon=True)
    health_thread.start()
    
    # 启动Flask应用
    app.run(host='0.0.0.0', port=service_port, debug=False)

# service_consumer.py - 服务消费者
from flask import Flask, jsonify
import consul
import requests
import time

app = Flask(__name__)

# 初始化Consul客户端
consul_client = consul.Consul(host='localhost', port=8500)

def discover_service(service_name):
    """从Consul发现服务"""
    try:
        # 获取健康的服务实例
        services = consul_client.health.service(service_name, passing=True)[1]
        
        if not services:
            raise Exception(f"未找到可用的 {service_name} 服务")
        
        # 简单的负载均衡：选择第一个可用实例
        service = services[0]
        service_address = service['Service']['Address']
        service_port = service['Service']['Port']
        
        return f"http://{service_address}:{service_port}"
        
    except Exception as e:
        print(f"服务发现失败: {e}")
        return None

@app.route('/api/call-service', methods=['GET'])
def call_service():
    """调用远程服务"""
    service_url = discover_service('flask-service')
    
    if not service_url:
        return jsonify({"error": "服务不可用"}), 503
    
    try:
        response = requests.get(f"{service_url}/api/hello", timeout=5)
        return jsonify({
            "message": "服务调用成功",
            "service_response": response.json(),
            "service_url": service_url
        })
    except requests.exceptions.RequestException as e:
        return jsonify({"error": f"服务调用失败: {str(e)}"}), 500

@app.route('/api/services', methods=['GET'])
def list_services():
    """列出所有注册的服务"""
    try:
        services = consul_client.agent.services()
        return jsonify({"services": services})
    except Exception as e:
        return jsonify({"error": str(e)}), 500

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5001, debug=False)

# requirements.txt
Flask==2.3.3
python-consul==1.1.0
requests==2.31.0

使用说明：

1. 安装依赖：

pip install -r requirements.txt

1. 启动Consul（需要先安装Consul）：

# 下载Consul并启动开发模式
consul agent -dev -ui -client=0.0.0.0

1. 运行服务提供者：

python service_provider.py

1. 运行服务消费者：

python service_consumer.py

核心功能说明：

1. 服务注册：service_provider.py启动时自动将Flask服务注册到Consul，包含服务名称、IP、端口和健康检查配置。

2. 健康检查：Consul通过TCP连接定期检查服务端口是否可达，确保只有健康的服务才会被发现。

3. 服务发现：service_consumer.py通过Consul API查询健康的服务实例，获取服务地址进行调用。

4. 服务调用：消费者通过HTTP调用提供者的API端点，实现服务间通信。

测试端点：

• 服务提供者：http://localhost:5000/api/hello
• 服务消费者：http://localhost:5001/api/call-service
• 查看所有服务：http://localhost:5001/api/services
• Consul UI：http://localhost:8500

这个Demo展示了微服务的基本模式：服务注册、发现和健康检查。实际项目中你可能还需要添加配置管理、熔断器、负载均衡等组件。

总结：用Flask+Consul可以快速搭建微服务基础架构。