我是Kafka的新手,今天我尝试创建Java Producer,以在不同分区上产生有关Kafka主题的消息。
首先,我创建了一个程序包,raggieKafka
在该程序包下创建了2个类:TestProducer
和SimplePartitioner
。
TestProducer类具有以下代码:
package raggieKafka;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.*;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
public class TestProducer{
public static void main(String args[]) throws Exception
{
long events = 0;
BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
events = Integer.parseInt(reader.readLine());
Random rnd = new Random();
Properties props = new Properties();
props.put("metadata.broker.list", "localhost:9092");
props.put("topic.metadata.refresh.interval.ms", "1");
props.put("serializer.class", "kafka.serializer.StringEncoder");
props.put("partitioner.class", "raggieKafka.SimplePartitioner");
props.put("request.required.acks", "1");
ProducerConfig config = new ProducerConfig(props);
Producer<String, String> prod = new Producer<String, String>(config);
for(long i = 0; i < events; i++)
{
long runtime = new Date().getTime();
String ip = "192.168.2." + rnd.nextInt(255);
String msg = runtime + ",www.example.com, " + ip;
KeyedMessage<String,String> data = new KeyedMessage<String, String>("page_visits", ip, msg);
prod.send(data);
}
prod.close();
}
}
并且SimplePartitioner类具有以下代码:
package raggieKafka;
import kafka.producer.Partitioner;
import kafka.utils.VerifiableProperties;
public class SimplePartitioner implements Partitioner{
public SimplePartitioner(VerifiableProperties props)
{
}
public int partition(Object Key, int a_numPartitions)
{
int partition = 0;
String stringKey = (String) Key;
int offset = stringKey.indexOf(stringKey);
if(offset > 0)
{
partition = Integer.parseInt(stringKey.substring(offset+1)) % a_numPartitions;
}
return partition;
}
}
在编译这些Java程序之前,我在Kafka Broker上创建了主题:
C:\kafka_2.11-0.9.0.1>.\bin\windows\kafka-topics.bat --create --topic page_visit
s --zookeeper localhost:2181 --partitions 5 --replication-factor 1
WARNING: Due to limitations in metric names, topics with a period ('.') or under
score ('_') could collide. To avoid issues it is best to use either, but not bot
h.
Created topic "page_visits".
现在,当我编译Java程序时,它将所有消息仅放入1个分区,即page_visits-0,在该分区下发布所有消息,而其余所有其他分区保持为空。
有人可以告诉我为什么我的Java生产者不将我所有的消息分发到其他分区吗?
实际上,我查看了google,然后又添加了一个属性:
props.put("topic.metadata.refresh.interval.ms", "1");
但Producer仍未针对所有主题生成消息。
请帮忙。
您的SimplePartitioner代码在以下行中有bug
int offset = stringKey.indexOf(stringKey);
它总是返回,0
因此您的偏移量始终等于,0
并且它永远不会大于0,因此if块将不会执行。最后,它总是返回您的partition 0
。
解决方案:由于您的密钥是IP地址,因此以下更改可能会按预期进行。
int offset = stringKey.lastIndexOf('.');
希望这可以帮助!
本文收集自互联网,转载请注明来源。
如有侵权,请联系[email protected] 删除。
我来说两句