apache / apache/rocketmq-client-python

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

Open
#153 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Python
Stars
305
Forks
98
PR merge metrics
No merged PRs in 30d

Description

`
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)

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.