我正在做一个项目，使用Kafka的生产者和消费者，以便每两个小时从news_API获取文章（有特定主题），然后与消费者一起将它们保存在mongodb中。
所以我创建了三个类一个用于KafkaAdminClient一个用于KafkaProducer一个用于KafkaConsumer。
我的Kafka的服务器运行在一个Docker容器上。主要的应用程序是一个 flask 应用程序，我在那里启动所有的线程和Kafka的线程。
我一直试图改变很多小东西，但它似乎很不稳定，我不知道为什么。首先，数据在消费者上，并最终在随机时间进入mongodb。然后，消费者中的旧主题不会被删除，数据库不断填充新旧值。
现在我在消费者中放置了一个组，并添加了kafkaAdminClient类，我在消费者中根本没有得到消息。

articleretrieval-flask_api-1          | WARNING:kafka.cluster:Topic health is not available during auto-create initialization articleretrieval-flask_api-1          | WARNING:kafka.cluster:Topic business is not available during auto-create initialization articleretrieval-flask_api-1          | WARNING:kafka.cluster:Topic war is not available during auto-create initialization articleretrieval-flask_api-1         
 | WARNING:kafka.cluster:Topic motorsport is not available during auto-create initialization articleretrieval-flask_api-1          
| WARNING:kafka.cluster:Topic sources is not available during auto-create initialization articleretrieval-flask_api-1         
 | WARNING:kafka.cluster:Topic science is not available during auto-create initialization articleretrieval-flask_api-1         
 | WARNING:kafka.cluster:Topic technology is not available during auto-create initialization articleretrieval-flask_api-1         
 | WARNING:kafka.cluster:Topic education is not available during auto-create initialization articleretrieval-flask_api-1          
| WARNING:kafka.cluster:Topic space is not available during auto-create initialization articleretrieval-flask_api-1          
| INFO:kafka.consumer.subscription_state:Updated partition assignment: [] articleretrieval-flask_api-1    
| INFO:kafka.conn:<BrokerConnection node_id=bootstrap-0 host=kafka:29092 <connected> [IPv4 ('172.19.0.4', 29092)]>: Closing connection.

kafkaConsumerThread.py:
x一个一个一个一个x一个一个二个x
kafkaAdminClient.py:

class KafkaAdminThread:
    def __init__(self,topics):
        self.topics = topics

    def start(self):
        admin_client = KafkaAdminClient(
            bootstrap_servers=['kafka:29092'], 
                client_id='my_client'
        )
        topic_list = []
        for topic in self.topics:
            topic_list.append(NewTopic(name=topic, num_partitions=1, replication_factor=1))
        admin_client.create_topics(new_topics=topic_list, validate_only=False)

app.py:

if __name__ == "__main__":
    # Creating a new connection with mongo
    # threading.Thread(target=lambda: app.run(port=8080, host="0.0.0.0",debug=True,use_reloader=False)).start()
    executor = ThreadPoolExecutor(max_workers=4)
    producerThread = KafkaProducerThread(TOPICS,logging)
    adminThread = KafkaAdminThread(TOPICS)
    executor.submit(adminThread.start)
    flaskThread = threading.Thread(target=lambda: app.run(port=8080, host="0.0.0.0", debug=True, use_reloader=False))
    executor.submit(flaskThread.start())
    time.sleep(15)
    executor.submit(producerThread.start)
    consumerThread = KafkaConsumerThread(TOPICS, db,logging)
    executor.submit(consumerThread.start)

docker-compose.yml:

zookeeper:
    image: wurstmeister/zookeeper
    ports:
      - "2181:2181"

  kafka:
    container_name: kafka_broker_1
    image: wurstmeister/kafka
    links:
      - zookeeper
    ports:
      - "9092:9092"
      - "29092:29092"
    depends_on:
      - zookeeper
    environment:
      KAFKA_ADVERTISED_HOSTNAME: kafka
      KAFKA_ADVERTISED_LISTENERS: INSIDE://kafka:29092,OUTSIDE://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT
      KAFKA_LISTENERS: INSIDE://0.0.0.0:29092,OUTSIDE://0.0.0.0:9092
      KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock%

  flask_api:
    build:
      context: . #Very important it refers where the root will be for the build.
      dockerfile: Dockerfile
    links:
        - kafka
    environment:
      - FLASK-KAFKA_BOOTSTRAP-SERVERS=kafka:29092
      - SERVER_PORT=8080
    ports:
      - "8080:8080"
    depends_on:
      - kafka