初识TDMQ

一：需求背景

• 目前公司需要将决策引擎处理的结果， 一部分数据交给下游分析/入黑/通知等功能。因此就需要决策引擎生产结果让多方下游去消费。 而我需要实现下游的一部分功能。

二：相关文档

• TDMQ官方文档(TCP的SDK)： https://www.tencentcloud.com/zh/document/product/1110/42950

三：验证TDMQ广播消息

• 将TDMQ使用配置写入到一个配置文件中： config.py

  topic = ''
  tdmq_url = ''
  tdmq_secret_key = ''
  

• 编写生产者发送消息：

  #!/usr/bin/env python
  # -*- coding: utf-8 -*-
  # @Time   : 2023/7/24 16:28
  # 模拟TDMQ生产者发送消息
  import json
  
  import pulsar
  
  from conf.config import tdmq_secret_key, tdmq_url, topic
  
  
  def produser_send_msg():
      """生产者发送消息"""
      tdmq_client = pulsar.Client(
          authentication=pulsar.AuthenticationToken(tdmq_secret_key),
          service_url=tdmq_url
      )
      producer = tdmq_client.create_producer(
          topic=topic
      )
      send_data = json.dumps({"uniq_id": "12345670", "project_id": 2})
      producer.send(
          send_data.encode('utf-8'),
          properties={},
          partition_key=''
      )
  
  
  if __name__ == '__main__':
      produser_send_msg()
  

• 编写两个消费者， 消费消息(第二个只需要将subscription_name改为sub_topic2)

  #!/usr/bin/env python
  # -*- coding: utf-8 -*-
  # @Time   : 2023/7/24 18:55
  # @Author : shanwen.ren
  import pulsar
  
  from conf.config import tdmq_secret_key, tdmq_url, post_fund_topic
  
  tdmq_client = pulsar.Client(
          authentication=pulsar.AuthenticationToken(tdmq_secret_key),
          service_url=tdmq_url
      )
  
  consumer = tdmq_client.subscribe(
      # topic完整路径，格式为persistent://集群（租户）ID/命名空间/Topic名称，从【Topic管理】处复制
      topic=topic,
      # 订阅名称
      subscription_name='sub_topic1'
  )
  
  
  def produser_send_msg():
      """消费者消费消息"""
      # 获取消息
      msg = consumer.receive()
      try:
          # 模拟业务
          print("Received message '{}' id='{}'".format(msg.data(), msg.message_id()))
          # 消费成功，回复ack
          consumer.acknowledge(msg)
      except:
          # 消费失败，消息将会重新投递
          consumer.negative_acknowledge(msg)
  
  
  if __name__ == '__main__':
      while True:
          produser_send_msg()
  

• 开启两个消费者进程。
  

• 启动生产者发送消息