download：基于 Vue3 打造前台+中台通用提效解决方案残缺无密内置文档资料

分布式消息通信之Kafka的实现原理
消息中间件次要解决的就是分布式零碎之间消息传送的问题，它能够屏蔽各种平台以及协定之间的个性，实现应用程序之间的协同。举个非常简略的例子，就拿一个电商平台的注册功能来简略分析下，用户注册这一个服务，不单单只是insert一条数据到数据库外面就完事了，还需要发送激活邮件、发送新人红包或者积分、发送营销短信等一系列操作。假如说这外面的每一个操作，都需要消耗1s，那么整个注册过程就需要耗时4s才能响应给用户。

Java中使用kafka进行通信
依赖
<dependency>

 <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId>  <version>2.0.0</version>

</dependency>
复制代码
发送端代码
public class Producer extends Thread {

    private final KafkaProducer<Integer, String> producer;    private final String topic;    public Producer(String topic) {        Properties properties = new Properties();        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.13.102:9092,192 .168.13.103:9092,192.168.13.104:9092");        properties.put(ProducerConfig.CLIENT_ID_CONFIG, "practice-producer");        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());        生产端代码        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());        producer = new KafkaProducer<Integer, String>(properties);        this.topic = topic;    }    @Override    public void run() {        int num = 0;        while (num < 50) {            String msg = "pratice test message:" + num;            try {                producer.send(new ProducerRecord<Integer, String>(topic, msg)).get();                TimeUnit.SECONDS.sleep(2);                num++;            } catch (InterruptedException e) {                e.printStackTrace();            } catch (ExecutionException e) {                e.printStackTrace();            }        }    }    public static void main(String[] args) {        new Producer("test").start();    }}

复制代码
生产端代码
public class Consumer extends Thread {

    private final KafkaConsumer<Integer, String> consumer;    private final String topic;    public Consumer(String topic) {        Properties properties = new Properties();        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.13.102:9092,192 .168.13.103:9092,192.168.13.104:9092");        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "practice-consumer");        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");//设置 offset主动提交        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");// 主动提交间隔工夫         properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerDeserializer");        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");//对于 以后groupid来说，消息的offset从最早的消息开始生产         consumer = new KafkaConsumer<>(properties);        this.topic = topic;    }    @Override    public void run() {        while (true) {            consumer.subscribe(Collections.singleton(this.topic));            ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofSeconds(1));            records.forEach(record -> {                System.out.println(record.key() + " " + record.value() + " -> offset:" + record.offset());            });        }    }    public static void main(String[] args) {        new Consumer("test").start();    }}

复制代码
异步发送
kafka对于消息的发送，可能反对同步和异步，后面演示的案例中，咱们是基于同步发送消息。同步会需要阻塞，而异步不需要等待阻塞的过程。
从本质上来说，kafka都是采纳异步的形式来发送消息到broker，然而kafka并不是每次发送消息都会间接发送到broker上，而是把消息放到了一个发送队列中，而后通过一个后盾线程不断从队列取出消息进行发送，发送胜利后会触发callback。kafka客户端会积累一定量的消息对立组装成一个批量消息发送进来，触发条件是后面提到batch.size和linger.ms
而同步发送的方法，无非就是通过future.get()来等待消息的发送返回后果，然而这种方法会重大影响消
息发送的性能。
public void run() {

    int num = 0;    while (num < 50) {        String msg = "pratice test message:" + num;        try {            producer.send(new ProducerRecord<>(topic, msg), new Callback() {                @Override                public void onCompletion(RecordMetadata recordMetadata, Exception e) {                    System.out.println("callback: " + recordMetadata.offset() + "->" + recordMetadata.partition());                }            });            TimeUnit.SECONDS.sleep(2);            num++;        } catch (InterruptedException e) {            e.printStackTrace();        }    }}

复制代码
batch.size
消费者发送多个消息到broker上的同一个分区时，为了缩小网络请求带来的性能开销，通过批量的形式来提交消息，可能通过这个参数来管制批量提交的字节数大小，默认大小是16384byte,也就是16kb，意味着当一批消息大小达到指定的batch.size的时候会对立发送。
linger.ms
Producer默认会把两次发送工夫间隔内收集到的所有Requests进行一次聚合而后再发送，以此提高吞吐量，而linger.ms就是为每次发送到broker的请求减少一些delay，以此来聚合更多的Message请求。这个有点想TCP外面的Nagle算法，在TCP协定的传输中，为了缩小大量小数据包的发送，采纳了Nagle算法，也就是基于小包的等-停协定。

batch.size和linger.ms这两个参数是kafka性能优化的要害参数，很多同学会发现batch.size和
linger.ms这两者的作用是一样的，如果两个都配置了，那么怎么工作的呢？实际上，当二者都配
置的时候，只需满足其中一个申请，就会发送请求到broker上

一些基础配置分析
group.id
consumer group是kafka提供的可扩大且具备容错性的消费者机制。既然是一个组，那么组内必然可能有多个消费者或消费者实例(consumer instance)，它们共享一个公共的ID，即group ID。组内的所有消费者调和在一起来生产订阅主题(subscribed topics)的所有分区(partition)。当然，每个分区只能由同一个生产组内的一个consumer来生产.如下图所示，别离有三个消费者，属于两个不同的group，那么对于firstTopic这个topic来说，这两个组的消费者都能同时生产这个topic中的消息，对于此事的架构来说，这个firstTopic就类似于ActiveMQ中的topic概念。如右图所示，如果3个消费者都属于同一个group，那么此事firstTopic就是一个Queue的概念。