काफ्का-पायथन का उपयोग करके एकाधिक कफका वाइल्डकार्ड पैटर्न की सूची में कैसे सदस्यता लें?

Question

मैं नीचे दिखाए गए अनुसार वाइल्डकार्ड वाले पैटर्न का उपयोग करके काफ्का की सदस्यता ले रहा हूं। वाइल्डकार्ड गतिशील ग्राहक आईडी का प्रतिनिधित्व करता है।

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

यह अच्छी तरह से काम करता है, क्योंकि मैं विषय आईडी से ग्राहक आईडी को निकाल सकता हूं। लेकिन अब मुझे थोड़ा अलग उद्देश्य के लिए एक समान विषय सुनने के लिए कार्यक्षमता पर विस्तार करने की आवश्यकता है। आइए इसे customer.*.additional-validations पर कॉल करें। कोड को उसी प्रोजेक्ट में रहने की आवश्यकता है क्योंकि बहुत अधिक कार्यक्षमता साझा की जाती है, लेकिन मुझे कतार के प्रकार के आधार पर एक अलग पथ लेने में सक्षम होना चाहिए।

Kafka documentation में मैं देख सकता हूं कि विषयों की एक श्रृंखला की सदस्यता लेना संभव है। हालांकि ये हार्ड-कोड वाले तार हैं। पैटर्न नहीं जो लचीलापन की अनुमति देते हैं।

>>> # 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']) 

Answer 1

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). 

तो तुम एक regex बना सकते हैं , या | का उपयोग कर स्थिति के साथ, इसे कई गतिशील विषयों regex की सदस्यता के रूप में काम करना चाहिए, क्योंकि यह मिलान के लिए आंतरिक रूप से re मॉड्यूल का उपयोग करता है।

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