kafka
Consumer:
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.tencent.kafka;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Collections;
import java.util.Properties;
import java.util.Random;
public class KafkaConsumerDemo implements Runnable{
private final KafkaConsumer<String, String> consumer;
private final String topic;
public KafkaConsumerDemo(String topic, String group,String brokerList, String secureId, String secureKey) {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
props.put(ConsumerConfig.GROUP_ID_CONFIG, group);
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
confForAuthentication(props, secureId, secureKey);
consumer = new KafkaConsumer<String,String>(props);
this.topic = topic;
}
private void confForAuthentication(Properties props,String secureId,String secureKey) {
// 设置认证参数
props.put(TbdsAuthenticationUtil.KAFKA_SECURITY_PROTOCOL, TbdsAuthenticationUtil.KAFKA_SECURITY_PROTOCOL_AVLUE);
props.put(TbdsAuthenticationUtil.KAFKA_SASL_MECHANISM, TbdsAuthenticationUtil.KAFKA_SASL_MECHANISM_VALUE);
props.put(TbdsAuthenticationUtil.KAFKA_SASL_TBDS_SECURE_ID,secureId);
props.put(TbdsAuthenticationUtil.KAFKA_SASL_TBDS_SECURE_KEY,secureKey);
}
public void run() {
while(true){
doWork();
try {
Thread.sleep(1000);
} catch (Exception e) {
}
}
}
public void doWork() {
consumer.subscribe(Collections.singletonList(this.topic));
ConsumerRecords<String, String> records = consumer.poll(1000);
for (ConsumerRecord<String, String> record : records) {
System.out.println("Received message: (" + record.key() + ", " + record.value() + ") at offset " + record.offset());
}
}
public static void main(String[] args) {
if (args == null || args.length != 5) {
System.out.println("Usage: topic group brokerlist[host1:port,host2:port,host3:port] secureId secureKey");
System.exit(1);
}
KafkaConsumerDemo kafkaConsumerDemo = new KafkaConsumerDemo(args[0], args[1],args[2],args[3],args[4]);
Thread thread = new Thread(kafkaConsumerDemo);
thread.start();
}
}Producer
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.tencent.kafka;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
//import com.tencent.tbds.sdk.core.util.ServicesUtil;
//import com.tencent.tbds.sdk.core.util.TbdsAuthenticationUtil;
import java.util.Properties;
import java.util.Random;
public class KafkaProducerDemo extends Thread {
public static long count=0l;
public static Object lock = 0 ;
public static long startTime =System.currentTimeMillis();
public static long lastKilloTime = System.currentTimeMillis();
private final KafkaProducer<Long, String> producer;
private final String topic;
private final Boolean isAsync;
private static int messageCount = 1000;
public KafkaProducerDemo(String topic, Boolean isAsync,String brokerList, String secureId, String secureKey) {
Properties props = new Properties();
props.put("client.id", "DemoProducer");
props.put("key.serializer", "org.apache.kafka.common.serialization.LongSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("bootstrap.servers", brokerList);
props.put("batch.num.messages",5000);//每个批次的消息数量
props.put("queue.buffering.max.messages", 300000);//缓存最多缓存的消息数量
props.put("linger.ms", "1000");
props.put("block.on.buffer.full", true);//缓存满之后阻塞producer
props.put("queue.buffering.max.ms", 1000000);//缓存最多缓存的时间
props.put("batch.size", 500000);//批次发送大小，字节
props.put("buffer.memory", 7000000);//缓存大小（字节）
props.put("metadata.fetch.timeout.ms", 12000);//首次获取metadata信息的超时时间
props.put("metadata.max.age.ms", 1000000);//metadata过期的时间
props.put("request.timeout.ms", 1000000);//请求超时时间，异步发送时某个批次在缓冲区中存在时间超过这个时间就会报metadata超时
confForAuthentication(props, secureId, secureKey);
producer = new KafkaProducer<Long, String>(props);
this.topic = topic;
this.isAsync = isAsync;
}
private void confForAuthentication(Properties props,String secureId,String secureKey) {
//测试TBDS
props.put(TbdsAuthenticationUtil.KAFKA_SECURITY_PROTOCOL, TbdsAuthenticationUtil.KAFKA_SECURITY_PROTOCOL_AVLUE);
props.put(TbdsAuthenticationUtil.KAFKA_SASL_MECHANISM, TbdsAuthenticationUtil.KAFKA_SASL_MECHANISM_VALUE);
props.put(TbdsAuthenticationUtil.KAFKA_SASL_TBDS_SECURE_ID,secureId);
props.put(TbdsAuthenticationUtil.KAFKA_SASL_TBDS_SECURE_KEY,secureKey);
//测试PLAIN
// props.put(TbdsAuthenticationUtil.KAFKA_SECURITY_PROTOCOL,"SASL_PLAINTEXT");
// props.put(TbdsAuthenticationUtil.KAFKA_SASL_MECHANISM, "PLAIN");
//测试PLAINTEXT
//props.put(TbdsAuthenticationUtil.KAFKA_SECURITY_PROTOCOL,"PLAINTEXT");
}
public void run() {
long messageNo = 1;
while (messageNo<KafkaProducerDemo.messageCount) {
// String messageStr = java.util.UUID.randomUUID().toString() + "--" + messageNo;
String messageStr = "{\"metrics\":[{\"timestamp\":0,\"type\":\"Long\",\"metricname\":\"rpcdetailed.rpcdetailed.IOExceptionNumOps\",\"appid\":\"journalnode\",\"hostname\":\"tbds-10-254-97-135\",\"starttime\":1512531389656,\"metrics\":{\"1512531389656\":0.0,\"1512531399657\":0.0,\"1512531409656\":0.0,\"1512531419656\":0.0,\"1512531429657\":0.0,\"1512531439656\":0.0,\"1512531449656\":0.0}},{\"timestamp\":0,\"type\":\"Double\",\"metricname\":\"rpcdetailed.rpcdetailed.IOExceptionAvgTime\",\"appid\":\"journalnode\",\"hostname\":\"tbds-10-254-97-135\",\"starttime\":1512531389656,\"metrics\":{\"1512531389656\":0.0,\"1512531399657\":0.0,\"1512531409656\":0.0,\"1512531419656\":0.0,\"1512531429657\":0.0,\"1512531439656\":0.0,\"1512531449656\":0.0}}]}";
long startTime = System.currentTimeMillis();
if (isAsync) { // Send asynchronously
producer.send(new ProducerRecord<Long, String>(topic,
messageNo,
messageStr), new DemoCallBack(startTime, messageNo, messageStr));
} else { // Send synchronously
try {
producer.send(new ProducerRecord<Long, String>(topic,
messageNo,
messageStr)).get();
System.out.println("thread "+Thread.currentThread().getName() +" Sent message: (" + messageNo + ", " + messageStr + ")");
} catch (Exception e) {
e.printStackTrace();
}
}
++messageNo;
}
}
public static void main(String[] args) {
if (args == null || args.length != 6) {
System.out.println("Usage: topic brokerlist[host1:port,host2:port,host3:port] secureId secureKey messageCount producerInstance");
System.exit(1);
}
KafkaProducerDemo.messageCount = Integer.parseInt( args[4] );
int producerCount = Integer.parseInt( args[5] );
for( int i = 0 ; i < producerCount ; i ++ ) {
KafkaProducerDemo kafkaProducerDemo = new KafkaProducerDemo(args[0], true, args[1], args[2], args[3]);
kafkaProducerDemo.start();
}
}
}
class DemoCallBack implements Callback {
private final long startTime;
private final long key;
private final String message;
public DemoCallBack(long startTime, long key, String message) {
this.startTime = startTime;
this.key = key;
this.message = message;
}
/**
* A callback method the user can implement to provide asynchronous handling of request completion. This method will
* be called when the record sent to the server has been acknowledged. Exactly one of the arguments will be
* non-null.
*
* @param metadata The metadata for the record that was sent (i.e. the partition and offset). Null if an error
* occurred.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
public void onCompletion(RecordMetadata metadata, Exception exception) {
long elapsedTime = System.currentTimeMillis() - startTime;
if (metadata != null) {
synchronized (KafkaProducerDemo.lock){
KafkaProducerDemo.count ++;
if( KafkaProducerDemo.count % 10000 == 0 ){
long curTime = System.currentTimeMillis();
System.out.println( " produced count: "+KafkaProducerDemo.count+" speed avg : " +
(KafkaProducerDemo.count*1000 / ( curTime - KafkaProducerDemo.startTime) ) +" "+
(10000000/(1+curTime - KafkaProducerDemo.lastKilloTime)) );
KafkaProducerDemo.lastKilloTime = curTime;
if( KafkaProducerDemo.count % 100000 == 0 )
System.out.println(
Thread.currentThread().getName() + " send message(" + key + ", " + message + ") sent to partition(" + metadata.partition() +
"), " +
"offset(" + metadata.offset() + ") in " + elapsedTime + " ms");
}
}
} else {
exception.printStackTrace();
}
}
}Last updated