如何在Memgraph+Cypher环境下创建AMQP消息并触发节点变更通知？

1. 如何在Memgraph与Cypher环境下创建AMQP消息？ #

Memgraph原生Cypher不支持直接发送AMQP消息，需通过自定义查询模块实现：

• 用Python/Go/C++编写查询模块，集成AMQP客户端库（比如Python的pika、Go的amqp包），实现发送AMQP消息的函数。
• 将模块文件放到Memgraph的查询模块目录（默认路径为/usr/lib/memgraph/query_modules/）。
• 在Memgraph中加载模块后，直接在Cypher中调用函数发送消息：

  CALL amqp.send_amqp_message('exchange_name', 'routing_key', 'message_content') YIELD *;
  

示例Python查询模块（amqp.py）：

import mgp
import pika

@mgp.read_proc
def send_amqp_message(exchange: str, routing_key: str, message: str) -> mgp.Record(status=str):
    try:
        connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
        channel = connection.channel()
        channel.exchange_declare(exchange=exchange, exchange_type='direct')
        channel.basic_publish(exchange=exchange, routing_key=routing_key, body=message)
        connection.close()
        return mgp.Record(status="Message sent successfully")
    except Exception as e:
        return mgp.Record(status=f"Failed to send message: {str(e)}")

2. 特定节点增删时生成AMQP消息的最优方案 #

最优方案是结合Memgraph触发器与自定义查询模块：
Memgraph触发器仅支持Cypher，但可通过Cypher调用查询模块函数，间接触发Python/Go/C++逻辑。具体步骤：

1. 编写并加载上述AMQP查询模块。
2. 创建针对特定节点标签的触发器，在节点增删时调用AMQP发送函数：

监控节点插入的触发器 #

CREATE TRIGGER node_created_trigger
ON CREATE NODE
WHERE (n:TargetLabel)  -- 替换为你要监控的节点标签
EXECUTE CALL amqp.send_amqp_message(
    'node_events', 
    'node.created', 
    json_set('{}', '$.node_id', id(n), '$.properties', properties(n))
) YIELD *;

监控节点删除的触发器 #

CREATE TRIGGER node_deleted_trigger
ON DELETE NODE
WHERE (n:TargetLabel)
EXECUTE CALL amqp.send_amqp_message(
    'node_events', 
    'node.deleted', 
    json_set('{}', '$.node_id', id(n))
) YIELD *;

关于触发脚本的说明 #

虽然触发器本身仅执行Cypher，但通过调用查询模块的自定义函数，完全可以触发Python/Go/C++脚本逻辑——查询模块本身就是用这些语言编写的，函数执行时会运行对应的代码逻辑。

内容的提问来源于stack exchange，提问作者Moraltox