文档章节

kafka java 基本使用

屌丝Lee
 屌丝Lee
发布于 2016/06/23 19:07
字数 553
阅读 1186
收藏 3

maven依赖:

<dependency>
	<groupId>org.apache.kafka</groupId>
	<artifactId>kafka_2.11</artifactId>
	<version>0.9.0.0</version>
</dependency>

kafka producer连接,注意import:

import java.util.Properties;
import org.apache.kafka.clients.producer.ProducerRecord;
import config.Configloader;
import config.MsgCreater;
import config.Timer;
import threads.WriteKFKThreads;

public class TestKFKWriteQPS {
	
	private static int count_callback=0;
	private static long timeStart=0;
	private static int count;
	private static int threads;
	private static int length;
	private static String topic;

	public static void main(String[] args) throws Exception {
		
		Properties kp=Configloader.getProperties("../kafka.properties");
		Properties mp=Configloader.getProperties("../main.properties");
		count=Integer.parseInt(mp.getProperty("w_count"));
		threads=Integer.parseInt(mp.getProperty("w_threads"));
		count_callback = threads;
		length=Integer.parseInt(mp.getProperty("msg_len"));
		topic=mp.getProperty("topic");
		
		ProducerRecord<String,String> data = new ProducerRecord<String, String>(topic, MsgCreater.getMsg(length));
		timeStart=System.currentTimeMillis();
		
		for(int i=0;i<threads;i++){
			new Thread(new WriteKFKThreads(count, kp, data)).start();
		}
	}
	
	public static void threadCallback(){
		count_callback--;
		if(count_callback==0){
			long used =System.currentTimeMillis()-timeStart;
			System.out.println("SET --END-- "+count*threads+" time : "+ ((double) used/1000));
			System.out.println("QPS: "+count*threads/((double) used/1000));
			System.out.println(Timer.getAllDelayDistribution());
		}
	}
	
}

kafka set 线程 请无视Timer:

import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import config.Timer;
import runner.TestKFKWriteQPS;

public class WriteKFKThreads implements Runnable{

	int count=0;
	Producer<String, String> producer;
	ProducerRecord<String, String> data;
	public WriteKFKThreads(int count , Properties config ,ProducerRecord<String, String> data) {
		this.count=count;
		this.producer =new KafkaProducer<String,String>(config); 
		this.data = data;
	}

