Skip to content

Commit 3eda34f

Browse files
txl1290txl1290
authored
Merge pull request #22 from LianjiaTech/dev_pjy_consumer_opt
支持业务隔离拉起动态消费组
2 parents f4f1499 + d5242ac commit 3eda34f

10 files changed

Lines changed: 1431 additions & 5 deletions

File tree

‎scripts/config_helper.py‎

Lines changed: 225 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,225 @@
1+
#!/usr/bin/env python3
2+
"""
3+
简化的S3隔离配置管理脚本
4+
专为测试设计，提供常用配置模板和快速上传功能
5+
"""
6+
7+
import json
8+
import sys
9+
import os
10+
from datetime import datetime
11+
from typing import Dict
12+
13+
# 添加项目根目录到路径
14+
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
15+
16+
from utils import s3
17+
18+
19+
class ConfigHelper:
20+
"""配置管理助手"""
21+
22+
def __init__(self, config_key: str = "isolation/config.json"):
23+
self.config_key = config_key
24+
25+
def get_current_config(self):
26+
"""获取当前S3配置"""
27+
try:
28+
config_content = s3.get_file_text_content(self.config_key)
29+
config = json.loads(config_content)
30+
print("📥 当前S3配置:")
31+
print(json.dumps(config, indent=2, ensure_ascii=False))
32+
return config
33+
except Exception as e:
34+
print(f"❌ 获取配置失败: {e}")
35+
print("💡 可能是配置文件不存在，可以先上传一个配置")
36+
return None
37+
38+
def _get_current_config_silent(self):
39+
"""静默获取当前配置（不打印输出）"""
40+
try:
41+
config_content = s3.get_file_text_content(self.config_key)
42+
return json.loads(config_content)
43+
except:
44+
return None
45+
46+
def upload_config(self, config: Dict):
47+
"""上传配置到S3"""
48+
try:
49+
# 先获取当前版本号
50+
current_version = "1.0.0" # 默认版本
51+
try:
52+
current_config = self._get_current_config_silent()
53+
if current_config and "version" in current_config:
54+
current_version = current_config["version"]
55+
print(f"📊 当前版本: {current_version}")
56+
except:
57+
print("🆕 未找到现有配置，使用默认版本")
58+
59+
# 生成新版本号
60+
version_parts = current_version.split(".")
61+
version_parts[-1] = str(int(version_parts[-1]) + 1)
62+
new_version = ".".join(version_parts)
63+
64+
# 设置新版本和时间戳
65+
config["version"] = new_version
66+
config["last_updated"] = datetime.now().isoformat()
67+
68+
# 上传到S3
69+
s3.upload_dict_content(config, self.config_key)
70+
print(f"✅ 配置已上传到S3，新版本: {new_version}")
71+
print(f"📤 上传的配置:")
72+
print(json.dumps(config, indent=2, ensure_ascii=False))
73+
74+
except Exception as e:
75+
print(f"❌ 上传配置失败: {e}")
76+
77+
def upload_from_file(self, file_path: str):
78+
"""从文件上传配置"""
79+
try:
80+
with open(file_path, 'r', encoding='utf-8') as f:
81+
config = json.load(f)
82+
83+
print(f"📁 从文件加载配置: {file_path}")
84+
self.upload_config(config)
85+
86+
except Exception as e:
87+
print(f"❌ 从文件上传失败: {e}")
88+
89+
def get_empty_config(self) -> Dict:
90+
"""空配置 - 清空所有隔离"""
91+
return {
92+
"spaces": {},
93+
"default_config": {
94+
"max_workers": 3,
95+
"callback_timeout": 300,
96+
"task_types": ["short", "long", "image", "docx"]
97+
}
98+
}
99+
100+
def get_single_test_config(self) -> Dict:
101+
"""单业务测试配置"""
102+
return {
103+
"spaces": {
104+
"test_space_001": {
105+
"task_types": ["short", "long"],
106+
"max_workers": 2,
107+
"callback_timeout": 180
108+
}
109+
},
110+
"default_config": {
111+
"max_workers": 3,
112+
"callback_timeout": 300,
113+
"task_types": ["short", "long", "image", "docx"]
114+
}
115+
}
116+
117+
def get_multi_test_config(self) -> Dict:
118+
"""多业务测试配置"""
119+
return {
120+
"spaces": {
121+
"business_001": {
122+
"task_types": ["short", "long"],
123+
"max_workers": 2,
124+
"callback_timeout": 180
125+
},
126+
"business_002": {
127+
"task_types": ["image", "docx"],
128+
"max_workers": 4,
129+
"callback_timeout": 240
130+
},
131+
"high_priority": {
132+
"task_types": ["short", "long", "image", "docx"],
133+
"max_workers": 6,
134+
"callback_timeout": 120
135+
}
136+
},
137+
"default_config": {
138+
"max_workers": 3,
139+
"callback_timeout": 300,
140+
"task_types": ["short", "long", "image", "docx"]
141+
}
142+
}
143+
144+
def get_emergency_config(self) -> Dict:
145+
"""紧急隔离配置 - 大流量业务隔离"""
146+
return {
147+
"spaces": {
148+
"emergency_business": {
149+
"task_types": ["short", "long", "image", "docx"],
150+
"max_workers": 8,
151+
"callback_timeout": 600
152+
}
153+
},
154+
"default_config": {
155+
"max_workers": 3,
156+
"callback_timeout": 300,
157+
"task_types": ["short", "long", "image", "docx"]
158+
}
159+
}
160+
161+
162+
def main():
163+
if len(sys.argv) < 2:
164+
print("🔧 S3隔离配置管理助手")
165+
print("\n用法:")
166+
print(" python scripts/config_helper.py show # 查看当前配置")
167+
print(" python scripts/config_helper.py upload empty # 上传空配置(清空隔离)")
168+
print(" python scripts/config_helper.py upload single # 上传单业务测试配置")
169+
print(" python scripts/config_helper.py upload multi # 上传多业务测试配置")
170+
print(" python scripts/config_helper.py upload emergency # 上传紧急隔离配置")
171+
print(" python scripts/config_helper.py upload-file <file> # 从文件上传配置")
172+
print("\n配置模板:")
173+
print(" empty - 清空所有业务隔离")
174+
print(" single - 隔离一个测试业务(test_space_001)")
175+
print(" multi - 隔离多个业务进行测试")
176+
print(" emergency - 紧急大流量业务隔离")
177+
return
178+
179+
helper = ConfigHelper()
180+
command = sys.argv[1]
181+
182+
if command == "show":
183+
helper.get_current_config()
184+
185+
elif command == "upload":
186+
if len(sys.argv) < 3:
187+
print("❌ 请指定配置模板: empty, single, multi, emergency")
188+
return
189+
190+
template = sys.argv[2]
191+
config_map = {
192+
"empty": helper.get_empty_config(),
193+
"single": helper.get_single_test_config(),
194+
"multi": helper.get_multi_test_config(),
195+
"emergency": helper.get_emergency_config()
196+
}
197+
198+
if template not in config_map:
199+
print(f"❌ 未知的配置模板: {template}")
200+
print("💡 可用模板: empty, single, multi, emergency")
201+
return
202+
203+
config = config_map[template]
204+
print(f"🚀 上传配置模板: {template}")
205+
helper.upload_config(config)
206+
207+
elif command == "upload-file":
208+
if len(sys.argv) < 3:
209+
print("❌ 请指定配置文件路径")
210+
return
211+
212+
file_path = sys.argv[2]
213+
if not os.path.exists(file_path):
214+
print(f"❌ 文件不存在: {file_path}")
215+
return
216+
217+
helper.upload_from_file(file_path)
218+
219+
else:
220+
print(f"❌ 未知命令: {command}")
221+
print("💡 使用 'python scripts/config_helper.py' 查看帮助")
222+
223+
224+
if __name__ == "__main__":
225+
main()

