我们的团队一直在尝试Kafka的问题。自从我们开始开发应用程序以来，这个问题就一直存在。
起初，这些问题很快就解决了。”它失败了？只需重新启动服务器”。既然我们想开始向公众发布我们的应用程序，这个“解决方案”就不再可行了。
我们面临的问题基本上有两个：

消费者停止工作

这是一个反复出现的问题。突然间，一些消费者就停了下来。消息被成功地发送到kafka，我们甚至可以使用kafka工具看到实际的消息，但是消费者就是不工作。

循环消息

这恰恰相反。有时会发送一条消息，而使用者会一直使用该消息，直到我们重新启动服务器。
我们尝试将kafka直接配置到服务器中，但是我们意识到，出于某种原因，kafka忽略了这些配置，直接从spring boot获取配置。
我们的配置如下：
消费者：

properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_BOOTSTRAP);
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, KAFKA_CONSUMER_GROUP);
    properties.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 5000000);
    properties.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 10800000);

制作人：

properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.KAFKA_HOST);
    properties.put(ProducerConfig.RETRIES_CONFIG, 0);
    properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
    properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);
    properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
    properties.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 5000000);
    properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "gzip");
    properties.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 10800000);

您可以看到极端的超时和大小值，这是因为我们认为问题与消息的大小或服务器的超时有关。我们甚至重新设计了所有的应用程序流量，这样我们就可以发送更小的消息，这时我们才意识到这不是根本问题。
任何帮助都将不胜感激。
Kafka版本0.10
spring boot版本1.5.7 dalston.sr4
我们使用的是springcloudstarter流kafka依赖项。
我们检查了日志，但没有可识别的错误。事实上，这里只有信息信息，但没有一条是有用的。