监听器无法访问 hazelcast 队列答案

【问题标题】：hazelcast queue is not accesssible from listener监听器无法访问 hazelcast 队列
【发布时间】：2023-03-31 05:57:01
【问题描述】：

我有一个名为 (train) 的消息队列，每当向其中添加任何内容时，我都想创建一个新队列或执行一些操作。所以我已将侦听器附加到队列，并从侦听器开始新队列。但问题是我无法从侦听器启动队列，因为它抛出了一些错误。

'''
This function will be called when we add anything to message_queue.
'''
def notify_when_added(event):
    res = event.item
    if res.status == Activity.train:
        print(res.text)
    elif res.status == Activity.predict:
        print('predict')
        queue = MessageQueue.create_msg_queue("predict") #error in this line



if __name__ == "__main__":
    #load_predictor()
    train_queue = MessageQueue.create_msg_queue("train")

    train_queue.add_listener(True, notify_when_added) #creating a listener
    res = Response(Activity.predict, "Model Trained with test accuracy of: ", "coles_auto_4", 5002)
    train_queue.add_all([res])
    app.run(threaded=True)

所以基本上我在这里添加一些东西到 train_queue 并且想要创建将被其他进程使用的新队列。但我收到以下错误。仅当我使用 add_listener 并尝试在侦听器中创建或访问队列时才会发生此错误。我想从侦听器开始一个队列，其他进程可以使用它。

Traceback (most recent call last):
  File "C:\Users\shiva_burade\AppData\Roaming\Python\Python36\site-packages\hazelcast\cluster.py", line 124, in _connect_to_cluster
    self._connect_to_address(address)
  File "C:\Users\shiva_burade\AppData\Roaming\Python\Python36\site-packages\hazelcast\cluster.py", line 171, in _connect_to_address
    connection = f.result()
  File "C:\Users\shiva_burade\AppData\Roaming\Python\Python36\site-packages\hazelcast\future.py", line 58, in result
    self._reactor_check()
  File "C:\Users\shiva_burade\AppData\Roaming\Python\Python36\site-packages\hazelcast\future.py", line 70, in _reactor_check
    "Synchronous result for incomplete operation must not be called from Reactor thread. "
RuntimeError: Synchronous result for incomplete operation must not be called from Reactor thread. Use add_done_callback instead.
Aug 27, 2019 05:41:19 PM HazelcastClient.ClusterService
WARNING: [3.12.1] [dev] [hz.client_1] Unable to get alive cluster connection, attempt 2 of 2
Aug 27, 2019 05:41:19 PM HazelcastClient.InvocationService
WARNING: [3.12.1] [dev] [hz.client_0] Error handling event ClientMessage:{length=327, correlationId=6, messageType=204, partitionId=216, isComplete=True, isRetryable=False, isEvent=True, writeOffset=22}
Traceback (most recent call last):
  File "C:\Users\shiva_burade\AppData\Roaming\Python\Python36\site-packages\hazelcast\invocation.py", line 293, in _handle_event
    invocation.event_handler(message)
  File "C:\Users\shiva_burade\AppData\Roaming\Python\Python36\site-packages\hazelcast\proxy\queue.py", line 91, in <lambda>
    lambda m: queue_add_listener_codec.handle(m, handle_event_item),
  File "C:\Users\shiva_burade\AppData\Roaming\Python\Python36\site-packages\hazelcast\protocol\codec\queue_add_listener_codec.py", line 48, in handle
    handle_event_item(item=item, uuid=uuid, event_type=event_type)
  File "C:\Users\shiva_burade\AppData\Roaming\Python\Python36\site-packages\hazelcast\proxy\queue.py", line 85, in handle_event_item
    item_added_func(item_event)
    File "controller_flask.py", line 33, in notify_when_added
queue = MessageQueue.create_msg_queue("predict")
  File "D:\text category\message_queue.py", line 56, in create_msg_queue
    client = hazelcast.HazelcastClient(config)
  File "C:\Users\shiva_burade\AppData\Roaming\Python\Python36\site-packages\hazelcast\client.py", line 59, in __init__
    self._start()
  File "C:\Users\shiva_burade\AppData\Roaming\Python\Python36\site-packages\hazelcast\client.py", line 64, in _start
    self.cluster.start()
  File "C:\Users\shiva_burade\AppData\Roaming\Python\Python36\site-packages\hazelcast\cluster.py", line 50, in start
    self._connect_to_cluster()
  File "C:\Users\shiva_burade\AppData\Roaming\Python\Python36\site-packages\hazelcast\cluster.py", line 142, in _connect_to_cluster
    raise HazelcastError(error_msg)
hazelcast.exception.HazelcastError: Could not connect to any of {Address(host=127.0.0.1, port=5701)} after 2 tries

【问题讨论】：

标签： python message-queue hazelcast

【解决方案1】：

它可能无法完全回答问题，但是当我想在从另一个主题收到消息时调用的回调函数中从一个主题获取/订阅/取消订阅时，我也收到了以下错误。

不完整操作的同步结果不能从反应堆线程。请改用 add_done_callback。

显然，某些暗示 Hazelcast 客户端的操作无法在回调函数（“Reactor”线程）内执行。由于我无法在任何这些操作中添加 add_done_callback，我最终在回调函数中设置了一个变量，并在主函数中检查了这个变量的状态（例如在循环中）。这当然不是最方便/最有效的方法，但至少有一种解决方法......

【讨论】：