如何在Python Faust程序中配置不同Kafka消费者组或启动多应用

一、启动同一文件中的多个Faust应用 #

Faust命令行工具一次只能指定一个应用实例，你之前用逗号分隔多个应用的方式不被支持，所以需要分开启动worker进程：

方式1：多终端分别启动 #

打开两个终端窗口，分别执行以下命令：

faust -A test_faust:app1 worker -l info

faust -A test_faust:app2 worker -l info

每个命令对应启动一个应用实例，分别绑定consumer_group1和consumer_group2两个消费组。

方式2：单命令后台启动（适合服务器环境） #

如果想在单个终端里启动两个后台进程，可以用&符号让进程后台运行：

faust -A test_faust:app1 worker -l info &
faust -A test_faust:app2 worker -l info &

注意这种方式需要自行管理进程生命周期，避免意外退出。

二、同一应用内为不同主题配置独立消费组 #

你提到的consumer_id参数完全可以实现这个需求，该参数会覆盖默认的消费组（默认消费组等于应用ID），让不同主题使用独立的消费组。

修改后的代码示例 #

import faust

# 创建单个应用实例
app = faust.App(
    'main_application',
    broker='kafka://localhost:9092',
    value_serializer='json',
    consumer_auto_offset_reset='earliest'
)

# 为每个主题指定独立消费组
topic1 = app.topic('topic1', value_type=str, consumer_id='consumer_group1')
topic2 = app.topic('topic2', value_type=str, consumer_id='consumer_group2')

@app.agent(topic1)
async def process1(stream):
    async for value in stream:
        print(f'Group1处理消息: {value}')

@app.agent(topic2)
async def process2(stream):
    async for value in stream:
        print(f'Group2处理消息: {value}')

启动命令 #

只需要一条命令即可启动整个应用，两个agent会分别使用各自的消费组消费对应主题：

faust -A test_faust:app worker -l info

两种方案对比 #

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