[I] 同时订阅多个server_address的topic，产生的问题 [rocketmq-client-python]

["kennard520 (via GitHub)" <gi...@apache.org>]

kennard520 opened a new issue, #153:
URL: https://github.com/apache/rocketmq-client-python/issues/153

   `
   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上创建了两个消费者
   ![image](https://github.com/user-attachments/assets/59327d86-aa27-4396-98ec-3e315328719d)
   
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@rocketmq.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org