当前位置: 首页 > 编程笔记 >

Kafka Java Producer代码实例详解

裴令秋
2023-03-14
本文向大家介绍Kafka Java Producer代码实例详解,包括了Kafka Java Producer代码实例详解的使用技巧和注意事项,需要的朋友参考一下

根据业务需要可以使用Kafka提供的Java Producer API进行产生数据,并将产生的数据发送到Kafka对应Topic的对应分区中,入口类为:Producer

Kafka的Producer API主要提供下列三个方法

  •   public void send(KeyedMessage<K,V> message) 发送单条数据到Kafka集群
  •   public void send(List<KeyedMessage<K,V>> messages) 发送多条数据(数据集)到Kafka集群
  •   public void close() 关闭Kafka连接资源

一、JavaKafkaProducerPartitioner:自定义的数据分区器,功能是:决定输入的key/value键值对的message发送到Topic的那个分区中,返回分区id,范围:[0,分区数量); 这里的实现比较简单,根据key中的数字决定分区的值。具体代码如下:

import kafka.producer.Partitioner;
import kafka.utils.VerifiableProperties;

/**
 * Created by gerry on 12/21.
 */
public class JavaKafkaProducerPartitioner implements Partitioner {

  /**
   * 无参构造函数
   */
  public JavaKafkaProducerPartitioner() {
    this(new VerifiableProperties());
  }

  /**
   * 构造函数,必须给定
   *
   * @param properties 上下文
   */
  public JavaKafkaProducerPartitioner(VerifiableProperties properties) {
    // nothings
  }

  @Override
  public int partition(Object key, int numPartitions) {
    int num = Integer.valueOf(((String) key).replaceAll("key_", "").trim());
    return num % numPartitions;
  }
}

二、 JavaKafkaProducer:通过Kafka提供的API进行数据产生操作的测试类;具体代码如下:

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import org.apache.log4j.Logger;

import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.ThreadLocalRandom;

/**
 * Created by gerry on 12/21.
 */
public class JavaKafkaProducer {
  private Logger logger = Logger.getLogger(JavaKafkaProducer.class);
  public static final String TOPIC_NAME = "test";
  public static final char[] charts = "qazwsxedcrfvtgbyhnujmikolp1234567890".toCharArray();
  public static final int chartsLength = charts.length;


  public static void main(String[] args) {
    String brokerList = "192.168.187.149:9092";
    brokerList = "192.168.187.149:9092,192.168.187.149:9093,192.168.187.149:9094,192.168.187.149:9095";
    brokerList = "192.168.187.146:9092";
    Properties props = new Properties();
    props.put("metadata.broker.list", brokerList);
    /**
     * 0表示不等待结果返回<br/>
     * 1表示等待至少有一个服务器返回数据接收标识<br/>
     * -1表示必须接收到所有的服务器返回标识,及同步写入<br/>
     * */
    props.put("request.required.acks", "0");
    /**
     * 内部发送数据是异步还是同步
     * sync:同步, 默认
     * async:异步
     */
    props.put("producer.type", "async");
    /**
     * 设置序列化的类
     * 可选:kafka.serializer.StringEncoder
     * 默认:kafka.serializer.DefaultEncoder
     */
    props.put("serializer.class", "kafka.serializer.StringEncoder");
    /**
     * 设置分区类
     * 根据key进行数据分区
     * 默认是:kafka.producer.DefaultPartitioner ==> 安装key的hash进行分区
     * 可选:kafka.serializer.ByteArrayPartitioner ==> 转换为字节数组后进行hash分区
     */
    props.put("partitioner.class", "JavaKafkaProducerPartitioner");

    // 重试次数
    props.put("message.send.max.retries", "3");

    // 异步提交的时候(async),并发提交的记录数
    props.put("batch.num.messages", "200");

    // 设置缓冲区大小,默认10KB
    props.put("send.buffer.bytes", "102400");

    // 2. 构建Kafka Producer Configuration上下文
    ProducerConfig config = new ProducerConfig(props);

    // 3. 构建Producer对象
    final Producer<String, String> producer = new Producer<String, String>(config);

    // 4. 发送数据到服务器,并发线程发送
    final AtomicBoolean flag = new AtomicBoolean(true);
    int numThreads = 50;
    ExecutorService pool = Executors.newFixedThreadPool(numThreads);
    for (int i = 0; i < 5; i++) {
      pool.submit(new Thread(new Runnable() {
        @Override
        public void run() {
          while (flag.get()) {
            // 发送数据
            KeyedMessage message = generateKeyedMessage();
            producer.send(message);
            System.out.println("发送数据:" + message);

            // 休眠一下
            try {
              int least = 10;
              int bound = 100;
              Thread.sleep(ThreadLocalRandom.current().nextInt(least, bound));
            } catch (InterruptedException e) {
              e.printStackTrace();
            }
          }

          System.out.println(Thread.currentThread().getName() + " shutdown....");
        }
      }, "Thread-" + i));

    }

    // 5. 等待执行完成
    long sleepMillis = 600000;
    try {
      Thread.sleep(sleepMillis);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    flag.set(false);

    // 6. 关闭资源

    pool.shutdown();
    try {
      pool.awaitTermination(6, TimeUnit.SECONDS);
    } catch (InterruptedException e) {
    } finally {
      producer.close(); // 最后之后调用
    }
  }

  /**
   * 产生一个消息
   *
   * @return
   */
  private static KeyedMessage<String, String> generateKeyedMessage() {
    String key = "key_" + ThreadLocalRandom.current().nextInt(10, 99);
    StringBuilder sb = new StringBuilder();
    int num = ThreadLocalRandom.current().nextInt(1, 5);
    for (int i = 0; i < num; i++) {
      sb.append(generateStringMessage(ThreadLocalRandom.current().nextInt(3, 20))).append(" ");
    }
    String message = sb.toString().trim();
    return new KeyedMessage(TOPIC_NAME, key, message);
  }

  /**
   * 产生一个给定长度的字符串
   *
   * @param numItems
   * @return
   */
  private static String generateStringMessage(int numItems) {
    StringBuilder sb = new StringBuilder();
    for (int i = 0; i < numItems; i++) {
      sb.append(charts[ThreadLocalRandom.current().nextInt(chartsLength)]);
    }
    return sb.toString();
  }
}

三、Pom.xml依赖配置如下

<properties>
  <kafka.version>0.8.2.1</kafka.version>
</properties>

<dependencies>
  <dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.10</artifactId>
    <version>${kafka.version}</version>
  </dependency>
</dependencies>

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持小牛知识库。

