kafka_2.11-0.10.0.1生产者producer的Java实现

时间：2025-02-27 08:07:08

转载自：http://blog.****.net/qq_26479655/article/details/52555283

首先导入包

  1. 将kafka目录下的libs中的jar包导入
  2. 用maven建立


    

  1. <dependency>
    2. 

  2. <groupId>org.apache.kafka</groupId>
    3. 

  3. <artifactId>kafka-clients</artifactId>
    4. 

  4. <version>0.10.0.1</version>
    5. 

  5. </dependency>
    6. 





写好properties配置文件 

一下为项目结构 
kafka_2.11-0.10.0.1生产者producer的Java实现




    

  1. #kafka集群地址
    2. 

  2. bootstrap.servers = 192.168.168.200:9092
    3. 

  3. client.id = testProducer
    4. 

  4. key.serializer = org.apache.kafka.common.serialization.IntegerSerializer
    5. 

  5. value.serializer = org.apache.kafka.common.serialization.StringSerializer
    6. 





然后上代码




    

  1. package kafka.producer;
    2. 

  2. import java.io.IOException;
    3. 

  3. import java.util.Properties;
    4. 

  4. import java.util.concurrent.ExecutionException;
    5. 

  5. import org.apache.kafka.clients.producer.Callback;
    6. 

  6. import org.apache.kafka.clients.producer.KafkaProducer;
    7. 

  7. import org.apache.kafka.clients.producer.ProducerRecord;
    8. 

  8. import org.apache.kafka.clients.producer.RecordMetadata;
    9. 

  9. public class ProducerTest extends Thread{
    10. 

  10. private final KafkaProducer<Integer, String> producer;
    11. 

  11. private final String topic;
    12. 

  12. private final Boolean isAsync;
    13. 

  13. /*isAsync同步、异步*/
    14. 

  14. public ProducerTest(String topic, Boolean isAsync) {
    15. 

  15. Properties properties = new Properties();
    16. 

  16. /*加载配置文件*/
    17. 

  17. try {
    18. 

  18. properties.load(ProducerTest.class.getClassLoader().getResourceAsStream("conf/kafka.producer.properties"));
    19. 

  19. } catch (IOException e) {
    20. 

  20. e.printStackTrace();
    21. 

  21. }
    22. 

  22. producer = new KafkaProducer<>(properties);
    23. 

  23. this.topic = topic;
    24. 

  24. this.isAsync = isAsync;
    25. 

  25. }
    26. 

  26. public void run() {
    27. 

  27. int messageNo = 1;
    28. 

  28. while (true) {
    29. 

  29. String messageStr = "Message_" + messageNo;
    30. 

  30. long startTime = System.currentTimeMillis();
    31. 

  31. if (isAsync) { // Send asynchronously
    32. 

  32. producer.send(new ProducerRecord<>(topic,
    33. 

  33. messageNo,
    34. 

  34. messageStr), new DemoCallBack(startTime, messageNo, messageStr));
    35. 

  35. } else { // Send synchronously
    36. 

  36. try {
    37. 

  37. producer.send(new ProducerRecord<>(topic,
    38. 

  38. messageNo,
    39. 

  39. messageStr)).get();
    40. 

  40. System.out.println("Sent message: (" + messageNo + ", " + messageStr + ")");
    41. 

  41. } catch (InterruptedException | ExecutionException e) {
    42. 

  42. e.printStackTrace();
    43. 

  43. }
    44. 

  44. }
    45. 

  45. ++messageNo;
    46. 

  46. }
    47. 

  47. }
    48. 

  48. }
    49. 

  49. class DemoCallBack implements Callback {
    50. 

  50. private final long startTime;
    51. 

  51. private final int key;
    52. 

  52. private final String message;
    53. 

  53. public DemoCallBack(long startTime, int key, String message) {
    54. 

  54. this.startTime = startTime;
    55. 

  55. this.key = key;
    56. 

  56. this.message = message;
    57. 

  57. }
    58. 

  58. /**
    59. 

  59. * @param metadata  The metadata for the record that was sent (i.e. the partition and offset). Null if an error
    60. 

  60. *                  occurred.
    61. 

  61. * @param exception The exception thrown during processing of this record. Null if no error occurred.
    62. 

  62. */
    63. 

  63. public void onCompletion(RecordMetadata metadata, Exception exception) {
    64. 

  64. long elapsedTime = System.currentTimeMillis() - startTime;
    65. 

  65. if (metadata != null) {
    66. 

  66. System.out.println(
    67. 

  67. "message(" + key + ", " + message + ") sent to partition(" + metadata.partition() +
    68. 

  68. "), " +
    69. 

  69. "offset(" + metadata.offset() + ") in " + elapsedTime + " ms");
    70. 

  70. } else {
    71. 

  71. exception.printStackTrace();
    72. 

  72. }
    73. 

  73. }
    74. 

  74. }
    75. 



测试代码




    

  1. package kafka.producer;
    2. 

  2. public class Main {
    3. 

  3. public static void main(String[] args) {
    4. 

  4. ProducerTest test = new ProducerTest("TestTopic", true);
    5. 

  5. test.start();
    6. 

  6. }
    7. 

  7. }
    8. 





运行结果




    

  1. message(51215, Message_51215) sent to partition(0), offset(161830) in 3497 ms
    2. 

  2. message(51216, Message_51216) sent to partition(0), offset(161831) in 3497 ms
    3. 

  3. message(51217, Message_51217) sent to partition(0), offset(161832) in 3497 ms
    4. 

  4. message(51218, Message_51218) sent to partition(0), offset(161833) in 3497 ms
    5. 

  5. message(51219, Message_51219) sent to partition(0), offset(161834) in 3497 ms
    6. 

  6. message(51220, Message_51220) sent to partition(0), offset(161835) in 3497 ms
    7. 

  7. message(51221, Message_51221) sent to partition(0), offset(161836) in 3497 ms
    8. 

  8. message(51222, Message_51222) sent to partition(0), offset(161837) in 3497 ms
    9. 

  9. message(51223, Message_51223) sent to partition(0), offset(161838) in 3497 ms
    10. 

  10. message(51224, Message_51224) sent to partition(0), offset(161839) in 3497 ms
    11. 

  11. message(51225, Message_51225) sent to partition(0), offset(161840) in 3497 ms
    12. 

  12. message(51226, Message_51226) sent to partition(0), offset(161841) in 3497 ms
    13. 

  13. message(51227, Message_51227) sent to partition(0), offset(161842) in 3497 ms
    14. 

  14. message(51228, Message_51228) sent to partition(0), offset(161843) in 3497 ms
    15. 

  15. .............​
    16. 



相关文章