青岛创世网络网站建设西乡做网站

写在前面

        如果只有一个kafka实例的话，那么文章中提到kafka集群=kafka实例

一、什么是消息发送者端的ack机制

        ack机制：消息确认发送成功的标识

        由谁发起该标识：kafka集群

        发起该标识的场景：kafka集群确认已经收到了消息。

        由谁接收该标识：消息发送者端

        接收该标识的意义：消息发送者接到该标识，才表示消息发送成功。

二、消息发送者端ack机制的策略有哪些

2.1、ack=0

        kafka集群不需要任何broker收到消息，就立即返回ack给生产者。

        该模式最容易丢失消息，但效率最高

2.2、akc=1

        kafka默认的ack机制策略

        kafka集群中，Leader的broker收到了消息，并且把消息写入到了log文件中，此时才返回ack给kafka。

        该模式性能和安全性是最均衡的。

2.3、ack=-1/all

        配置文件中的min.insyc.replicas=2(默认为1,推荐配置2)将生效。

        min.insyc.replicas=n：当kafka集群接受到消息以后，当kafka集群完成了n个broker(包含Leader和follower)上的备份以后，再回复ack。

        该模式最安全，但性能最差

三、消息发送失败的重试策略

        消息发送失败会重试，重试的默认时间是100ms,重试的默认次数是3。

        重试保证了消息的可靠性，但也可能造成消息的重复发送，比如说网络抖动。

        所以需要在消费者做好消息接口的幂等性操作

四、通过Java设置kafka的ack机制和重试策略

Properties props = new Properties();// 一、设置参数// 配置kafka地址
//        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
//                "192.168.151.28:9092"); // 单机配置props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.154.128:9092,192.168.154.128:9093,192.168.154.128:9094"); // 集群配置// 配置消息 键值的序列化规则props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());// 配置生产者发送消息的ack机制/*** ack=0：*   kafka集群不需要任何broker收到消息，就立即返回ack给生产者。*   该模式最容易丢失消息，但效率最高* akc=1：(kafka默认的方式)*   kafka集群中，Leader的broker收到了消息，并且把消息写入到了log文件中，此时才返回ack给kafka。*   该模式性能和安全性是最均衡的。* ack=-1/all：*   默认配置文件中的min.insyc.replicas=2(默认为1,推荐配置2)将生效。*   min.insyc.replicas=n解释：当kafka集群接受到消息以后，当kafka集群完成了n个broker(包含Leader和follower)上的备份以后，再回复ack。*   该模式最安全，但性能最差*/props.put(ProducerConfig.ACKS_CONFIG,"1");// 配置消息发送时间(未接收到ack)的重试次数和重试间隔时间/***  消息发送失败(即没有收到ack)的重试次数和重试间隔*   消息发送失败会重试，重试的默认时间是100ms,重试的默认次数是3.*   重试保证了消息的可靠性，但也可能造成消息的重复发送，比如说网络抖动。所以需要在消费者做好消息接口的幂等性操作*/props.put(ProducerConfig.RETRIES_CONFIG,"3");props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG,"300"); // 重试间隔