java操作kafka
java操作kafka
Zero零_度 发表于2年前
java操作kafka
  • 发表于 2年前
  • 阅读 63
  • 收藏 0
  • 点赞 1
  • 评论 0

新睿云服务器60天免费使用,快来体验!>>>   

maven依赖:

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

生产者:

package com.sniper.kafka.helloworld;

import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;

public class KafkaProducer {
    
    public static void main(String[] args) {
        Properties prop = new Properties();
        //生成者可以不设置zookeeper
        //prop.setProperty("zookeeper.connect", "sniper5:2181");
        prop.setProperty("serializer.class", StringEncoder.class.getName());
        prop.setProperty("metadata.broker.list", "sniper5:9092, sniper6:9092, sniper7:9092");
        
        Producer<Integer, String> producer = new Producer<Integer, String>(new ProducerConfig(prop));
        
        for(int i=0; i<3; i++) {
            KeyedMessage<Integer, String> msg = new KeyedMessage<Integer, String>("test1", "hi-" + i);
            producer.send(msg);
        }
        
        producer.close();
        
    }
    
}

消费者:

package com.sniper.kafka.helloworld;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

public class KafkaConsumer {
    
    public static void main(String[] args) {
        Properties prop = new Properties();
        prop.setProperty("zookeeper.connect", "sniper5:2181,sniper6:2181,sniper7:2181");
        prop.put("zk.connectiontimeout.ms", "1000000");  
        prop.setProperty("group.id", "group1");
        
        ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(prop));
        
        Map<String, Integer> map = new HashMap<String, Integer>();
        map.put("test1", 1);
        
        Map<String, List<KafkaStream<byte[], byte[]>>> msgStreams = consumer.createMessageStreams(map);
        KafkaStream<byte[], byte[]> kafkaStreams = msgStreams.get("test1").get(0);
        ConsumerIterator<byte[], byte[]> iterator = kafkaStreams.iterator();
        while(iterator.hasNext()) {
            String msg = new String(iterator.next().message());
            System.out.println(msg);
        }
    }
    
}

遇到问题:

1、消费端设置zookeeper集群,逗号之间不允许有空格

2、没有主题的情况下,启动消费端会报错,顺序为:创建主题,启动消费端关注主题,生产端生产消息到主题

  • 打赏
  • 点赞
  • 收藏
  • 分享
共有 人打赏支持
粉丝 65
博文 730
码字总数 241581
×
Zero零_度
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: