Heroku部署LineBot：子程序定时向主程序传数据的技术问询

嘿，我来帮你梳理下这些问题的解决方案，都是基于Heroku和Python开发LineBot的实际经验哈：

完全可行！你说主程序里用Timer容易中断，核心原因是LineBot本身是Web服务（比如基于Flask/FastAPI），请求处理会阻塞Timer的执行，再加上Heroku的web dyno可能因空闲休眠，导致定时任务失效。把定时逻辑单独放到子程序里作为独立进程运行，就能完美避开这个问题。

不过要注意两个点：

• Heroku的免费dyno（包括worker类型）会在30分钟无活动后休眠，所以如果你的定时任务需要24小时不间断运行，免费额度可能不够用；
• 主程序和子程序是独立进程，不能直接共享内存，得用专门的跨进程通信方式传递数据。

在Heroku上配置 #

Heroku支持多进程部署，步骤很简单：

1. 在项目根目录创建一个名为Procfile的文件（没有后缀），内容如下：

web: python A.py  # 主程序，作为web服务接收Line的请求
worker: python B.py  # 子程序，运行定时检测任务

2. 部署代码到Heroku后，去Heroku dashboard的Resources页面，手动开启worker dyno（默认可能是关闭的）。

小贴士：如果用免费dyno，worker进程也会休眠，你可以让子程序每隔一段时间给主程序的web接口发个请求，唤醒dyno，但这种方法稳定性一般，长期用建议升级到付费dyno。

替代平台 #

如果Heroku的休眠问题影响你的定时任务，还有这些更稳定的选择：

• 云函数+定时触发器：比如AWS Lambda、Google Cloud Functions，配合它们的定时调度工具，每20秒触发一次任务，执行完就释放资源，成本极低；
• VPS/轻量云服务器：自己买个便宜的VPS，直接部署主程序和子程序，用systemd或者supervisor管理进程，完全不会有休眠问题，稳定性拉满。

因为两者是独立进程，不能直接调用函数，得用跨进程通信方式，下面两种是最常用的：

方式一：HTTP接口通信（最简单） #

这种方式不需要额外中间件，适合小型项目：

1. 给主程序A.py加一个内部接口（以Flask为例）：

from flask import Flask, request, jsonify
import os

app = Flask(__name__)

# 接收子程序传来的地震信息
@app.route('/send-earthquake', methods=['POST'])
def receive_earthquake():
    data = request.get_json()
    # 这里写把地震信息批量发送给所有Line用户的逻辑
    print(f"收到最新地震信息：{data}")
    return jsonify({"status": "success"})

# LineBot的核心逻辑...

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=int(os.environ.get('PORT', 5000)))

2. 修改子程序B.py，检测到地震信息后用requests发送POST请求，返回格式用JSON（通用且易解析）：

from datetime import datetime
from threading import Timer
import requests
import os

# 主程序的Heroku域名，比如https://your-linebot-app.herokuapp.com
MAIN_APP_URL = os.environ.get('MAIN_APP_URL', 'http://localhost:5000')

def check_earthquake_and_send(inc):
    # 第一步：替换成实际的政府地震信息检测逻辑
    # 模拟获取到的地震数据
    earthquake_data = {
        "timestamp": datetime.now().isoformat(),
        "location": "台北市",
        "magnitude": 4.2,
        "depth": 10
    }
    
    # 第二步：发送数据给主程序
    try:
        response = requests.post(f"{MAIN_APP_URL}/send-earthquake", json=earthquake_data)
        response.raise_for_status()
        print("地震信息已成功发送给主程序")
    except Exception as e:
        print(f"发送失败：{str(e)}")
    
    # 第三步：继续定时执行
    t = Timer(inc, check_earthquake_and_send, (inc,))
    t.start()

# 启动定时任务，每20秒执行一次
check_earthquake_and_send(20)

方式二：Redis消息队列（更可靠） #

如果担心HTTP请求偶尔失败，可以用Redis做中间件，数据会暂存在队列里，不会丢失：

1. 在Heroku上添加Heroku Redis插件，或者自己搭建Redis服务；
2. 子程序B.py把地震信息推送到Redis队列：

from datetime import datetime
from threading import Timer
import redis
import json
import os

# 从环境变量获取Redis连接地址
redis_client = redis.from_url(os.environ.get('REDIS_URL'))

def check_earthquake_and_push(inc):
    earthquake_data = {
        "timestamp": datetime.now().isoformat(),
        "location": "台北市",
        "magnitude": 4.2,
        "depth": 10
    }
    # 把JSON字符串推送到队列
    redis_client.rpush('earthquake_queue', json.dumps(earthquake_data))
    print("地震信息已存入Redis队列")
    
    t = Timer(inc, check_earthquake_and_push, (inc,))
    t.start()

check_earthquake_and_push(20)

3. 主程序A.py启动后台线程，定时读取队列数据：

from flask import Flask
import redis
import json
from threading import Thread
import time
import os

app = Flask(__name__)
redis_client = redis.from_url(os.environ.get('REDIS_URL'))

def process_earthquake_queue():
    while True:
        # 从队列头部取出数据
        data = redis_client.lpop('earthquake_queue')
        if data:
            earthquake_info = json.loads(data)
            # 执行发送给所有用户的逻辑
            print(f"处理地震信息：{earthquake_info}")
        time.sleep(5)  # 每隔5秒检查一次队列

# 启动后台线程，daemon=True表示随主程序退出而终止
Thread(target=process_earthquake_queue, daemon=True).start()

# LineBot核心逻辑...

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=int(os.environ.get('PORT', 5000)))

内容的提问来源于stack exchange，提问作者Chevady Ju