‎server/workers/__init__.py‎

Lines changed: 81 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
from doc_parser.context import logger_context
55
from server.workers.listeners.file_api_listener import FileApiLongTaskListener, FileApiShortTaskListener, \
66
FileApiImageTaskListener, FileApiDocx2PdfTaskListener
7+
from server.workers.dynamic.dynamic_manager import dynamic_manager
78
from utils.kafka_tool import KafkaConsumer
89

910
logger = logger_context.get_logger()
@@ -21,17 +22,96 @@
2122

2223

2324
def start_workers():
24-
# 提交任务到线程池
25+
"""启动消费者工作进程"""
26+
# 启动原有的通用消费者
2527
for i, consumer in enumerate(consumers):
2628
logger.info("启动kafka消费者 topic=%s group_id=%s ", consumer.topic, consumer.group_id)
2729
executor.submit(consumer.consume_messages)
30+
31+
# 初始化动态消费者管理器
32+
dynamic_manager.initialize()
33+
logger.info("Dynamic consumer manager initialized")
34+
35+
# 启动S3配置管理器
36+
_start_config_manager()
37+
logger.info("S3 isolation config manager started")
2838

2939

3040
def stop_workers():
41+
"""停止消费者工作进程"""
42+
# 停止S3配置管理器
43+
try:
44+
from server.workers.dynamic.isolation_config import isolation_config_manager
45+
isolation_config_manager.stop_polling()
46+
except Exception as e:
47+
logger.error(f"Error stopping S3 config manager: {e}")
48+
49+
# 停止通用消费者
3150
for consumer in consumers:
3251
consumer.stop()
52+
53+
# 停止所有动态消费者
54+
dynamic_manager.stop_all()
55+
56+
# 关闭线程池
3357
if executor:
3458
executor.shutdown()
3559

