Spring批处理Kafka消费者对于多线程访问不安全

Question

我是一个新手。我有需要读取Kafka流和过滤数据的要求

这是Spring批处理配置。

    @Autowired
    TaskExecutor taskExecutor;

    @Autowired
    JobRepository jobRepository;

    @Bean
    KafkaItemReader<Long, Event> kafkaItemReader() {
        Properties props = new Properties();
        props.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false);
        props.putAll(this.properties.buildConsumerProperties());
        return new KafkaItemReaderBuilder<Long, Event>()
                .partitions(0)
                .consumerProperties(props)
                .name("event-reader")
                .saveState(true)
                .topic(topicName)
                .build();
    }

    @Bean
    public TaskExecutor taskExecutor(){
        SimpleAsyncTaskExecutor asyncTaskExecutor=new SimpleAsyncTaskExecutor("spring_batch");
        asyncTaskExecutor.setConcurrencyLimit(5);
        return asyncTaskExecutor;
    }

    @Bean(name = "JobLauncher")
    public JobLauncher simpleJobLauncher() throws Exception {
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        jobLauncher.setJobRepository(jobRepository);
        jobLauncher.setTaskExecutor(taskExecutor);
        jobLauncher.afterPropertiesSet();
        return jobLauncher;
    }

并且存在开始新作业的控制器endpoint。这是我必须使用的方式：开始新工作

    @Autowired
    @Qualifier("JobLauncher")
    private JobLauncher jobLauncher;


    Map<String, JobParameter> items = new HashMap<>();
    items.put("userId", new JobParameter("UserInputId"));
    JobParameters paramaters = new JobParameters(items);
    try {
        jobLauncher.run(job, paramaters);
    } catch (Exception e) {
        e.printStackTrace();
    }

我已经看到KafkaItemReader不是线程安全的。我想知道这种方法是正确的，还是有任何方法可以在多线程spring批处理环境中读取kafka流。谢谢

Answer 1

根据spring文档，它使用KafkaConsumer；根据其详细文档，其本身不是线程安全的。

请查看您是否可以使用该文档中提到的任何方法（即解耦或每个线程使用一个消费者）。在您的示例中，您可能需要为taskexecutor使用单独的处理程序（如果您遵循解耦方法）。

Answer 2

< code>KafkaItemReader被证明是非线程安全的，以下是其Javadoc的摘录:

Since KafkaConsumer is not thread-safe, this reader is not thread-safe.

所以在多线程环境中使用它是不正确的，并且不符合留档。您可以做的是每个分区使用一个读取器。