您现在的位置是：首页 > 大数据

当前栏目

AMQP: Kafka code

Kafka code

2023-09-11 14:16:16 时间

No Callback:

package io.veer;


import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class Producer{
  public static void main(String[] args){
    Properties properties = new Properties();
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.8.11:9092,192.168.8.12:9092,192.168.8.13:9092");
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    // 创建KafkaProducer对象
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

    // 发送数据
    for(int i = 0; i < 5; i++){
      kafkaProducer.send(new ProducerRecord<>("zion", "etymology" + i));
    }
    kafkaProducer.close();
  }
}

Callback:

package io.veer;


import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class Producer{
  public static void main(String[] args){
    Properties properties = new Properties();
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.8.11:9092,192.168.8.12:9092,192.168.8.13:9092");
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    // 创建KafkaProducer对象
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

    // 发送数据
    for(int i = 0; i < 5; i++){
      kafkaProducer.send(new ProducerRecord<>("zion", "etymology" + i), new Callback(){
        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e){
          if(e == null){
            System.out.printf("\033[37;7m Topic: %s, Partition: %s \033[0m\n", recordMetadata.topic(), recordMetadata.partition());
          }
        }
      });
    }
    kafkaProducer.close();
  }
}

同步发送:

package io.veer;


import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class Producer{
  public static void main(String[] args) throws ExecutionException, InterruptedException{
    Properties properties = new Properties();
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.8.11:9092,192.168.8.12:9092,192.168.8.13:9092");
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    // 创建KafkaProducer对象
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

    // 发送数据
    for(int i = 0; i < 5; i++){
      kafkaProducer.send(new ProducerRecord<>("zion", "etymology" + i)).get();  // 加入.get() 变成同步
    }
    kafkaProducer.close();
  }
}

指定partition, partition必须存在, 否则hang:

package io.veer;


import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class Producer{
  public static void main(String[] args){
    Properties properties = new Properties();
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.8.11:9092,192.168.8.12:9092,192.168.8.13:9092");
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    // 创建KafkaProducer对象
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

    // 发送数据
    for(int i = 0; i < 5; i++){
      kafkaProducer.send(new ProducerRecord<>("zion", 0, "", "etymology" + i), new Callback(){
        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e){
          if(e == null){
            System.out.printf("\033[37;7m Topic: %s, Partition: %s \033[0m\n", recordMetadata.topic(), recordMetadata.partition());
          }
        }
      });
    }
    kafkaProducer.close();
  }
}

不指定partition, 根据key的hash值取模

package io.veer;


import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class Producer{
  public static void main(String[] args){
    Properties properties = new Properties();
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.8.11:9092,192.168.8.12:9092,192.168.8.13:9092");
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    // 创建KafkaProducer对象
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

    // 发送数据
    for(int i = 0; i < 5; i++){
      kafkaProducer.send(new ProducerRecord<>("zion", "moniker", "etymology" + i), new Callback(){
        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e){
          if(e == null){
            System.out.printf("\033[37;7m Topic: %s, Partition: %s \033[0m\n", recordMetadata.topic(), recordMetadata.partition());
          }
        }
      });
    }
    kafkaProducer.close();
  }
}

既没有partition值又没有key值的情况下，Kafka采用Sticky Partition（黏性分区器），会随机选择一个分区，并尽可能一直使用该分区，待该分区的batch已满或者已完成，Kafka再随机一个分区进行使用（和上一次的分区不同）

public class Producer{
  public static void main(String[] args) throws InterruptedException{
    Properties properties = new Properties();
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.8.11:9092,192.168.8.12:9092,192.168.8.13:9092");
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    // 创建KafkaProducer对象
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

    // 发送数据
    for(int i = 0; i < 50; i++){
      kafkaProducer.send(new ProducerRecord<>("zion", "etymology" + i), new Callback(){
        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e){
          if(e == null){
            System.out.printf("\033[37;7m Topic: %s, Partition: %s \033[0m\n", recordMetadata.topic(), recordMetadata.partition());
          }
        }
      });
      Thread.sleep(2);
    }
    kafkaProducer.close();
  }
}

自定义partitioner:

package io.veer;


import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

import java.util.Map;

public class MyPartitioner implements Partitioner{
  @Override
  public int partition(String s, Object o, byte[] bytes, Object o1, byte[] bytes1, Cluster cluster){
    String msg = o1.toString();
    System.out.println("msg = " + msg);
    int partition;
    if(msg.contains("moniker")){
      partition = 0;
    }else{
      partition = 1;
    }
    return partition;
  }

  @Override
  public void close(){

  }

  @Override
  public void configure(Map<String, ?> map){

  }
}

package io.veer;