 类似资料:
  • 本文向大家介绍Java Annotation详解及实例代码,包括了Java Annotation详解及实例代码的使用技巧和注意事项,需要的朋友参考一下 一、Annotation简介 从Java1.5开始,Java增加了元数据(MetaData)的支持,也就是Annotation(注释); Annotation能被用来为程序元素(类、方法、成员变量等)设置元数据; Annotation不能影响程序代

  • 本文向大家介绍java HashMap详解及实例代码,包括了java HashMap详解及实例代码的使用技巧和注意事项,需要的朋友参考一下 java HashMap  Map集合的遍历 方式1,根据键查询值 获取所有键的集合 遍历键的集合,获取每一个键 根据键,查询值 方式2,根据键值对的对象查询键和值 获取所有键值对的对象的集合 遍历键值对的对象的集合,获取到每一个键值对的对象 根据键值对的对象

  • 本文向大家介绍ReactNative Alert详解及实例代码,包括了ReactNative Alert详解及实例代码的使用技巧和注意事项,需要的朋友参考一下 Alert顾名思义一就是一个警告框,一般使用情况比如:退出登录,清楚缓存,提示修改密码等等。。。ReactNative中的Alert只有一个静态方法alert()其中有四个参数:标题,信息,按钮和按钮类型 在Android按钮至多有三个 下

  • 本文向大家介绍Angularjs CURD 详解及实例代码,包括了Angularjs CURD 详解及实例代码的使用技巧和注意事项,需要的朋友参考一下 Angularjs CURD 前言        基于一个手机端的项目使用了angularjs,硬着头皮去用,有很多的疑问还需要一一去验证,刚开始总是感觉找不到北,总是感觉有很多概念,而且似乎ng既夹杂MVC又夹杂MVVM的思想, 忙里偷闲敲了个简

  • 本文向大家介绍java 实现 stack详解及实例代码,包括了java 实现 stack详解及实例代码的使用技巧和注意事项,需要的朋友参考一下 栈是限制插入和删除只能在一个位置上进行的 List,该位置是 List 的末端,叫做栈的顶(top),对于栈的基本操作有 push 和 pop,前者是插入,后者是删除。 栈也是 FIFO 表。 栈的实现有两种,一种是使用数组,一种是使用链表。 栈的应用 平

  • 本文向大家介绍Android 混淆代码详解及实例,包括了Android 混淆代码详解及实例的使用技巧和注意事项,需要的朋友参考一下 为了防止自己的劳动成果被别人窃取,混淆代码能有效防止被反编译,下面来总结以下混淆代码的步骤: 1. 大家也许都注意到新建一个工程会看到项目下边有这样proguard-project.txt一个文件,这个对混淆代码很重要,如果你不小心删掉了,没关系,从其他地方拷贝一个过