22

我正在使用带有通配符的模式订阅 Kafka，如下所示。通配符代表动态客户 ID。

consumer.subscribe(pattern='customer.*.validations')

这很好用，因为我可以从主题字符串中提取客户 ID。但是现在我需要扩展功能以听一个类似的主题，目的略有不同。让我们称之为customer.*.additional-validations。代码需要存在于同一个项目中，因为共享了太多功能，但我需要能够根据队列的类型采用不同的路径。

在Kafka 文档中，我可以看到可以订阅一系列主题。然而，这些是硬编码的字符串。不是允许灵活性的模式。

>>> # Deserialize msgpack-encoded values
>>> consumer = KafkaConsumer(value_deserializer=msgpack.loads)
>>> consumer.subscribe(['msgpackfoo'])
>>> for msg in consumer:
...     assert isinstance(msg.value, dict)

所以我想知道是否有可能以某种方式将两者结合起来？有点像这样（非工作）：

consumer.subscribe(pattern=['customer.*.validations', 'customer.*.additional-validations'])
4

2 回答 2

30

在 KafkaConsumer 代码中，它支持主题列表或模式，

https://github.com/dpkp/kafka-python/blob/68c8fa4ad01f8fef38708f257cb1c261cfac01ab/kafka/consumer/group.py#L717

   def subscribe(self, topics=(), pattern=None, listener=None):
        """Subscribe to a list of topics, or a topic regex pattern
        Partitions will be dynamically assigned via a group coordinator.
        Topic subscriptions are not incremental: this list will replace the
        current assignment (if there is one).

因此，您可以使用 OR 条件创建一个正则表达式，|它应该可以订阅多个动态主题正则表达式，因为它在内部使用re模块进行匹配。

(customer.*.validations)|(customer.*.additional-validations)

于 2016-09-19T13:38:16.157 回答
1

在Confluent Kafka 库中，订阅没有pattern关键字，而是处理以^.

def subscribe(self, topics, on_assign=None, *args, **kwargs):
    """
    Set subscription to a supplied list of topics
    This replaces a previous subscription.
        
    Regexp pattern subscriptions are supported by prefixing the topic string with ``"^"``, e.g.::
        
        consumer.subscribe(["^my_topic.*", "^another[0-9]-?[a-z]+$", "not_a_regex"])
    """
于 2020-11-23T22:10:01.487 回答