启动kafka服务
./bin/kafka-server-start.sh config/server.properties
创建话题my-replicated-topic
bin/kafka-topics.sh --create --zookeeper yun1:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic
查看话题列表
bin/kafka-topics.sh --list --zookeeper yun1:2181
生产者写消息:
bin/kafka-console-producer.sh --broker-list yun1:9092 --topic myboys
消费者读消息:
bin/kafka-console-consumer.sh --zookeeper yun1:2181 --from-beginning --topic myboys
查看某个话题的状态信息:
bin/kafka-topics.sh --describe --zookeeper yun1:2181 --topic myboys
生产者的Javademo实现:
import java.util.Properties;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;
public class ProducerDemo {
public static void main(String[] args) throws Exception{
Properties props=new Properties();
props.put("zk.connect","192.168.0.190:2181,192.168.0.191:2181,192.168.0.192:2181");
props.put("metadata.broker.list","192.168.0.190:9092,192.168.0.191:9092,192.168.0.192:9092");
props.put("serializer.class", StringEncoder.class.getName());
ProducerConfig config=new ProducerConfig(props);
Producer producer=new Producer<String,String>(config);
for(int i=1;i<1000;i++){
try {
Thread.sleep(500);
}catch (Exception e){
e.printStackTrace();
}
System.out.println(i);
producer.send(new KeyedMessage<String,String>("myboys","hello "+i+" times"));
}
}
}
问题记录:
- centos7 安装zookeeper集群时,需要关闭防火墙 sudo systemctl stop firewalld && sudo systemctl disable firewalld