在Spring Boot中通过自定义ConcurrentKafkaListenerContainerFactory可灵活配置消费者并发数、手动提交offset、异常重试及反序列化策略，并需在@KafkaListener中显式指定containerFactory。
Docker Image Puller
下载
使用 Python 脚本下载 Docker 镜像并打包为 tar 文件。用户说“下载镜像”“拉取镜像”“pull xxx”“帮我下载 nginx 镜像”等时触发。支持 SOCKS5 代理直连（推荐）和国内镜像站加速两种模式。
<p>在 Spring Boot 项目中使用 <code>spring-kafka</code> 时，通过自定义 <code>ConcurrentKafkaListenerContainerFactory</code> 可以灵活控制消费者线程数、异常处理、反序列化方式、重试机制等核心行为。</p>
<H3>引入 spring-kafka 依赖</H3>
<p>确保 Maven 中已添加 <code>spring-kafka</code> 和自动配置支持（Spring Boot 项目通常只需以下依赖）：</p>
<p><font color="gray"><dependency><br> <groupId>org.springframework.kafka</groupId><br> <artifactId>spring-kafka</artifactId><br></dependency></font></p>
<p>Spring Boot 2.1+ 会自动拉取兼容版本；若手动指定版本，建议与 Spring Boot 版本对齐（如 Boot 3.x 对应 Kafka 3.x 客户端）。</p>
<H3>配置 Kafka 基础连接参数</H3>
<p>在 <code>application.yml</code> 或 <code>application.properties</code> 中设置集群地址和默认序列化器：</p>
<p><font color="gray">spring:<br> kafka:<br> bootstrap-servers: localhost:9092<br> consumer:<br> key-deserializer: org.apache.kafka.common.serialization.StringDeserializer<br> value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer<br> properties:<br> spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer<br> spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer<br> spring.json.trusted.packages: "com.example.*"</font></p>
<p>注意：使用 <code>ErrorHandlingDeserializer</code> 可避免反序列化失败直接导致容器停止；<code>trusted.packages</code> 控制 JSON 反序列化白名单。</p>
<H3>定义自定义 ConcurrentKafkaListenerContainerFactory</H3>
<p>通过 @Bean 配置工厂，覆盖默认行为。常见定制包括并发度、监听器适配器、错误处理器等：</p>
<p><font color="gray">@Configuration<br>public class KafkaConfig {<br><br> @Bean<br> public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(<br> ConsumerFactory<String, String> consumerFactory) {<br> ConcurrentKafkaListenerContainerFactory<String, String> factory =<br> new ConcurrentKafkaListenerContainerFactory<>();<br> factory.setConsumerFactory(consumerFactory);<br> factory.setConcurrency(3); // 每个 topic-partition 组分配 3 个线程（即最多 3 个并发消费者）<br> factory.getContainerProperties().setPollTimeout(3000);<br> factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);<br> factory.setErrorHandler(new SeekToCurrentErrorHandler(new FixedBackOff(1000L, 2)));<br> return factory;<br> }<br>}</font></p>
<p>关键点说明：</p>
<ul>
<li><strong>setConcurrency(n)</strong>：决定并发消费线程数，实际消费者实例数 ≤ 分区总数；若 topic 有 6 个分区，设为 3 则启动 3 个消费者实例，每个负责 2 个分区</li>
<li><strong>AckMode.MANUAL_IMMEDIATE</strong>：配合 <code>@KafkaListener</code> 方法参数中的 <code>Acknowledgment</code> 手动提交 offset，适合精确一次语义或复杂业务逻辑</li>
<li><strong>SeekToCurrentErrorHandler</strong>：发生异常时，对当前批次消息重新 seek 并重试（配合 FixedBackOff 实现重试间隔与次数）</li>
<li>若需全局异常兜底，可设置 <code>factory.setCommonErrorHandler(...)</code>（Spring Kafka 2.8+）</li>
</ul>
<H3>在 @KafkaListener 中使用自定义工厂</H3>
<p>通过 <code>containerFactory</code> 属性指定 Bean 名称（默认为 <code>kafkaListenerContainerFactory</code>，自定义后需显式声明）：</p>
<p><font color="gray">@KafkaListener(<br> topics = "my-topic",<br> containerFactory = "kafkaListenerContainerFactory" // 必须与 @Bean 方法名一致<br>)<br>public void listen(String data, Acknowledgment ack) {<br> try {<br> // 处理业务<br> ack.acknowledge(); // 手动提交<br> } catch (Exception e) {<br> // 异常由 SeekToCurrentErrorHandler 统一处理<br> }<br>}</font></p>
<p>注意：若多个 Listener 使用不同策略（如不同并发、不同重试逻辑），可定义多个 <code>@Bean</code> 工厂并分别引用。</p> 
