请给我一段使用kafka库，按自定义消费策略进行分配的，有10个生产者和10个消费者的python3代码

好的，以下是一个使用 Kafka-Python 库，按照自定义消费策略进行分配的，包含10个生产者和10个消费者的 Python 代码示例：

from kafka import KafkaProducer, KafkaConsumer, TopicPartition
from kafka import KafkaConsumer, TopicPartition
from kafka.errors import KafkaError
import time
import threading


# 生产者
class ProducerThread(threading.Thread):
    def __init__(self, topic, partition):
        threading.Thread.__init__(self)
        self.topic = topic
        self.partition = partition

    def run(self):
        producer = KafkaProducer(bootstrap_servers=['localhost:9092'], api_version=(0, 10))
        for i in range(10):
            msg = 'message from producer {} on partition {}'.format(self.name, self.partition)
            producer.send(self.topic, value=msg.encode('utf-8'), partition=self.partition)
        producer.close()


# 消费者
class ConsumerThread(threading.Thread):
    def __init__(self, topic, partition):
        threading.Thread.__init__(self)
        self.topic = topic
        self.partition = partition

    def run(self):
        consumer = KafkaConsumer(bootstrap_servers=['localhost:9092'], api_version=(0, 10))
        tp = TopicPartition(self.topic, self.partition)
        consumer.assign([tp])
        while True:
            msg_pack = consumer.poll(1000)
            for tp, messages in msg_pack.items():
                for message in messages:
                    print('consumer {} on partition {} got message: {}'.format(self.name, self.partition, message.value))
        consumer.close()


if __name__ == '__main__':
    topic = 'test-topic'
    producers = []
    consumers = []
    for i in range(10):
        producer_thread = ProducerThread(topic, i)
        producer_thread.start()
        producers.append(producer_thread)

        consumer_thread = ConsumerThread(topic, i)
        consumer_thread.start()
        consumers.append(consumer_thread)

    for producer_thread in producers:
        producer_thread.join()

    for consumer_thread in consumers:
        consumer_thread.join()

这段代码中，我们创建了一个 `ProducerThread` 类和一个 `ConsumerThread` 类，分别用于创建生产者和消费者线程。在主程序中，我们创建了10个生产者线程和10个消费者线程，并启动它们。生产者线程向 Kafka 的 `test-topic` 主题的10个分区中依次发送10条消息，消费者线程按照自定义的消费策略，每个线程消费一个分区中的消息，并将消息打印出来。

需要注意的是，这段代码中的消费者线程是永久运行的，需要手动停止。在实际应用中，可以使用信号量或者定时器来控制消费者线程的启停。