分布式计算：使用Python的Flask或Django实现简单Web服务案例

原创 ╰+哭是因爲堅強的太久メ 2025-01-26 09:24 50阅读 0赞

分布式计算通常指的是将一个大任务分解为多个子任务，然后在多台计算机上并行执行这些子任务。这里我们将使用Python的两个web框架Flask和Django来实现一个简单的分布式计算Web服务。

**一、Flask实现**

1. **安装Flask和其他依赖库**：
   ```
   pip install Flask redis
   ```

2. **创建Flask应用并初始化Redis缓存**：
   ```python
   from flask import Flask, jsonify, request
   import redis

app = Flask(__name__)
   r = redis.Redis(host='localhost', port=6379)

# 示例分布式计算任务：累加一个列表中的所有元素
   @app.route('/add', methods=['POST'])
   def add():
       data = request.get_json()
       if not isinstance(data, list):
           return jsonify({'error': 'Invalid input - please provide a list of numbers.'}), 400

result = sum(data)
       r.set('sum', result)

return jsonify({'result': result}}), 200

# 部署Flask应用
   if __name__ == '__main__':
       app.run(host='0.0.0.0', port=5000, debug=True)
   ```

3. **启动Redis和Flask服务器，访问示例任务**：
   - 启动Redis：`redis-server`
   - 启动Flask服务器：`python your_script.py`

4. **在浏览器中访问`http://localhost:5000/add`，提交一个包含数字的列表，查看返回结果。**

**二、Django实现**

Django相比于Flask更强大，更适合处理复杂的Web应用。以下是一个简单的Django分布式计算任务示例：

1. **安装Django和其他依赖库**：
   ```
   pip install Django redis
   ```

2. **创建Django项目并设置数据库**：
   ```shell
   django-admin startproject distributed_computation
   cd distributed_computation/
   python manage.py migrate
   ```

3. **创建Django应用并定义模型和视图**：
   ```python
   from django.db import models
   from django.shortcuts import render

class Task(models.Model):
       data = models.ListField(models.FloatField(), blank=True))

def __str__(self):
           return str(self.id)

@app.route('/add_task', methods=['POST'])
   def add_task():
       task_data = request.get_json()
       if not isinstance(task_data, list)):
           return jsonify({'error': 'Invalid input - please provide a list of numbers.'}}), 400

# 检查数据是否符合模型定义
       for num in task_data:
           if not isinstance(num, float)) or num <= 0:
               return jsonify({'error': 'Invalid data - elements must be non-negative floating point numbers.'}}), 422

new_task = Task(data=task_data))
       new_task.save()

return jsonify({'result': new_task.id}}), 201

# 启动Django开发服务器
   if __name__ == '__main__':
       from django.core.wsgi import get_wsgi.application
       wsgi_app = get_wsgi_application()
       app.run(host='0.0.0.0', port=8000, debug=True)
   ```

4. **在浏览器中访问`http://localhost:8000/add_task`，提交一个包含数字的列表，查看返回结果。**

以上两种方式都实现了分布式计算任务，你可以根据需求选择合适的框架。