Kafka是否支持双向通信？双向通信配置异常求助

我有两台服务器需要实现双向通信，基于Kafka配置了2个Topic、2个Producer和2个Consumer：

• 服务器1到服务器2：使用sockets Topic，对应producer.py中的Producer和Consumer.py中的Consumer
• 服务器2到服务器1：使用game_events Topic，对应Consumer.py中的Producer和producer.py中的Consumer

目前遇到的问题是：Consumer仅能接收sockets Topic的消息，无法接收game_events Topic的消息。

原代码 #

producer.py #

from kafka import KafkaProducer, KafkaConsumer
import gevent
import time
producer = KafkaProducer(bootstrap_servers='localhost:9092')
consumer = KafkaConsumer('game_events', bootstrap_servers='localhost:9092')

def prod():
    while True:
        time.sleep(2)
        producer.send('sockets', b'Hello  world') 
        producer.flush()
    

def cons():
    while True:
        game_server = consumer.__next__()
        print(game_server)


gevent.spawn(prod())
gevent.spawn(cons())

Consumer.py #

from kafka import KafkaConsumer, KafkaProducer
import gevent
import time
consumer = KafkaConsumer('sockets', bootstrap_servers=['localhost:9092'])
producer = KafkaProducer(bootstrap_servers='localhost:9092')

def cons():
    while(True):
        game_server = consumer.__next__()
        print(game_server)


def prod():
    while True:
        time.sleep(2)
        print("game_server")
        producer.send('game_events', b'Hello  Game Event') 
        producer.flush()

gevent.spawn(cons())
gevent.spawn(prod())

问题原因 #

核心错误在于gevent.spawn的调用方式：你传入的是prod()和cons()（函数执行结果），而不是函数对象本身。这会导致：

• 主线程直接执行prod()，进入while死循环，后续的gevent.spawn(cons())根本不会被执行
• 因此producer.py中的Consumer（订阅game_events）从未启动，自然接收不到消息

另外需要明确：Kafka完全支持这种双向通信场景，你的整体架构思路是可行的，只是代码存在调用错误。

解决方法 #

修正gevent.spawn的调用，去掉函数后的括号，直接传递函数对象；同时添加gevent.wait()保持主线程运行，避免进程直接退出。

修正后的producer.py #

from kafka import KafkaProducer, KafkaConsumer
import gevent
import time
producer = KafkaProducer(bootstrap_servers='localhost:9092')
consumer = KafkaConsumer('game_events', bootstrap_servers='localhost:9092')

def prod():
    while True:
        time.sleep(2)
        producer.send('sockets', b'Hello  world') 
        producer.flush()
    

def cons():
    while True:
        game_server = consumer.__next__()
        print(game_server)


# 传递函数对象，不要加括号执行
gevent.spawn(prod)
gevent.spawn(cons)
# 保持主线程运行，避免进程退出
gevent.wait()

修正后的Consumer.py #

from kafka import KafkaConsumer, KafkaProducer
import gevent
import time
consumer = KafkaConsumer('sockets', bootstrap_servers=['localhost:9092'])
producer = KafkaProducer(bootstrap_servers='localhost:9092')

def cons():
    while(True):
        game_server = consumer.__next__()
        print(game_server)


def prod():
    while True:
        time.sleep(2)
        print("game_server")
        producer.send('game_events', b'Hello  Game Event') 
        producer.flush()

# 传递函数对象，不要加括号执行
gevent.spawn(cons)
gevent.spawn(prod)
# 保持主线程运行，避免进程退出
gevent.wait()

验证步骤 #

1. 确保Kafka服务正常运行，两个Topic（sockets、game_events）已创建
2. 分别启动修正后的producer.py和Consumer.py
3. 观察控制台输出：
   • producer.py会打印来自game_events Topic的消息
   • Consumer.py会打印来自sockets Topic的消息
   • 双向通信即可正常工作

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