如何使用spring boot app中的“li apache kafka客户端”从kafka producer发送大消息（1MB以上）？

Question

如何在spring boot app中使用li apache kafka客户端从kafka制作人向kafka消费者发送大消息（1MB以上）？以下是li apache kafka客户端的GitHub链接：https://github.com/linkedin/li-apache-kafka-clients

我已经导入了。li apache kafka客户端的jar文件，并为producer设置以下配置：

props.put("large.message.enabled", "true");
props.put("max.message.segment.bytes", 1000 * 1024);
props.put("segment.serializer", DefaultSegmentSerializer.class.getName());

对于消费者：

message.assembler.buffer.capacity, 
max.tracked.messages.per.partition, 
exception.on.message.dropped, 
segment.deserializer.class

但对于大型消息，仍然会出现错误。请帮我解决错误。

下面是我的代码，请告诉我需要在哪里创建LiKafkaProducer：

@Configuration
public class KafkaProducerConfig {

    @Value("${kafka.boot.server}")
    private String kafkaServer;


    @Bean
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfig());
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<String, String>(producerFactory());
    }

    @Bean
    public Map<String, Object> producerConfig() {
        // TODO Auto-generated method stub
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put("bootstrap.servers", "localhost:9092");
        config.put("acks", "all");
        config.put("retries", 0);
        config.put("batch.size", 16384);
        config.put("linger.ms", 1);
        config.put("buffer.memory", 33554432);

        // The following properties are used by LiKafkaProducerImpl
        config.put("large.message.enabled", "true");
        config.put("max.message.segment.bytes", 1000 * 1024);
        config.put("segment.serializer", DefaultSegmentSerializer.class.getName());
        config.put("auditor.class", LoggingAuditor.class.getName());

        return config;
    }
}

@RestController
@RequestMapping("/kafkaProducer")
public class KafkaProducerController {

    @Autowired
    private KafkaSender sender;

    @PostMapping
    public ResponseEntity<List<Student>> sendData(@RequestBody List<Student> student){

        sender.sendData(student);
        return new ResponseEntity<List<Student>>(student, HttpStatus.OK);
    }
}

@Service
public class KafkaSender {

    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaSender.class);

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Value("${kafka.topic.name}")
    private String topicName;

    public void sendData(List<Student> student) {

        // TODO Auto-generated method stub
        Map<String, Object> headers = new HashMap<>();
        headers.put(KafkaHeaders.TOPIC, topicName);
        headers.put("payload", student.get(0));

        // Construct a JSONObject from a Map.
        JSONObject HeaderObject = new JSONObject(headers);
        System.out.println("\nMethod-2: Using new JSONObject() ==> " + HeaderObject);
        final String record = HeaderObject.toString();

        Message<String> message = MessageBuilder.withPayload(record).setHeader(KafkaHeaders.TOPIC, topicName)
                .setHeader(KafkaHeaders.MESSAGE_KEY, "Message")
                .build();

        kafkaTemplate.send(topicName, message.toString());
    }
}

Answer 1

您需要实现自己的消费工厂（ConsumerFactory）和生产工厂（ProducerFactory），以分别创建LiKafkaConsumer和LiKafkaProducer。

您应该能够子类化框架提供的默认工厂。