乐趣区

关于vue3:基于-Vue3-打造前台中台通用提效解决方案完整无密内置文档资料

jiezi

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 的概念。

退出移动版