在Python中发送带有键（key）的Kafka消息，通常会使用`confluent-kafka`或`kafka-python`这样的库。这里我将分别展示如何使用这两个库来实现这个功能。

 

### 使用 `confluent-kafka`

 

首先，确保你已经安装了`confluent-kafka`库。如果没有安装，可以使用pip进行安装：

```bash

pip install confluent-kafka

```

 

然后，你可以使用以下代码来发送带有键的消息：

```python

from confluent_kafka import Producer

 

def delivery_report(err, msg):

    """ Called once for each message produced to indicate delivery result.

        Triggered by poll() or flush(). """

    if err is not None:

        print(f'Message delivery failed: {err}')

    else:

        print(f'Message delivered to {msg.topic()} [{msg.partition()}]')

 

# 配置生产者

conf = {'bootstrap.servers': 'localhost:9092'}

 

# 创建生产者实例

producer = Producer(conf)

 

# 消息的键和值

message_key = 'my_key'

message_value = 'Hello, Kafka!'

 

# 发送消息

producer.produce('my_topic', key=message_key, value=message_value, callback=delivery_report)

 

# 触发所有消息的回调函数

producer.flush()

```

 

### 使用 `kafka-python`

 

同样地，确保你已经安装了`kafka-python`库。如果未安装，可以通过pip安装：

```bash

pip install kafka-python

```

 

接下来，使用以下代码来发送带有键的消息：

```python

from kafka import KafkaProducer

 

# 创建生产者实例

producer = KafkaProducer(bootstrap_servers='localhost:9092',

                         key_serializer=str.encode,

                         value_serializer=str.encode)

 

# 消息的键和值

message_key = 'my_key'

message_value = 'Hello, Kafka!'

 

# 发送消息

producer.send('my_topic', key=message_key, value=message_value)

 

# 确保所有消息都已发送

producer.flush()

 

# 关闭生产者

producer.close()

```

 

在这两个例子中，我们创建了一个Kafka生产者，并指定了一个本地运行的Kafka服务器地址（`localhost:9092`）。然后，我们定义了要发送的消息的键和值，并调用了相应的方法来发送消息。对于`confluent-kafka`，我们还设置了一个回调函数来处理消息的交付结果。

 

请根据你的实际环境调整配置，例如Kafka服务器的地址等。希望这能帮助到你！如果有任何其他问题，请随时提问。