3660

61+
def _start_config_manager():
62+
"""启动S3配置管理器并设置变更回调"""
63+
from server.workers.dynamic.isolation_config import isolation_config_manager
64+
from server.workers.dynamic.dynamic_manager import sync_with_s3_config
65+
66+
# 添加配置变更回调
67+
isolation_config_manager.add_change_callback(_on_config_changed)
68+
69+
# 启动轮询
70+
isolation_config_manager.start_polling()
71+
72+
# 初始同步一次
73+
try:
74+
results = sync_with_s3_config(isolation_config_manager)
75+
logger.info(f"Initial S3 config sync results: {results}")
76+
77+
# 初始化 IsolationStateManager 状态
78+
from server.workers.dynamic.dynamic_manager import isolation_state
79+
isolated_spaces = isolation_config_manager.get_isolated_spaces()
80+
for space in isolated_spaces:
81+
isolation_state.add_isolated_space(space)
82+
logger.info(f"Initialized IsolationStateManager with spaces: {isolated_spaces}")
83+
84+
except Exception as e:
85+
logger.error(f"Error in initial S3 config sync: {e}")
86+
87+
88+
def _on_config_changed(old_config, new_config):
89+
"""配置变更回调函数"""
90+
try:
91+
from server.workers.dynamic.isolation_config import isolation_config_manager
92+
from server.workers.dynamic.dynamic_manager import sync_with_s3_config
93+
94+
logger.info("S3 configuration changed, syncing consumers...")
95+
96+
# 分析配置变更
97+
changes = isolation_config_manager.get_config_changes(old_config, new_config)
98+
logger.info(f"Config changes detected: {changes}")
99+
100+
# 同步消费者
101+
results = sync_with_s3_config(isolation_config_manager)
102+
logger.info(f"Consumer sync results: {results}")
103+
104+
# 同步到 IsolationStateManager
105+
from server.workers.dynamic.dynamic_manager import isolation_state
106+
for space in changes["added"]:
107+
isolation_state.add_isolated_space(space)
108+
for space in changes["removed"]:
109+
isolation_state.remove_isolated_space(space)
110+
logger.info(f"Updated IsolationStateManager: added={changes['added']}, removed={changes['removed']}")
111+
112+
113+
except Exception as e:
114+
logger.error(f"Error handling config change: {e}")
115+
116+
37117
__all__ = ['start_workers', 'stop_workers']

0 commit comments

Comments
 (0)