Kafka（三）生产者发送消息

文章目录

生产者发送思路

如何确保消息格式正确的前提下最终一定能发送到Kafka? 这里的实现思路是

1. ack使用默认的all
2. 开启重试
3. 在一定时间内重试不成功，则入库，后续由定时任务继续发送
4. 这里在某些异常情况下一定会生产重复消息，如何确保消息只消费一次，后续在Consumer实现中详细展开
5. 这里我们只要确保生产的消息，不论重试多少次，最终都只会被发送到同一分区。Kafka的确定消息的分区策略是： 如果提供了key，则根据hash(key)计算分区。由于我们每个消息都有一个消息ID，不管是重试多少次，ID是不会变的，同时我们不会在消息高峰阶段调整分区数量。所以基于这些，我们保证一个消息无论多少次，都会发送到同一分区。

自定义序列化类

消息格式为JSON, 使用Jackson将类序列化为JSON字符串

publicclassUserDTOSerializerimplementsSerializer<UserDTO>{@Override@SneakyThrowspublicbyte[]serialize(finalString s,finalUserDTO userDTO){ObjectMapper objectMapper =newObjectMapper();return objectMapper.writeValueAsBytes(userDTO);}}

配置生产者参数

有几点需要注意

1. 开启压缩
2. retries 官方建议不配置， 官方建议使用delivery.timeout.ms 参数来控制重试时间， 默认2分钟
3. buffer.memory 如果没有什么特别情况，使用默认的即可， 32MB
4. ack使用默认的all

/**
     * 以下配置建议搭配 官方文档 + kafka权威指南相关章节 + 实际业务场景吞吐量需求 自己调整
     * 如果是本地， bootstrap.server的IP地址和docker-compose.yml中的EXTERNAL保持一致
     * @return
     */publicstaticPropertiesloadProducerConfig(String valueSerializer){Properties result =newProperties();
        result.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"l192.168.0.102:9093");
        result.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        result.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer);
        result.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"gzip");// 每封邮件消息大小大约20KB, 使用默认配置吞吐量不高，下列配置增加kafka的吞吐量// 默认16384 bytes，太小了，这会导致邮件消息一个一个发送到kafka，达不到批量发送的目的，不符合发送邮件的场景
        result.put(ProducerConfig.BATCH_SIZE_CONFIG,1048576*10);// 默认1048576 bytes，限制的是一个batch的大小，对于20KB的消息来说，消息太小
        result.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG,1048576*10);// 等10ms, 为了让更多的消息聚合到一个batch中，提高吞吐量
        result.put(ProducerConfig.LINGER_MS_CONFIG,10);return result;}

提升吞吐量

• 在实际场景中，我们的邮件消息一个大概20KB，而batch.size默认是16KB，也就是说，在不修改该参数的情况下，生产者只能一个一个的发消息，这会导致我们的吞吐量上不去， 所以修改batch.size为10MB
• 只修改这个参数还不行， max.request.size 限制了单次请求的大小，默认为1MB，也就是说即使batch.size为10MB，但是由于一次只能最多发1MB，吞吐量也上不去，所以这里将max.request.size也改为10MB
• 由于我们将一个批次可发送的数量大大提高，所以可以让生产者等一会再发，等更多的数据到达。linger.ms默认是为0，也就是立刻发送，根据实际情况适当增加等待时间

发送消息

