要保持 Kafka 套接字连接活跃的Spring Cloud Stream，可以使用Spring Kafka提供的@KafkaListener注解和KafkaTemplate类。以下是一个示例解决方案：

首先，添加所需的依赖项到pom.xml文件中：

    org.springframework.boot
    spring-boot-starter



    org.springframework.boot
    spring-boot-starter-web



    org.springframework.cloud
    spring-cloud-stream-binder-kafka

然后，在application.properties文件中配置Kafka连接属性：

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=my-group
spring.kafka.consumer.auto-offset-reset=earliest

接下来，创建一个消息监听器类，使用@KafkaListener注解指定要监听的主题和分区：

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class KafkaMessageListener {

    @KafkaListener(topics = "my-topic")
    public void listen(String message) {
        // 处理接收到的消息
        System.out.println("Received message: " + message);
    }
}

最后，创建一个生产者类，使用KafkaTemplate发送消息：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class KafkaProducer {

    private final KafkaTemplate kafkaTemplate;

    @Autowired
    public KafkaProducer(KafkaTemplate kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
    }
}

现在，当应用程序启动时，它将创建一个Kafka消费者，始终保持与Kafka服务器的连接，并且可以监听my-topic主题的消息。您可以使用KafkaProducer类发送消息到Kafka主题。

请注意，这只是一个简单示例，您可以根据您的需求进行更多的定制和配置。