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 条评论
登录 后参与评论

相关文章

来自专栏互联网研发闲思录

在多线程中使用spring的bean

  由于spring在java开发中的广泛运用大大的方便了开发的同时,当运用一些技术比如多线程等 在由spring管理的配置文件中,可以通过封装spring提供...

1760
来自专栏Java学习网

Java中使用线程时,请不要忘记Spring TaskExecutor组件

当我们实现的web应用程序需要长时间运行一个任务时,Spring TaskExecutor管理组件是一个很好选择,会给我们代码的实现提供很大的方便,也会节省时间...

3045
来自专栏A周立SpringCloud

Spring Cloud Edgware新特性之八:Zuul回退的改进

Spring Cloud Edgware对Hystrix回退的逻辑进行了一些改进。本文将信息探讨新旧版本的回退操作,并分析的原因及改进后的优势。 Dalston...

2636
来自专栏Android知识点总结

去吧!设计模式之模板方法模式

552
来自专栏码匠的流水账

聊聊HystrixThreadPool

hystrix-core-1.5.12-sources.jar!/com/netflix/hystrix/HystrixThreadPool.java

491
来自专栏Java架构师学习

Java Map中常遇到的几个问题

1.将Map转化成List Map接口提供了三种collection:key set,value set 和 key-value set,每一种都可以转成Lis...

2654
来自专栏Spring相关

第5章—构建Spring Web应用程序—关于spring中的validate注解后台校验的解析

在后台开发过程中,对参数的校验成为开发环境不可缺少的一个环节。比如参数不能为null,email那么必须符合email的格式,如果手动进行if判断或者写正则表达...

631
来自专栏编程软文

redis是如何存储对象和集合的

1927
来自专栏ImportSource

Spring5以来注册Bean的各种姿势,特别最后的纯编码注册值得尝试

各位好,今天我们的内容是有关Spring 5以来有关注册bean的几种方式。前面两三个是比较常用的方式,最后两种是只有在特殊的场合下才会被用到和想到。我们会分别...

6046
来自专栏Spring相关

关于spring中的validate注解后台校验的解析

在后台开发过程中,对参数的校验成为开发环境不可缺少的一个环节。比如参数不能为null,email那么必须符合email的格式,如果手动进行if判断或者写正则表达...

861

扫码关注云+社区