java实现Kafka生产者Producer工具类

摘 要

本文将介绍java实现Kafka生产者Producer的简单工具类

相关版本

kafka:kafka_2.10-0.10.1.1

jdk:1.7

相关代码实现

package com.itunic.util;

import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import net.sf.json.JSONObject;

public class KafkaTools {
 /**
     * 
     * 私有静态方法,创建Kafka生产者
     * 
     * @author IG
     * @Date 2017年4月14日 上午10:32:32
     * @version 1.0.0
     * @return KafkaProducer
     */
 private static KafkaProducer<String, String> createProducer() {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "huakServer1:9092,huakServer2:9092");// 声明kafka
 // properties.put("value.serializer",
 // "org.apache.kafka.common.serialization.ByteArraySerializer");
 // properties.put("key.serializer",
 // "org.apache.kafka.common.serialization.ByteArraySerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 return new KafkaProducer<String, String>((properties));
    }

 /**
     * 
     * 传入kafka约定的topicName,json格式字符串,发送给kafka集群
     * 
     * @author IG
     * @Date 2017年4月14日 下午1:29:09
     * @version 1.0.0
     * @param topicName
     * @param jsonMessage
     */
 public static void sendMessage(String topicName, String jsonMessage) {
        KafkaProducer<String, String> producer = createProducer();
        producer.send(new ProducerRecord<String, String>(topicName, jsonMessage));
        producer.close();
    }

 /**
     * 
     * 传入kafka约定的topicName,json格式字符串数组,发送给kafka集群<br> 
     * 用于批量发送消息,性能较高。
     * 
     * @author IG
     * @Date 2017年4月14日 下午2:00:12
     * @version 1.0.0
     * @param topicName
     * @param jsonMessages
     * @throws InterruptedException
     */
 public static void sendMessage(String topicName, String... jsonMessages) throws InterruptedException {
        KafkaProducer<String, String> producer = createProducer();
 for (String jsonMessage : jsonMessages) {
            producer.send(new ProducerRecord<String, String>(topicName, jsonMessage));
        }
        producer.close();
    }

 /**
     * 
     * 传入kafka约定的topicName,Map集合,内部转为json发送给kafka集群 <br>
     * 用于批量发送消息,性能较高。
     * 
     * @author IG
     * @Date 2017年4月14日 下午2:01:18
     * @version 1.0.0
     * @param topicName
     * @param mapMessageToJSONForArray
     */
 public static void sendMessage(String topicName, List<Map<Object, Object>> mapMessageToJSONForArray) {
        KafkaProducer<String, String> producer = createProducer();
 for (Map<Object, Object> mapMessageToJSON : mapMessageToJSONForArray) {
            String array = JSONObject.fromObject(mapMessageToJSON).toString();
            producer.send(new ProducerRecord<String, String>(topicName, array));
        }
        producer.close();
    }

 /**
     * 
     * 传入kafka约定的topicName,Map,内部转为json发送给kafka集群
     * 
     * @author IG
     * @Date 2017年4月14日 下午1:30:10
     * @version 1.0.0
     * @param topicName
     * @param mapMessageToJSON
     */
 public static void sendMessage(String topicName, Map<Object, Object> mapMessageToJSON) {
        KafkaProducer<String, String> producer = createProducer();
        String array = JSONObject.fromObject(mapMessageToJSON).toString();
        producer.send(new ProducerRecord<String, String>(topicName, array));
        producer.close();
    }

 public static void main(String[] args) throws InterruptedException {
 // System.out.println(System.getProperty("file.encoding"));
        String[] s = new String[] { "{\"userName\":\"赵四31\",\"pwd\":\"lisi\",\"age\":13}",
 "{\"userName\":\"赵四41\",\"pwd\":\"lisi\",\"age\":14}",
 "{\"userName\":\"赵四51\",\"pwd\":\"lisi\",\"age\":15}" };
 // KafkaTools.sendMessage("logstest",
 // "{\"userName\":\"赵四\",\"pwd\":\"lisi\",\"age\":13}");
 /*
         * for (String a : s) { System.out.println(a); Thread.sleep(3000);
         * KafkaTools.sendMessage(topicName, jsonMessages); }
         */
         KafkaTools.sendMessage("logstest", s);
    }

}

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏牛肉圆粉不加葱

[Spark源码剖析] DAGScheduler划分stage划分stage源码剖析

在DAGScheduler内部通过post一个JobSubmitted事件来触发Job的提交

953
来自专栏Spark生态圈

[spark] Task执行流程

在文章TaskScheduler 任务提交与调度源码解析 中介绍了Task在executor上的逻辑分配,调用TaskSchedulerImpl的resourc...

1881
来自专栏个人分享

Spark常用函数(源码阅读六)

  源码层面整理下我们常用的操作RDD数据处理与分析的函数,从而能更好的应用于工作中。

1042
来自专栏恰童鞋骚年

Hadoop学习笔记—7.计数器与自定义计数器

  在上图所示中,计数器有19个,分为四个组:File Output Format Counters、FileSystemCounters、File Input...

1132
来自专栏码字搬砖

Spark Streaming 中使用 zookeeper 保存 offset 并重用 Java版

最近中使用spark Streaming +kafka,由于涉及到金额,所以需要保证at only one, 而网上关于java版的kafka offset...

3282
来自专栏Spark学习技巧

Spark源码系列之foreach和foreachPartition的区别

一,基本使用 1,RDD分布式数据集的五大特性 1),A list of partitions(一系列的分区) 2),A function for comput...

4717
来自专栏拭心的安卓进阶之路

Android 进阶6:两种序列化方式 Serializable 和 Parcelable

什么是序列化 我们总是说着或者听说着“序列化”,它的定义是什么呢? 序列化 (Serialization)将对象的状态信息转换为可以存储或传输的形式的过程。在...

3175
来自专栏Spark生态圈

[spark] DAGScheduler 提交stage源码解析

DAGScheduler在划分完Stage后([spark] DAGScheduler划分stage源码解析 ),将会通过submitStage(finalSt...

1113
来自专栏about云

spark2的SparkSession思考与总结2:SparkSession有哪些函数及作用是什么

问题导读 1.spark SparkSession包含哪些函数? 2.创建DataFrame有哪些函数? 3.创建DataSet有哪些函数? 上一篇spa...

7545
来自专栏浪淘沙

KafKa 代码实现

2093

扫码关注云+社区

领取腾讯云代金券