这几乎是节点lib amqp-match的直接端口：

import re

def amqp_match(key: str, pattern: str) -> bool:
    if key == pattern:
        return True
    replaced = pattern.replace(r'*', r'([^.]+)').replace(r'#', r'([^.]+.?)+')
    regex_string = f"^{replaced}$"
    match = re.search(regex_string, key)
    return match is not None

我有一些Java代码，如果可以帮助您的话

Pattern toRegex(String pattern) {
    final String word = "[a-z]+";

    // replace duplicate # (this makes things simpler)
    pattern = pattern.replaceAll("#(?:\\.#)+", "#");

    // replace *
    pattern = pattern.replaceAll("\\*", word);

    // replace #

    // lone #
    if ("#".equals(pattern)) return Pattern.compile("(?:" + word + "(?:\\." + word + ")*)?");

    pattern = pattern.replaceFirst("^#\\.", "(?:" + word + "\\.)*");
    pattern = pattern.replaceFirst("\\.#", "(?:\\." + word + ")*");

    // escape dots that aren't escapes already
    pattern = pattern.replaceAll("(?<!\\\\)\\.", "\\\\.");

    return Pattern.compile("^" + pattern + "$");
}

也许有人能把它翻译成python。

我们使用这个模式将RabbitMQ模式转换为正则表达式：

from typing import Pattern

def convert(pattern: str) -> Pattern:
    pattern = (
        pattern
        .replace('*', r'([^.]+)')
        .replace('.#', r'(\.[^.]+)*')
        .replace('#.', r'([^.]+\.)*')
    )
    return re.compile(f"^{pattern}$")

我正在使用这个java函数：

public boolean matchRoutingPatterns(String routingTopic, String routingKey)
{
    // replace duplicates
    routingTopic = routingTopic.replaceAll("#(?:\\.#)+", "#").replaceAll("([#*])\\1{2,}", "$1");
    routingKey   = routingKey.replaceAll("#(?:\\.#)+", "#").replaceAll("([#*])\\1{2,}", "$1");

    String[] routingTopicPath = Strings.splitList( routingTopic.replace('.', ','));
    String[] routingKeyPath   = Strings.splitList( routingKey.replace('.', ','));

    int i=0;
    int j=0;
    while (i<routingTopicPath.length && j<routingKeyPath.length)
    {
        if ("#".equals(routingTopicPath[i]) || "#".equals(routingKeyPath[j]))
        {
            if (routingTopicPath.length-i > routingKeyPath.length-j)
                { i++; }
            else
            if (routingTopicPath.length-i < routingKeyPath.length-j)
                { j++; }
            else
                { i++; j++; }
        }
        else
        if ("*".equals(routingTopicPath[i]) || "*".equals(routingKeyPath[j]))
        {
            i++; j++;
        }
        else
        if (routingTopicPath[i].equals(routingKeyPath[j]))
        {
            i++; j++;
        }
        else
        {
            return false;
        }
    }

    return true;
}