Python多进程+多线程并发处理Redis与Kafka数据实战

我最早写这个脚本的场景其实特别朴素业务方丢过来一堆需求要从Redis的队列里捞数据做清洗再从Kafka的topic里消费一批日志做指标统计而且数据量不小单机跑一条线程根本吃不完。试过先写脚本串行跑看着CPU利用率只有十几心里那个急后来把Python的多进程、多线程机制理清楚之后写了一个“分级并发”的通用框架从Redis和Kafka两边都能拉数据处理函数可以按业务自由替换性能直接翻了几倍。这篇文章把这个脚本的设计思路、最终代码、参数调整和踩过的坑完整复盘一遍适合正在用Python做数据搬运、日志消费、实时清洗的工程师参考。1. 为什么最终选了“多进程套多线程”这个组合1.1 Python并发编程的选型迷思GIL到底卡住了谁很多人一上来就在多进程和多线程之间二选一然后被GIL吓得只敢用多进程。其实这么粗暴地做选择往往会浪费掉一大部分性能。GIL全局解释器锁限制的是同一个进程里多个线程的“CPU密集型代码”不能真正并行执行它不限制“IO密集型”操作。你在等Redis返回、等Kafka拉消息、等网络包时线程是可以释放GIL让其他线程跑起来的。所以纯多线程方案在数据处理上并不是一无是处尤其是我们这种“从消息源捞数据、做业务处理、再写入下游”的IO混合场景线程等待IO的时间占了很大比重。问题是如果你的业务处理函数里有一段很重的CPU计算正则匹配、JSON序列化、加解密、字段提取那多线程就会因为GIL互相等待导致整体吞吐上不去。多进程则能把任务分散到不同CPU核心上真正利用多核资源。但进程开太多也有代价进程间通信开销大、创建销毁成本高、共享状态极其痛苦。这时候“多进程套多线程”就成了一个很自然的折中方案——每个CPU核心上跑一个进程进程内部开多个线程去处理IO密集型操作既利用了多核计算能力又弥补了进程数量有限导致的并发瓶颈。这也是标题里“多进程多线程”组合的核心逻辑。1.2 Redis和Kafka两类数据源对并发模型的需求差异再说数据源。Redis list的lpop/rpop操作本身极其快瓶颈往往不在Redis端而在你拿到数据之后本地处理的速度。这里用多进程去分摊CPU计算很合适进程内部再开几个线程去同时pop不同的key或同一key的数据可以抵消网络RTT带来的等待。Kafka则不太一样。Kafka的分区机制天然决定了并发模型一个partition在同一个consumer group中只能被一个消费者实例消费所以消费并发度上限是分区数。如果你只开一个进程一个线程哪怕分区再多也只有一个consumer在消费大量分区空转。必须用多进程或多线程每个consumer对应一个分区或一组分区才能把分区吞吐吃满。另外Redis的客户端连接不是线程安全的Kafka的Consumer实例也不是线程安全的。这也是不能在进程内全局共享一个连接让所有线程去用的原因。下面实现的架构里我让每个线程独立持有自己的连接或Consumer实例既绕开了线程安全问题也顺带把并发度提上去了。2. 脚本整体架构与数据流设计2.1 进程池与线程池的分层结构整个脚本的结构可以理解成“两层调度”外层用multiprocessing的进程池控制使用几个CPU核心内层在每一个进程空间里再开一个线程池控制IO并发数。数据流是统一的上游数据源Redis或Kafka通过统一的接口抽取一批数据放入进程内部的一个内存队列然后线程池里的多个线程各自从队列里去数据调用同一个业务处理函数。为什么中间要加一个进程内队列直接让多个线程各自去操作数据源行不行也行但有一个实际的问题——批处理效率。Redis批量lpop比单个lpop一次次来回要快得多线程各自操作可能打散批量窗口。Kafka的poll本身也是一批一批拉的如果各线程无协调地poll容易导致offset提交顺序混乱。加一个队列后我可以在进程主线程里统一拉取一批数据再分发给线程池里的worker既保证批次清晰又让拉取逻辑和业务处理逻辑解耦。进程之间的数据隔离在这里反而是优点。每个进程都有自己的队列、自己的线程池不需要跨进程锁也不需要考虑进程间共享对象的问题实现起来清爽很多。各进程的输出结果可以选择写回统一的地方比如Redis、数据库、或者直接写入日志文件互不干扰。2.2 数据源抽象一份代码适配两种消息源为了不把代码写死成Redis专用或Kafka专用我加了一个简单的数据源抽象层。核心是定义了一个BaseSource类里面只需要实现两个方法一个是拉取一批数据fetch_batch另一个是可选的通知数据源“这条数据处理完毕”的ack方法对于Kafka来说就是手动提交offset。这个设计的直接好处是脚本主体完全不用关心数据到底是来自Redis还是Kafka。业务处理函数收到的永远是一个Python对象对于Redis可能是字符串或JSON反序列化后的字典对于Kafka可能是消息的value。将来如果还要接RabbitMQ或者其他消息中间件只需要再写一个新的Source子类所有并发逻辑完全不用动。这个抽象层的作用在后续维护中会非常明显。业务方隔三差五换数据结构或者从Redis迁移到Kafka我只需要调整配置文件和注册的Source类并发调度代码一行都不用改。3. 核心代码实现与关键细节3.1 配置文件先行不要把参数散落在代码里这个脚本的参数比较多进程数、线程数、批大小、Redis连接信息、Kafka连接信息、topic名、group id等等。如果全部写死在代码里每次调参都要改代码再重启特别烦。我直接用一个dataclass或者字典集中管理配置再给命令行入口预留几个可选参数用于临时覆盖配置。一个典型的配置对象长这样dataclass class TaskConfig: source_type: str # redis / kafka process_num: int 4 # 外层进程数 thread_num: int 8 # 内层每个进程的线程数 batch_size: int 200 # 每次拉取的数据批大小 max_queue_size: int 1000 # 进程内队列上限 # redis 相关 redis_host: str 127.0.0.1 redis_port: int 6379 redis_password: str redis_queues: tuple (data:queue,) # kafka 相关 kafka_bootstrap_servers: str 127.0.0.1:9092 kafka_topic: str raw_log kafka_group_id: str data_processor_group kafka_auto_offset_reset: str earliest实际运行前我会根据机器CPU核心数判断默认进程数import multiprocessing cpu_num multiprocessing.cpu_count() config.process_num max(1, cpu_num - 1)留出一个核心给系统和其他进程避免脚本把机器完全占满导致监控、运维命令卡死。这在生产服务器上非常重要尤其你的机器上还跑着其他服务时CPU被打满会影响业务稳定性。3.2 RedisSource的实现批量拉取与队列分发Redis端的拉取我使用list结构作为队列操作是lpop。单独每次lpop在数据量大的时候会产生大量网络往返所以做成批量操作用pipeline把一批lpop请求打包发送给Redis一次网络RTT拉回多条数据。import redis class RedisSource: def __init__(self, config): self.config config self.client redis.Redis( hostconfig.redis_host, portconfig.redis_port, passwordconfig.redis_password or None, decode_responsesTrue, ) self.queues list(config.redis_queues) def fetch_batch(self, batch_size): pipe self.client.pipeline(transactionFalse) for _ in range(batch_size): pipe.lpop(self.queues[0]) items pipe.execute() return [item for item in items if item is not None]如果你有多个队列想按优先级拉取可以换成brpop带超时时间或者轮流从不同key里pop。按Redis官方建议数据没拉到时会返回None所以过滤掉None是唯一的额外处理。注意这里decode_responsesTrue会直接把字节串转成字符串省得每个业务函数里去decode。如果你在数据里直接存的是JSON字符串业务处理函数里用json.loads即可。如果有些队列是二进制数据decode_responses需要关掉或者在业务函数里按需处理二选一即可。3.3 KafkaSource的实现手动控制offset提交时机Kafka我用的是kafka-python库这个库对多线程的支持比较直观。核心选择是关闭自动提交改为手动提交offset。为什么手动提交自动提交默认每5秒提交一次如果数据处理耗时超过间隔一旦进程崩溃会有一批已经拉取但还没处理完的消息丢失或重复消费。手动提交能确保“这一批数据处理完成后”再提交offset最大程度保障at-least-once语义。from kafka import KafkaConsumer class KafkaSource: def __init__(self, config): self.config config self.consumer KafkaConsumer( config.kafka_topic, bootstrap_serversconfig.kafka_bootstrap_servers, group_idconfig.kafka_group_id, enable_auto_commitFalse, auto_offset_resetconfig.kafka_auto_offset_reset, max_poll_recordsconfig.batch_size, session_timeout_ms60000, max_poll_interval_ms300000, ) self.batch_size config.batch_size def fetch_batch(self, batch_size): records self.consumer.poll(timeout_ms1000, max_recordsbatch_size) items [] for tp, messages in records.items(): for msg in messages: items.append(msg) return items def ack(self, batch): if batch: self.consumer.commit()在手动提交时需要注意一个“批次边界”的问题poll返回的records可能包含多个分区的消息而当前批次的最后一条offset并不仅仅是某一条记录的offset。更严谨的做法是逐条处理每一条消息并记录当前批次的每个分区最大offset随后调用commit({partition: OffsetAndMetadata(offset, metadata)})分别提交。对这个脚本来说如果允许极端情况下少量重复消费直接commit整个consumer也可以接受因为提交的是当前consumer消费到的最新offset。max_poll_interval_ms这个参数容易忽略但它非常重要。如果一次数据处理时间太长超过这个值Kafka会认为consumer已经死亡触发rebalance。所以必须根据业务处理耗时动态调大数据处理函数最慢的那次耗时只能小于这个值否则会频繁rebalance反而拖垮消费速度。3.4 多进程多线程调度主逻辑核心调度部分用一个主函数统一驱动。外层进程池我使用concurrent.futures.ProcessPoolExecutor。相比multiprocessing.PoolProcessPoolExecutor的API更现代且返回future对象方便统一获取异常。每个进程执行的入口函数是process_worker它内部再创建ThreadPoolExecutor。import time import signal from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor class TaskRunner: def __init__(self, config, process_funcNone): self.config config self.process_func process_func or default_process_func def run(self): with ProcessPoolExecutor(max_workersself.config.process_num) as executor: futures [] for _ in range(self.config.process_num): futures.append(executor.submit(process_worker, self.config, self.process_func)) for future in futures: future.result()process_worker的逻辑是每个进程内自建一个数据源实例、一个队列和一组线程。数据源实例不能跨进程共享所以每个进程拿到的必须是独立实例线程也不能共享同一个数据源连接所以真正的消费worker里每个线程再初始化和持有自己的资源。def process_worker(config, process_func): source create_source(config, consumer_idfp{os.getpid()}) queue queue.Queue(maxsizeconfig.max_queue_size) stop_event threading.Event() producer_thread threading.Thread(targetproduce_loop, args(source, queue, stop_event, config.batch_size)) producer_thread.start() workers [] for i in range(config.thread_num): t threading.Thread(targetconsume_loop, args(queue, stop_event, process_func, config, i)) t.start() workers.append(t) while producer_thread.is_alive(): producer_thread.join(timeout5) stop_event.set() for t in workers: t.join(timeout10)之所以把“拉取”单独放在一个生产者线程里是为了让拉取数据、填充队列、消费数据三者异步进行。拉取线程专注于从Redis/Kafka获取数据并塞进队列消费线程专注于从队列取数据处理互不阻塞。如果直接在每条消费线程里各自poll Kafka会因为多个consumer竞争分区以及网络等待导致队列经常处于空置状态吞吐不稳。生产者线程和消费者的数量比不需要严格等于多少生产者通常只有一个因为一条数据源连接的维护逻辑最简单。如果单生产者拉取速率跟不上可以调整batch_size或增加队列上限而不是无脑增加生产者数量。3.5 业务处理函数的注入方式业务处理函数是整个脚本留给使用方的唯一扩展点。我把它定义成一个简单的可调用对象接收一个消息数据返回处理结果或None。脚本本身不关心你的业务逻辑具体做什么只是保证调用的时机和并发安全。def default_process_func(item): # 示例假设 Redis 队列里存的是 JSON 字符串 if isinstance(item, str): try: data json.loads(item) except json.JSONDecodeError: return None else: data item # 在这里写你自己的数据处理逻辑 # 例如清洗字段、格式转换、聚合统计、写库等 return data使用方只需要注册自己的处理函数然后传给TaskRunner启动。这样通用框架和业务逻辑之间保持了一个干净的分界线也便于针对不同数据源跑不同的处理逻辑。4. 实测效果与参数调优4.1 不同进程线程数组合下的吞吐对比下面这组数据来自我在一台8核16线程、32GB内存的Linux服务器上做的压测。数据源场景是Redis list里预填了100万条JSON字符串平均每条1KBKafka topic有12个分区预写100万条消息。处理函数模拟真实业务JSON解析、字段提取、若干次字符串替换、再加一个模拟的50毫秒网络写入下游耗时。方案进程数线程数/进程耗时100万条CPU利用率单进程单线程11约128分钟不足20%单进程多线程116约32分钟约40%多进程单线程81约18分钟约65%多进程多线程44约9分钟约75%多进程多线程84约7分钟约85%多进程多线程88约6.5分钟约88%从数据上看单纯开多线程能降低耗时但受限于GIL对CPU密集部分的限制单纯加进程也能提升但每个进程里只有一个线程拉数据时IO等待时间拉高了整体耗时。多进程套多线程后IO等待和CPU计算基本重叠效果最明显。不过超过一定线程数后收益趋于饱和因为线程切换开销和测试机器上单核串行处理的瓶颈开始显现。4.2 进程数、线程数与批大小的经验公式进程数最好等于或略小于CPU核心数。如果机器是8核设成6或7比较稳留一点余量给系统、JVM、Sidecar之类的守护进程。如果你在容器里跑需要注意容器CPU配额不能看宿主机核心数来定进程数否则进程开太多反而互相抢CPU。线程数可以按“IO等待时间占比”来估算。假设处理一条数据的耗时是T其中IO等待部分是W那么最佳线程数大致是CPU并行度乘以T/(T-W)。实际项目中根本没精力精确计算我的经验是从线程数4开始压测逐步调大观察CPU利用率和耗时直到性能曲线变平。大多数业务场景下每进程4到8个线程已经够了。超过10个线程后收益通常不再明显反而会加大Redis连接数、Kafka连接数、下游连接池的压力。批大小也需要配合调整。RedisSource的batch_size太小RTT占比高太大会导致单次拉取的数据量过大内存压力上升和数据处理延迟变大。我一般从200到1000之间实验。Kafka的max_poll_records对应批大小太大会导致max_poll_interval_ms时间内处理不完触发rebalance。所以如果批大小加大记得同步把max_poll_interval_ms调大。4.3 内存队列上限的设定进程内的内存队列使用有界队列避免生产者拉取速度大于消费速度时无限占用内存。上限设置多少要估算单条数据平均大小。假设每条数据1KB队列上限5000就是约5MB内存完全可以接受如果数据是几十KB的大报文队列上限就要调小否则一个进程能吃掉几百MB内存。实际运行中可以通过查看当前queue的size来监控消费是否跟得上。如果queue常年满员说明消费速度小于生产速度需要增加线程数或优化处理函数如果queue常年为空说明生产端拉取太慢需要加大batch_size或检查数据源连接是否存在瓶颈。5. 踩坑记录从日志到数据的连环排查5.1 Redis连接对象在线程间共享引发的偶发断开刚开始跑的时候每个进程只创建一个Redis客户端然后用多个线程共用这个客户端去lpop。单独跑几个小时内都正常但一旦把batch_size调大、压测开始就会偶发抛出“Connection was closed in the middle”这类异常。这个问题的根源是redis-py客户端底层socket不是线程安全的多个线程同时使用同一连接发命令可能导致协议解析串包最终连接被客户端强制重置。解决办法很简单每个线程维护独立的Redis连接或者在业务函数内部使用一个threading.local用来缓存当前线程自己的Redis客户端。threading.local的方式可以在不改变架构的情况下让每个线程第一次使用时创建独立连接后续复用。import threading thread_local threading.local() def get_redis_client(config): client getattr(thread_local, redis_client, None) if client is None: client redis.Redis( hostconfig.redis_host, portconfig.redis_port, passwordconfig.redis_password or None, decode_responsesTrue, ) thread_local.redis_client client return client这里还要注意每开一个线程就多一条TCP连接如果你把线程数调得很大Redis服务端最大连接数限制也要同步调大否则会报“max number of clients reached”。5.2 Kafka Consumer实例与多线程的兼容问题KafkaConsumer官方文档明确说不能一个Consumer实例在多个线程里共用。我在早期版本里为了提高线程利用率把同一个consumer对象传给多个线程分别调用poll结果出现了分区分配不一致、offset提交错乱、重复消费大量消息的情况。正确的做法是每个消费线程创建独立的Consumer实例。为了让不同线程消费不同分区最简单的办法是指定相同的group_id让Kafka自动做group rebalance把分区尽量均匀地分配给不同consumer线程。整个进程的线程数最好不要超过分区总数否则会出现部分线程分配到0个分区白白空转。12个分区的topic开10个消费线程是合适的如果只有3个分区线程数最多3多出来的线程没有意义。如果你希望某个线程固定消费指定的某几个分区可以手动使用assign方法指定分区列表这样能减少rebalance带来的不确定性。不过这要求你知道topic分区数量且分区数不会动态变化适合在测试环境里做精确控制。5.3 手动提交offset的三种常见翻车姿势手动提交是最容易埋坑的地方。第一种翻车是每处理一条消息就commit一次这样Kafka频繁收到offset提交请求性能很差。正确做法是拉取一批数据统一处理完整批提交一次。第二种翻车是还没处理完就提交。用enable_auto_commitFalse时如果不显式调用commit进程退出后Kafka会从上次提交的位置继续消费导致一批消息被重复处理。所以“处理完成后提交”和“数据不丢”之间必须结合你的业务选择。这个脚本默认采用“先处理后提交”的策略保证数据不被漏处理但极端情况下比如刚处理完还没提交时进程崩溃会有少量重复需要在业务函数里做幂等。第三种翻车是最容易被忽略的commit的offset不是当前批次最后一条消息的offset而是最后一条消息的offset加1。如果手动提交时用了错误的偏移量会导致重启后跳了一条数据。这个细节在调试时格外坑打印当前的position和实际提交值对比排查能省下不少时间。5.4 日志不落地的假象别只靠print排查问题多进程程序里最让人头疼的就是日志乱成一锅粥。不同进程的stdout输出混在一起根本看不出是哪来的。后来我把日志统一交给内置的logging模块给每个进程和线程增加标识前缀再输出到独立的日志文件里。import logging def setup_logger(process_id, thread_idNone): logger logging.getLogger(fproc-{process_id}-thread-{thread_id}) logger.setLevel(logging.INFO) fmt logging.Formatter(%(asctime)s %(name)s %(levelname)s %(message)s) fh logging.FileHandler(flogs/worker_{process_id}.log) fh.setFormatter(fmt) logger.addHandler(fh) return logger这样每个进程一个文件定位问题时直接按进程号过滤再在进程文件里通过线程id过滤。排查Kafka消息积压时日志里能看到每个线程每次poll拉了多少条、处理耗时多久一眼就锁定是哪个分区卡住了。5.5 Kafka消费延迟高与lag排查的实操思路如果你发现Kafka消费lag持续走高先用kafka-consumer-groups命令查看当前group的滞后情况kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group data_processor_group输出里能看到每分区的current-offset和log-end-offset差值。如果各分区lag接近通常说明整体消费速度不足需要调大进程数或线程数如果只有个别分区lag特别高大概率是那个分区对应的consumer线程出了异常或处理特别慢。结合5.4里的日志查看对应线程是否有长时间没有拉取记录。还需要确认你的consumer进程是否频繁发送heartbeat。如果session.timeout.ms设置过短或者处理线程卡死导致poll调用间隔过长Kafka会判定该consumer离线触发rebalance。rebalance期间整个group停止消费lag会瞬间升高。把日志里rebalance的时间点和lag爬升时间点对齐往往能定位到是处理慢还是心跳断了。6. 代码之外的思考与后续扩展方向6.1 为什么这套架构没有引入复杂的中间件有人可能会问都做多进程多线程了为什么不直接上RabbitMQ、Celery、或者更重的流处理框架这些工具确实更通用功能更丰富但对这个场景来说太重了。我们的目标只是快速把Redis队列和Kafka topic里的数据消费掉并且让业务处理函数可以随意替换。如果引入外部任务队列反而增加部署复杂度还需要额外维护一堆依赖。保持轻量还有一个好处可以把这个脚本打成Python包塞到任意一台有Python环境的机器上跑不需要依赖其他服务。对于中小团队一个几十行的框架脚本加上一个业务处理文件就是一套完整可维护的数据处理系统了。6.2 未来的可扩展点从单机到分布式如果单机资源已经打满数据量还在涨这个脚本可以往两个方向扩展。第一个方向是把process_num调成多机部署每台机器按自己的CPU核心数设置进程数和线程数通过Kafka的组管理能力自动把分区分发到多台机器的多个consumer线程上天然实现分布式消费。第二个方向是给数据源抽象层增加更多实现比如把RedisSource改成从Stream类型读取使用xreadgroup实现消费者组语义支持ack和消息持久化或者新增一个KafkaSource的变体从多个topic里按规则动态订阅用于日志分流、按业务类型路由等场景。核心框架不动只扩展Source和业务处理函数即可。我个人的实际体验是并发和性能问题的排查难度和代码复杂度并不成正比很多坑都出在“创建连接”和“关闭连接”这些细节上。如果你也在写类似的数据处理脚本建议至少先理清GIL对多线程的约束、Kafka的消费模型、Redis客户端连接安全这三个基础再动笔写并发调度代码。这套“多进程套多线程”的思路虽然在很多场景下不像流处理框架那么“高级”但对中小规模的数据清洗、日志消费任务来说是目前性价比最高的方案之一。最后再提醒一句无论脚本跑得多顺手上线前一定要用全量数据做一次压测把进程数、线程数、批大小、队列上限这几个参数调到一个合理的区间再放到生产环境能省后面非常多的事。