玖叶教程网

前端编程开发入门

2024年08月09日

消息队列kafka - 实战

本文介绍如何在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");//设置
    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");//对于
    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的概念。

enable.auto.commit

消费者消费消息以后自动提交，只有当消息提交以后，该消息才不会被再次接收到，还可以配合auto.commit.interval.ms控制自动提交的频率。

当然，我们也可以通过consumer.commitSync()的方式实现手动提交。

auto.offset.reset

这个参数是针对新的groupid中的消费者而言的，当有新groupid的消费者来消费指定的topic时，对于该参数的配置，会有不同的语义：

  • auto.offset.reset=latest情况下，新的消费者将会从其他消费者最后消费的offset处开始消费Topic下的消息
  • auto.offset.reset= earliest情况下，新的消费者会从该topic最早的消息开始消费
  • auto.offset.reset=none情况下，新的消费者加入以后，由于之前不存在offset，则会直接抛出异常。

max.poll.records

此设置限制每次调用poll返回的消息数，这样可以更容易地预测每次poll间隔要处理的最大值。通过调整此值，可以减少poll间隔。

Springboot+kafka

springboot的版本和kafka的版本，有一个对照表格，如果没有按照正确的版本来引入，那么会存在版本问题导致ClassNotFound的问题，具体请参考

https://spring.io/projects/spring-kafka

jar包依赖

<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
  <version>2.2.0.RELEASE</version>
</dependency>

KafkaProducer

@Component
public class KafkaProducer {
  @Autowired
  private KafkaTemplate<String,String> kafkaTemplate;
  public void send(){
    kafkaTemplate.send("test","msgKey","msgData");
 }
}

KafkaConsumer

@Component
public class KafkaConsumer {
  @KafkaListener(topics = {"test"})
 public void listener(ConsumerRecord record){
    Optional<?> msg=Optional.ofNullable(record.value());
    if(msg.isPresent()){
      System.out.println(msg.get());
   }
 }
}

application配置

spring.kafka.bootstrap-
servers=192.168.13.102:9092,192.168.13.103:9092,192.168.13.104:9092
spring.kafka.producer.key-
serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-
serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.consumer.group-id=test-consumer-group
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=true
spring.kafka.consumer.key-
deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-
deserializer=org.apache.kafka.common.serialization.StringDeserializer

测试

public static void main(String[] args) {
  ConfigurableApplicationContext context=SpringApplication.run
   (KafkaDemoApplication.class, args);
  KafkaProducer kafkaProducer=context.getBean(KafkaProducer.class);
  for(int i=0;i<3;i++){
    kafkaProducer.send();
    try {
      Thread.sleep(3000);
   } catch (InterruptedException e) {
      e.printStackTrace();
   }
 }
}

标签：springbootkafka 

作者:gojiuye , 分类:技术精选 , 浏览:3 , 评论:0

发表评论:

控制面板
您好，欢迎到访网站！
  查看权限
网站分类
最新留言