【世界时快讯】python 使用 kafka

• 2023-03-25 21:21:57 来源：博客园

python 使用 kafka

说明：关于 kafka 的启动与安装，命令行的使用，此处不做过多的解释，本篇文章主要描述 kafka 在 python 中的使用；

(相关资料图)

1. python 使用 kafka 生产者

**说明：**python 在操作 kafka 写入数据的时候，分为发送往已经存在的主题或者是不存在的主题，当主题不存在的时候，生产者会自动创建该主题，并将消息存贮在默认的 0 分区；

使用 python 操作 kafka 首先安装如下的包

pip install kafka pip install kafka-python # 由于 python 3.7 后的版本中 async 的关键字发生了变化,因此需要多安装该包;

常规的使用主要就是根据，第三方包的介绍使用，网上有许多基本的案例，此处不做介绍，下面直接将封装好的常用的方法进行封装；

import json​import kafka​​class Producer(object):  """ kafka 的生产者模型   """​  _coding = "utf-8"​  def __init__(self,         broker="192.168.74.136:9092",         topic="add_topic",         max_request_size=104857600,         batch_size=0, # 即时发送,提高并发可以适当增加,但是会造成消息的延迟;         **kwargs):    """初始化设置 kafka 生产者连接对象;参数不存在的情况下使用配置文件中的默认连接;     """    self.broker = broker    self.topic = topic    self.max_request_size = max_request_size    # 实例化生产者对象    self.producer_json = kafka.KafkaProducer(      bootstrap_servers=self.broker,      max_request_size=self.max_request_size,      batch_size=batch_size,      key_serializer=lambda k: json.dumps(k).encode(self._coding), # 设置键的形式使用匿名函数进行转换      value_serializer=lambda v: json.dumps(v).encode(self._coding), # 当需要使用 json 传输地时候必须加上这两个参数      **kwargs     )​    self.producer = kafka.KafkaProducer(      bootstrap_servers=broker,      max_request_size=self.max_request_size,      batch_size=batch_size,      api_version=(0, 10, 1),      **kwargs     )​  def send(self, message: bytes, partition: int = 0):    """     写入普通的消息;     Args:       message: bytes; 字节流数据;将字符串编码成 utf-8的格式;       partition: int; kafka 的分区,将消息发送到指定的分区之中;     Returns:       None     """    future = self.producer.send(self.topic, message, partition=partition)    record_metadata = future.get(timeout=30)    if future.failed(): # 发送失败,记录异常到日志;      raise Exception("send message failed:%s)" % future.exception)​  def send_json(self, key: str, value: dict, partition: int = 0):    """     发送 json 形式的数据;     Args:       key: str; kafka 中键的值       value: dict; 发送的具体消息       partition: int; 分区的信息     Returns:       None     """    future = self.producer_json.send(self.topic, key=key, value=value, partition=partition)    record_metadata = future.get(timeout=30)    if future.failed(): # 发送失败记录异常;      raise Exception("send json message failed:%s)" % future.exception)​  def close(self):    """     关闭kafka的连接。     Returns:       None     """    self.producer_json.close()    self.producer.close()​​if __name__ == "__main__":  """脚本调用执行;"""  kafka_obj = Producer()  print(kafka_obj.broker)  kafka_obj.send("自动生成".encode())​

发送的消息，主要是普通的字符串消息，和字典形式的消息，方便对接；

2. python 使用 kafka 消费者

由于 kafka 消费者的特性，阻塞循环是一个必然的过程，可以使用 python 中的生成器进行优化，但是循环阻塞是无可避免的；

操作 kafka 的消费者依旧只需要安装上述的两个第三方依赖包；

封装指定的操作

import json​from kafka import KafkaConsumer, KafkaProducerfrom kafka.structs import TopicPartition​​class KConsumer(object):  """kafka 消费者; 动态传参,非配置文件传入;    kafka 的消费者应该尽量和生产者保持在不同的节点上;否则容易将程序陷入死循环中;   """​  _encode = "UTF-8"​  def __init__(self, topics="start_server", bootstrap_server=None, group_id="start_task", partitions=None, **kwargs):    """ 初始化kafka的消费者;       1. 设置默认 kafka 的主题, 节点地址, 消费者组 id(不传入的时候使用默认的值)       2. 当需要设置特定参数的时候可以直接在 kwargs 直接传入,进行解包传入原始函数;       3. 手动设置偏移量     Args:       topics: str; kafka 的消费主题;       bootstrap_server: list; kafka 的消费者地址;       group_id: str; kafka 的消费者分组 id,默认是 start_task 主要是接收并启动任务的消费者,仅此一个消费者组id;       partitions: int; 消费的分区,当不使用分区的时候默认读取是所有分区;       **kwargs: dict; 其他原生kafka消费者参数的;     """​    if bootstrap_server is None:      bootstrap_server = ["192.168.74.136:9092", ]    self.consumer = KafkaConsumer(bootstrap_servers=bootstrap_server)    exist = self.exist_topics(topics)    if not exist: # 需要的主题不存在;      # 创建一条      self.create_topics(topics)    if partitions is not None:      self.consumer = KafkaConsumer(        bootstrap_servers=bootstrap_server,        group_id=group_id,         # 目前只有一个消费者,根据情况是否需要进行修改;当扩展多个消费者的时候需要进行扩展;        **kwargs       )      # print("指定分区信息:", partitions, topics, type(partitions))      self.topic_set = TopicPartition(topics, int(partitions))      self.consumer.assign([self.topic_set])    else:      # 默认读取主题下的所有分区, 但是该操作不支持自定义 offset, 因为 offset 一定是在指定的分区中进行的;      self.consumer = KafkaConsumer(        topics,        bootstrap_servers=bootstrap_server,        group_id=group_id,        **kwargs       )​  def exist_topics(self, topics):    """     检查 kafka 中的主题是否存在;     Args:       topics: 主题名称;​     Returns:       bool: True/False ; True,表示存在,False 表示不存在;     """    topics_set = set(self.consumer.topics())    if topics not in topics_set:      return False    return True​  @staticmethod  def create_topics(topics):    """     创建相关的 kafka 主题信息;说明本方法可以实现用户自定义 kafka 的启动服务,默认是使用的是 start_server;     Args:       topics: str; 主题的名字;​     Returns:       None     """    producer = KafkaProducer(      bootstrap_servers="192.168.74.136:9092",      key_serializer=lambda k: json.dumps(k).encode("utf-8"),      value_serializer=lambda v: json.dumps(v).encode("utf-8")     )    producer.send(topics, key="start", value={"msg": "aaaa"})    producer.close()​  def recv(self):    """     接收消费中的数据     Returns:       使用生成器进行返回;     """    for message in self.consumer:       # 这是一个永久阻塞的过程,生产者消息会缓存在消息队列中,并且不删除,所以每个消息在消息队列中都会有偏移      # print("主题:%s 分区:%d:连续值:%d: 键:key=%s 值:value=%s" % (      #   message.topic, message.partition, message.offset, message.key, message.value))      yield {"topic": message.topic, "partition": message.partition, "key": message.key,          "value": message.value.decode(self._encode)}​  def recv_seek(self, offset):    """     接收消费者中的数据,按照 offset 的指定消费位置;     Args:       offset: int; kafka 消费者中指定的消费位置;​     Returns:       generator; 消费者消息的生成器;     """    self.consumer.seek(self.topic_set, offset)    for message in self.consumer:      # print("主题:%s 分区:%d:连续值:%d: 键:key=%s 值:value=%s" % (      #   message.topic, message.partition, message.offset, message.key, message.value))      yield {"topic": message.topic, "partition": message.partition, "key": message.key,          "value": message.value.decode(self._encode)}​​if __name__ == "__main__":  """ 测试使用;   """​  obj = KConsumer("exist_topic", bootstrap_server=["192.168.74.136:9092"])  for i in obj.recv():    print(i)

该消费者多封装时增加了一个需求，消费的主题不存在的时候会默认创建，下次就可以继续消费

3. 使用 docker 中的 kafka

以上两种脚本适用于 Kafka 的生产者和消费者在大多数情况下的使用，在使用的时候只需要将相关的配置信息修改即可；

docker 中使用 kafka 的时候与前面的配置稍有不同，当使用docker-compose部署 Kafka 的时候，地址在文件中经过修改，可能会被改变，但是配置方式，因此只需要将相关的地址配好，即可；代码信息无需修改；

一般情况下如果是在 docker 中配置相关的参数，需要将端口映射出来，然后如果是 windows 可能需要将host的网络地址解析，与docker 中 kafka 的名称对应；

host 文件​127.0.0.1 kafka

当需要远程连接的时候，将地址改成该计算机在内网中的地址即可；

标签：