@LogpublicclassMessageProducer{publicstaticfinalKafkaProducer<String,UserDTO>PRODUCER=newKafkaProducer<>(KafkaConfiguration.loadProducerConfig(UserDTOSerializer.class.getName()));privateMessageFailedService messageFailedService =newMessageFailedService();/**
     * kafka producer 发送失败时会进行重试，相关参数 retries 和 delivery.timeout.ms, 官方建议使用delivery.timeout.ms，默认2分钟
     * callback函数只有在最后一次重试之后才会调用， 如果你想在本地测试Kafka生产者的重试，详情可以看https://lists.apache.org/thread/nwg326bxpo7ry116nqhxmsmc3bokc6hm
     * @param userDTO
     */publicvoidsendMessage(finalUserDTO userDTO){ProducerRecord<String,UserDTO> user =newProducerRecord<>("email", userDTO.getMessageId(),  userDTO);try{PRODUCER.send( user,(recordMetadata, e)->{if(Objects.nonNull(e)){
                    log.severe("message has sent failed");MessageFailedEntity messageFailedEntity =newMessageFailedEntity();
                    messageFailedEntity.setMessageId(userDTO.getMessageId());ObjectMapper mapper =newObjectMapper();try{
                        messageFailedEntity.setMessageContentJsonFormat(mapper.writeValueAsString(userDTO));}catch(JsonProcessingException jsonProcessingException){
                        log.severe("message content json format failed");}
                    messageFailedEntity.setMessageType(MessageType.EMAIL);
                    messageFailedEntity.setMessageFailedPhrase(MessageFailedPhrase.PRODUCER);
                    messageFailedEntity.setFailedReason(e.getMessage());// 如果sendMessage传进来的是个list，也同理，不能放到list.foreach外面// 如果放在主线程里，由于kafka producer是异步的，// kafka producer的执行速度可能慢于主线程，可能拿到的值是空的是有问题的，例如拿到的failedReason是空的
                    messageFailedService.saveOrUpdateMessageFailed(messageFailedEntity);}else{
                    log.info("message has sent to topic: "+ recordMetadata.topic()+", partition: "+ recordMetadata.partition());}});}catch(TimeoutException e){
            log.info("send message to kafka timeout, message: ");// TODO: 自定义逻辑，比如发邮件通知kafka管理员}}}

对上述代码做几点解释

1. 我们使用异步的方式发送，如果发送成功，打印一条消息
2. 关键在于重试，callback函数只有在最后一次重试之后才会调用。不会重试多少次就调用多少次callback， 这个问题我发邮件问过社区， 详情见这里的 邮件，想要本地测试或debug重试相关逻辑的话，可以将min.insync.replicas改大一点，例如我只有一个Kafka节点，那么我设置min.insync.replicas为大于1的值，并且设置Kafka的Producer的确认设置是all，那么这时候发送消息，就会看到重试相关日志

关闭生产者

实现ServletContextListener接口, 然后在web.xml的listener元素中配置

publicclassKafkaListenerimplementsServletContextListener{privatestaticfinalList<KafkaProducer>KAFKA_PRODUCERS=newLinkedList<>();@OverridepublicvoidcontextInitialized(ServletContextEvent sce){KAFKA_PRODUCERS.add(MessageProducer.PRODUCER);}@OverridepublicvoidcontextDestroyed(ServletContextEvent sce){KAFKA_PRODUCERS.forEach(KafkaProducer::close);}}

<?xml version="1.0" encoding="UTF-8" ?><web-appxmlns="https://jakarta.ee/xml/ns/jakartaee"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="https://jakarta.ee/xml/ns/jakartaee
                      https://jakarta.ee/xml/ns/jakartaee/web-app_6_0.xsd"version="6.0"><listener><listener-class>com.business.server.listener.KafkaListener</listener-class></listener></web-app>

结语

1. 在实际编码过程中，可以参考官方写的Kafka权威指南对应章节书写，或者参考各大云服务厂商的Kafak的开发者文档。不过我建议还是看Kafka权威指南， 我看了阿里云和华为云的，虽然都号称兼容开源Kafka，但是发现其版本和开源版本之间存在一定的滞后性，许多最佳实践已经过时
2. Kafka生产者端没什么特别的，主要是根据业务场景设计消息格式，以及如何尽可能的减小消息体积
3. 如果你的消息很大，比我的场景还大，达到了1M以上，生产者的吞吐量是个问题，消费者的消费速度也是个问题。你要是问我有什么好的想法，没有具体场景，我确实想不出什么好的方式

示例源码仓库

1. Github地址
2. 项目下business-server module代表生产者
3. 运行时IDEA配置如下 注意Application context的路径, 启动之后访问端口+Application context, 例如

http://localhost:8999/business-server

下一篇博文，将介绍消费者消费消息，以及消费者的重要参数配置，还有消费逻辑的重试机制等。