apache / apache/rocketmq-client-python

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

Đang mở
#153 0 bình luận 0 reaction 0 người được giao Xem trên GitHub
Ngôn ngữ chính
Python
Star
305
Fork
98
Chỉ số merge pull request
Không có pull request nào được merge trong 30 ngày

Mô tả

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

Hướng dẫn đóng góp

Mở hướng dẫn đóng góp

Đánh giá

Issue này chưa được đánh giá.

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.