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

相关文章

来自专栏转载gongluck的CSDN博客

cocos2dx 打灰机

#include "GamePlane.h" #include "PlaneSprite.h" #include "BulletNode.h" #include...

5436
来自专栏pangguoming

Spring Boot集成JasperReports生成PDF文档

由于工作需要,要实现后端根据模板动态填充数据生成PDF文档,通过技术选型,使用Ireport5.6来设计模板,结合JasperReports5.6工具库来调用渲...

1.2K7
来自专栏陈仁松博客

ASP.NET Core 'Microsoft.Win32.Registry' 错误修复

今天在发布Asp.net Core应用到Azure的时候出现错误InvalidOperationException: Cannot find compilati...

4848
来自专栏一个会写诗的程序员的博客

Spring Reactor 项目核心库Reactor Core

Non-Blocking Reactive Streams Foundation for the JVM both implementing a Reactiv...

2152
来自专栏张善友的专栏

Silverlight + Model-View-ViewModel (MVVM)

     早在2005年,John Gossman写了一篇关于Model-View-ViewModel模式的博文,这种模式被他所在的微软的项目组用来创建Expr...

2968
来自专栏闻道于事

js登录滑动验证,不滑动无法登陆

js的判断这里是根据滑块的位置进行判断,应该是用一个flag判断 <%@ page language="java" contentType="text/html...

6768
来自专栏ASP.NETCore

ASP.NET Core 整合Autofac和Castle实现自动AOP拦截

除了ASP.NETCore自带的IOC容器外,我们还可以使用其他成熟的DI框架,如Autofac,StructureMap等(笔者只用过Unity,Ninjec...

674
来自专栏hbbliyong

WPF Trigger for IsSelected in a DataTemplate for ListBox items

<DataTemplate DataType="{x:Type vm:HeaderSlugViewModel}"> <vw:HeaderSlug...

4064
来自专栏我和未来有约会

Kit 3D 更新

Kit3D is a 3D graphics engine written for Microsoft Silverlight. Kit3D was inita...

2526
来自专栏大内老A

The .NET of Tomorrow

Ed Charbeneau(http://developer.telerik.com/featured/the-net-of-tomorrow/) Exciti...

31610

扫码关注云+社区