rocketmq-client.zip

# _*_ coding: utf-8 _*_ import time from multiprocessing import Process, Queue from jpype import * import bigdata.mq.conf.settings_MQ as settings from bigdata.mq.consumer.consumer_form_where import get_from_where from bigdata.mq.consumer.consumer_config import ConsumerConfig class PushConsumer: """ push方式消费RocketMQ """ def __init__(self, server, topic, group_id): """ 初始化push consumer :param consumer_config: ConsumerConfig对象 :param messageHandler: MessageHandlerInterface实现类 """ self.config = ConsumerConfig(server, topic, group_id) self.queue = Queue() def _get_message(self): """ 启动jvm，接受RocketMQ数据 :param consumer_config: :param messageHandler: :return: """ # 获取默认jvm路径，并启动jvm jvmPath = getDefaultJVMPath() startJVM( jvmPath, settings.JVM_HEAP_XMS, settings.JVM_HEAP_XMX, settings.JVM_HEAP_XMN, settings.JAVA_EXT_DIRS ) # 获取java com.clife.utils.mq.consumer.ConsumerConfig2对象，并初始化 ConsumerConfig2 = JPackage("com.clife.utils.mq.consumer").ConsumerConfig2 consumerConfig2 = ConsumerConfig2() consumerConfig2.setNamesrvAddr(self.config.namesrv_addr) consumerConfig2.setTopic(self.config.topic) consumerConfig2.setGroupName(self.config.group_name) consumerConfig2.setTags(self.config.tag) consumerConfig2.setWhereFrom(get_from_where(self.config.from_where)) # 定义java接收到数据后的回调，将数据加载到Queue中 def handle(msg): num = msg.size() n = 0 while n < num: message = msg.get(JInt(n)).getMessage() n = n + 1 self.queue.put(message) fun = {"handle": handle} # 利用代理，定义java接口实现类 proxy = JProxy("com.clife.utils.mq.consumer.handler.ConsumerMsgHandler", dict=fun) Consumer = JClass("com.clife.utils.mq.consumer.Consumer") consumer = Consumer(consumerConfig2, proxy) consumer.init() # 阻塞当前进程 while True: time.sleep(10) def start(self): self.process = Process(target=self._get_message) self.process.start() def close(self): self.process.terminate() def __iter__(self): while True: message = self.queue.get() yield message