同时订阅多个server_address的topic，产生的问题

`
class RocketMQListener:
def __init__(self, server_address, topic, group_id, mq_tag, callback):
self.server_address = server_address
self.topic = topic
self.group_id = group_id
self.mq_tag = mq_tag
self.callback = callback
self.consumer = None
self.stop_event = threading.Event()

def start(self):
self.consumer = PushConsumer(self.group_id)
self.consumer.set_group(self.group_id)
self.consumer.set_thread_count(1)
self.consumer.set_name_server_address(self.server_address)
self.consumer.subscribe(self.topic, self.process_message, self.mq_tag)
self.consumer.start()
logging.info(f"Started RocketMQ listener for {self.server_address} topic {self.topic} and group {self.group_id} and mq_tag {self.mq_tag}.")

while not self.stop_event.is_set():
time.sleep(5)
pass

self.consumer.shutdown()

def process_message(self, message):
return self.callback(message)

def stop(self):
self.stop_event.set()`

`import threading
from contextlib import asynccontextmanager

import uvicorn
from fastapi import FastAPI
from starlette.responses import JSONResponse

from rocketmq_listener import RocketMQListener
from rocketmq.client import PushConsumer, ConsumeStatus

app = FastAPI()
brokers_and_topics = [
("192.168.0.137:9876", "kl-video-editing-l", "kl-group-video-process-15", "video_process_create_tag"),
("192.168.0.138:9876", "kl-video-editing", "kl-group-video-process-16", "video_process_create_tag")
]

listeners=[]

def test(msg):
print(msg)
return ConsumeStatus.RECONSUME_LATER

def start_listener(server_address, topic, group_id, tag):
listener = RocketMQListener(server_address, topic, group_id, tag, test)
listener.start()
listeners.append(listener)
@asynccontextmanager
async def lifespan(app: FastAPI):
for topic in brokers_and_topics:
threading.Thread(target=start_listener, args=topic).start()
yield
# Clean up the ML models and release the resources
app = FastAPI(lifespan=lifespan)

if __name__ == '__main__':
try:
uvicorn.run("main:app", host="0.0.0.0", port=5007, log_level="warning", workers=1)
except Exception as e:
print(e)
`如上所示，我用两个线程分别订阅192.168.0.137和192.168.0.138的topic，而且groupid不一样，最后却在192.168.0.137上创建了两个消费者

Start with RocketMQListener.start in rocketmq_listener and the FastAPI lifespan in main, then reproduce the two-thread setup using the two server addresses and PushConsumer instances shown. Inspect the resulting consumer registrations and verify whether each listener uses only its supplied server_address; done means the expected consumer is created for each broker without an extra consumer on 192.168.0.137.

Scritto dal modello di indicizzazione a partire dal testo della issue.