基于Flask与MySQL的微服务城市数据查询系统
服务模块实现
1. 数据存储服务（端口 5001）
该服务负责与 MySQL 数据库交互，提供城市相关数据的读写功能。使用 SQLAlchemy 进行数据库操作，Flask 框架处理 HTTP 请求。
from flask import Flask, request, jsonify
from sqlalchemy import create_engine, text
server = Flask(__name__)
# 请替换为实际的数据库连接信息：用户名:密码@主机地址:端口/数据库名称
db_engine = create_engine("mysql+pymysql://root:your_password@localhost:3306/your_database")
@server.post("/fetch_value")
def fetch_value():
city_name = request.json.get("city")
with db_engine.connect() as conn:
result = conn.execute(
text("SELECT value FROM city_data WHERE city = :city"),
{"city": city_name}
).fetchone()
return jsonify({"value": result[0]}) if result else ("", 404)
@server.post("/update_value")
def update_value():
data = request.json
with db_engine.connect() as conn:
conn.execute(
text("REPLACE INTO city_data (city, value) VALUES (:city, :value)"),
data
)
conn.commit()
return "", 201
if __name__ == "__main__":
server.run(port=5001, debug=False)
2. 业务逻辑服务（端口 5002）
此服务作为中间层，首先尝试从本地数据库获取城市高程数据；若未命中，则调用外部地理编码 API 获取，并将结果缓存至数据库。
from flask import Flask, request, jsonify
import requests
app = Flask(__name__)
database_url = "http://127.0.0.1:5001"
geocode_api = "https://geocoding-api.open-meteo.com/v1/search"
@app.post("/query_city_info")
def query_city_info():
payload = request.json
city = payload["city"]
# 尝试从数据库读取缓存
cache_response = requests.post(f"{database_url}/fetch_value", json={"city": city})
if cache_response.status_code == 200:
return jsonify(cache_response.json())
# 缓存未命中，请求第三方接口
api_response = requests.get(geocode_api, params={"name": city, "count": 1})
results = api_response.json().get("results", [])
if not results:
return "", 404
elevation_value = str(results[0]["elevation"])
# 写入数据库以供后续使用
requests.post(f"{database_url}/update_value", json={"city": city, "value": elevation_value})
return jsonify({"value": elevation_value})
if __name__ == "__main__":
app.run(port=5002, debug=False)
3. 网关接口服务（端口 5003）
对外暴露的统一入口，接收路径参数形式的城市名，转发请求至逻辑服务并格式化响应内容。
from flask import Flask
import requests
gateway = Flask(__name__)
logic_service = "http://127.0.0.1:5002"
output_field = "elevation"
@gateway.get("/gateway/<city_name>")
def handle_request(city_name):
response = requests.post(
f"{logic_service}/query_city_info",
json={"city": city_name}
)
if response.status_code == 200:
return f"The {output_field} of {city_name} is {response.json()['value']}.\n"
return f"Error: {city_name} not found.\n", 404
if __name__ == "__main__":
gateway.run(port=5003, debug=False)
部署流程说明
环境准备
- 在 PyCharm 中创建新项目。
- 通过终端安装所需依赖：
pip install flask sqlalchemy pymysql requests
- 在 MySQL 客户端执行建表语句（仅需一次）：
CREATE TABLE city_data (
city VARCHAR(50) PRIMARY KEY,
value VARCHAR(50)
);
文件配置步骤
- database.py：新建文件并粘贴数据服务代码，修改连接字符串中的用户名、密码和数据库名。
- logic.py：新建文件并插入业务逻辑代码，无需额外修改。
- gateway.py：新建文件并填入网关代码，保持默认设置即可。
启动顺序要求
必须按照以下顺序依次启动服务，确保依赖关系正确：
- 启动数据库服务：
打开终端运行：python database.py
确认输出包含* Running on http://127.0.0.1:5001 - 启动逻辑处理服务：
另开终端执行：python logic.py
等待服务监听在 5002 端口。 - 启动网关服务：
再开启一个终端窗口运行：python gateway.py
验证其已在 5003 端口就绪。
功能测试方法
服务全部启动后，在终端中执行以下命令进行验证：
curl http://127.0.0.1:5003/gateway/Shanghai
预期返回示例：
The elevation of Shanghai is 4.
再次请求相同城市时，应能从数据库快速响应，体现缓存机制生效。