	@Override
	public void run() {
		Timer t=new Timer();
		for(int i=0;i<count;i++){
			long start = System.currentTimeMillis();
			try {
					producer.send(data).get();				
					long take =System.currentTimeMillis() - start;
					t.inTime(take);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		Timer.threadAdder(t);
		TestKFKWriteQPS.threadCallback();
		producer.close();
	}
}

kafka consumer连接  :

import java.util.Arrays;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import config.Configloader;

public class TestKFKReadQPS {

	private static long timeStart = 0;
	private static int count;
	private static int threads;
	private static String topic;
	private static int total;
	

	public static void main(String[] args) throws Exception {

		Properties kp = Configloader.getProperties("../kafka.properties");
		Properties mp = Configloader.getProperties("../main.properties");
		count = Integer.parseInt(mp.getProperty("r_count"));
		threads = Integer.parseInt(mp.getProperty("r_threads"));
		topic = mp.getProperty("topic");
		total = count * threads;
	//	partition=Integer.parseInt(mp.getProperty("partition"));
	//	offset = Long.parseLong(mp.getProperty("offset"));
		
		KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(kp);
	//	consumer.seek(new TopicPartition(topic, partition), offset);
		


		boolean goon = true;
		consumer.subscribe(Arrays.asList(topic));
		int times=1000;
		while (goon) {
			ConsumerRecords<String, String> records = consumer.poll(100);
			if(records.count() == 0){
				times--;
				if(times <=0){
					goon=false;
					System.out.println("data loose , now get : "+ (count * threads-total));
					break;
				}
			}
			for (ConsumerRecord<String, String> record : records) {
//				 System.out.printf("partition = %d, offset = %d, key = %s,value = %s \n", record.partition(), record.offset(),
//				record.key(), record.value());
				setStart();
				total--;
				if(total <=0){
					goon = false;
					break;

				}
			}
		}
		consumer.close();
		long used =System.currentTimeMillis()-timeStart;
		System.out.println("GET --END-- "+count*threads+" time : "+ ((double) used/1000));
		System.out.println("QPS: "+count*threads/((double) used/1000));

	}
	private static void setStart(){
		if(timeStart == 0){
			timeStart = System.currentTimeMillis();
			System.out.println("Read start!");
		}
	}

}

配置文件 kafka.properties:

group.id=test
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer
zookeeper.connect=127.0.0.1:2181
bootstrap.servers=127.0.0.1:9092
metadata.broker.list=127.0.0.1:9092
serializer.class=kafka.serializer.StringEncoder
enable.auto.commit=true
auto.commit.interval.ms=1000
session.timeout.ms=30000
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

main.properties参数很简单,各位自己改改就行。

注:kafka部署时候server是按照linux机器hostname做白名单的(貌似mac下是IP?),也可以设置广播hostname

server.properties中添加:advertised.host.name = XX  

XX最好配置为IP,否则就要在本地host指定 IP =XX

 

 

© 著作权归作者所有

屌丝Lee
粉丝 1
博文 35
码字总数 14663
作品 0
杭州
私信 提问
Kafka连接器深度解读之JDBC源连接器

在现实业务中,Kafka经常会遇到的一个集成场景就是,从数据库获取数据,因为关系数据库是一个非常丰富的事件源。数据库中的现有数据以及对该数据的任何更改都可以流式传输到Kafka主题中,在这...

李玉珏
03/12
2.1K
4
Kafka与操作系统适配性研究

在《Apache Kafka实战》一书中,提到了操作系统与Kafka的适配性。涉及的的epoll与select问题,以及linux零拷贝技术较为重要,因此在此处将操作系统相关特性部分进行记录。 clients 底层网络库...

analanxingde
04/24
0
0
Grafana、elasticsearch、kafka、logstash和pinpoint结合

一、Grafana 1)下载安装 wget https://s3-us-west-2.amazonaws.com/grafana-releases/release/grafana-4.2.0-1.x8664.rpm sudo yum localinstall grafana-4.2.0-1.x86_64.rpm 2)启动 serv......

半船水
2017/10/25
301
0
阿里年薪50WJAVA工程师转大数据学习路线!

大数据有两个方向,一个是偏计算机的,另一个是偏经济的。你学过Java,所以你可以偏将计算机的。 Java程序员想转大数据可行吗?Java是全世界使用人数最多的编程语言。不少程序员选择Java做为...

JAVA丶学习
2018/04/25
0
0
Zookeeper+Kafka集群搭建

Zookeeper集群搭建 Kafka集群是把状态保存在Zookeeper中的,首先要搭建Zookeeper集群。 1、软件环境 (3台服务器-我的测试) 192.168.30.204 server1 192.168.30.205 server2 192.168.30.206...

qianghong000
2018/06/26
0
0

没有更多内容

加载失败,请刷新页面

加载更多

Linux 运行shell文件,出现 $'\r': command not found

运行编写的shell脚本时,出现了 $'\\r': command not found 这样的错误提示。 报错的原因是我们在windows系统操作时,编辑器里的换行符是\r\n ,而Linux上为\n,两个系统之间有差异导致的。 ...

芥末无敌
今天
7
0
Java数据结构(上)

枚举(Enumeration) 位集合(BitSet) 向量(Vector) 栈(Stack) 1.Enumeration(枚举) boolean hasMoreElements( ):测试是否有更多的元素 Object nextElement( ):如果此枚举对象至少还...

Firefly-
昨天
11
0
vue 跨层组件通讯 provide inject

https://cn.vuejs.org/v2/api/#provide-inject 类型: provide:Object | () => Object inject:Array<string> | { [key: string]: string | Symbol | Object } 详细: provide 和 inject 主......

阿豪boy
昨天
7
0
黑马程序员面试宝典(Java)Beta6.0免费下载

场景 JavaSE基础 面向对象特征以及理解 访问权限修饰符区别 理解clone对象 JavaSE语法 java有没有goto语句 &和&&的区别 如何跳出当前的多重嵌套循环? 是否可以继承String? 重载与重写的区别...

badaoliumang
昨天
9
0
监控linux系统状态

查看系统负载: w/uptime 最后面三个数字表示1分钟,5分钟,15分钟平均有多少个进程占用CPU 占用CPU的进程可以是Running,也可以是Waiting 某一时刻1颗CPU只能有一个进程在使用其资源 #查看c...

asnfuy
昨天
8
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部