import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class Producer{
  public static void main(String[] args) throws InterruptedException{
    Properties properties = new Properties();
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.8.11:9092,192.168.8.12:9092,192.168.8.13:9092");
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    // 关联自定义分区器
    properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "io.veer.MyPartitioner");
    // 创建KafkaProducer对象
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);


    // 发送数据
    for(int i = 0; i < 5; i++){
      kafkaProducer.send(new ProducerRecord<>("zion", "etymology" + i), new Callback(){
        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e){
          if(e == null){
            System.out.printf("\033[37;7m Topic: %s, Partition: %s \033[0m\n", recordMetadata.topic(), recordMetadata.partition());
          }
        }
      });
      Thread.sleep(2);
    }
    for(int i = 0; i < 5; i++){
      kafkaProducer.send(new ProducerRecord<>("zion", "moniker" + i), new Callback(){
        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e){
          if(e == null){
            System.out.printf("\033[37;7m Topic: %s, Partition: %s \033[0m\n", recordMetadata.topic(), recordMetadata.partition());
          }
        }
      });
    }
    kafkaProducer.close();
  }
}

设置参数

package io.veer;


import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class Producer{
  public static void main(String[] args) throws InterruptedException{
    Properties properties = new Properties();
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.8.11:9092,192.168.8.12:9092,192.168.8.13:9092");
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    // 关联自定义分区器
    properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "io.veer.MyPartitioner");
    properties.put(ProducerConfig.BATCH_SIZE_CONFIG, (int) Math.pow(2, 14));
    properties.put(ProducerConfig.LINGER_MS_CONFIG, 10);
    properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, (long) Math.pow(2, 25));
    properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
    // 创建KafkaProducer对象
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);


    // 发送数据
    for(int i = 0; i < 5; i++){
      kafkaProducer.send(new ProducerRecord<>("zion", "etymology" + i), new Callback(){
        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e){
          if(e == null){
            System.out.printf("\033[37;7m Topic: %s, Partition: %s \033[0m\n", recordMetadata.topic(), recordMetadata.partition());
          }
        }
      });
      Thread.sleep(2);
    }
    for(int i = 0; i < 5; i++){
      kafkaProducer.send(new ProducerRecord<>("zion", "moniker" + i), new Callback(){
        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e){
          if(e == null){
            System.out.printf("\033[37;7m Topic: %s, Partition: %s \033[0m\n", recordMetadata.topic(), recordMetadata.partition());
          }
        }
      });
    }
    Thread.sleep(1000 * 5);
    kafkaProducer.close();
  }
}

开启事务:

package io.veer;


import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class Producer{
  public static void main(String[] args) throws InterruptedException{
    Properties properties = new Properties();
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.8.11:9092,192.168.8.12:9092,192.168.8.13:9092");
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    // 关联自定义分区器
    properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "io.veer.MyPartitioner");
    // 指定事务ID, 全局唯一
    properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactional_id_01");
    properties.put(ProducerConfig.BATCH_SIZE_CONFIG, (int) Math.pow(2, 14));
    properties.put(ProducerConfig.LINGER_MS_CONFIG, 10);
    properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, (long) Math.pow(2, 25));
    properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
    // 创建KafkaProducer对象
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);


    kafkaProducer.initTransactions();
    kafkaProducer.beginTransaction();
    try{
      for(int i = 0; i < 5; i++){
        kafkaProducer.send(new ProducerRecord<>("zion", "etymology" + i), new Callback(){
          @Override
          public void onCompletion(RecordMetadata recordMetadata, Exception e){
            if(e == null){
              System.out.printf("\033[37;7m Topic: %s, Partition: %s \033[0m\n", recordMetadata.topic(), recordMetadata.partition());
            }
          }
        });
        Thread.sleep(2);
      }
      int b = 1 / 0;
      kafkaProducer.commitTransaction();
    }catch(Exception e){
      e.printStackTrace();
      kafkaProducer.abortTransaction();
    }finally{
      kafkaProducer.close();
    }

  }
}

数据有序:

猜你喜欢

ASP.net 判断上传文件类型的三种方法
打印赛道 - 完成车模调试与比赛
比核武器更恐怖？巴菲特说网络攻击成人类最大威胁
GIT error: You have not concluded your merge (MERGE_HEAD exists)
品高云助力南车时代电气云化之路
Three.js Example 注解 —— CombinedCamera.js
js模拟键盘事件
shell判断一个变量是否为空
MySQL基础之标准模式通配符
拖拽效果的实现原理分析2
Android Uri 转 Path
MVC使用jQuery从视图向控制器传递Model的2种方法
使用SourceTree管理Unity工程推送报错 “ RPC failed； curl 55 Send failure: Connection was aborted“
阿里巴巴NASA新科技“麒麟”首露面将应用于阿里云
协议的注冊与维护——ndpi源代码分析
HDU 2196 Computer (树形DP)

相关主题

Kafka 配置
集成Kafka
Kafka生产者
Spring Boot Kafka
kafka实战
Kafka 学习笔记
kafka—生产者
kafka的使用
Kafka笔记7
kafka笔记1
Kafka 与 RabbitMQ
spring集成kafka
入门 Kafka

zl程序教程

当前栏目

AMQP: Kafka code

相关文章