3、kafka的Java代码操作

梦里梦外; 2022-08-01 06:15 141阅读 0赞

maven依赖地址：http://www.mvnrepository.com/artifact/org.apache.kafka/kafka\_2.10/0.8.2.0  
maven依赖：

<span style="white-space:pre">	</span><dependencies>
    		<dependency>
    			<groupId>org.apache.kafka</groupId>
    			<artifactId>kafka_2.10</artifactId>
    			<version>0.8.2.0</version>
    		</dependency>
    	</dependencies>

生产者代码：

package kafka;
    
    import java.util.Properties;
    import kafka.javaapi.producer.Producer;
    import kafka.producer.KeyedMessage;
    import kafka.producer.ProducerConfig;
    import kafka.serializer.StringEncoder;
    
    public class Kafkaproducer extends Thread {
    	private String topic;
    	
    	public Kafkaproducer(String topic){
    		super();
    		this.topic=topic;
    	}
    	
    	@Override
    	public void run() {
    		Producer<Integer, String> producer=CreateProducer();
    		for (int i = 1; i < 10; i++) {
    			String message="message"+i;
    			producer.send(new KeyedMessage<Integer, String>(topic, message));
    			System.out.println("发送："+message);
    			try {
    				sleep(1000);
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    	
    	public Producer<Integer, String> CreateProducer(){
    		Properties props=new Properties();
    		props.setProperty("zookeeper.connect", "192.168.1.200:2181");
    		props.setProperty("serializer.class", StringEncoder.class.getName());
    		props.setProperty("metadata.broker.list", "192.168.1.200:9092");
    		Producer<Integer, String> producer = new Producer<Integer, String>(new ProducerConfig(props));
    		return producer;
    	}
    	
    	public static void main(String[] args) {
    		new Kafkaproducer("test").start();
    	}
    	
    }

eclipse的console上显示的信息：

log4j:WARN No appenders could be found for logger (kafka.utils.VerifiableProperties).
    log4j:WARN Please initialize the log4j system properly.
    log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
    发送：message1
    发送：message2
    发送：message3
    发送：message4
    发送：message5
    发送：message6
    发送：message7
    发送：message8
    发送：message9

在终端上用命令启动消费者：  
\[root@baozi kafka\_2.10-0.8.2.1\]\# bin/kafka-console-consumer.sh --zookeeper 192.168.1.200:2181 --topic test --from-beginning

终端上显示的信息：
    message1
    message2
    message3
    message4
    message5
    message6
    message7
    message8
    message9

==================================================================================================================================  
消费者代码：

package kafka;
    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 extends Thread{
    	private String topic;
    	
    	private KafkaConsumer(String topic) {
    		super();
    		this.topic=topic;
    	}
    	
    	@Override
    	public void run() {
    		ConsumerConnector consumer = createConsumer();
    		Map<String,Integer> topicCountMap=new HashMap<String, Integer>();
    		topicCountMap.put(topic, 1);
    		
    		Map<String, List<KafkaStream<byte[], byte[]>>> MessageStreams = consumer.createMessageStreams(topicCountMap);
    		KafkaStream<byte[], byte[]> kafkaStream = MessageStreams.get(topic).get(0);
    		ConsumerIterator<byte[], byte[]> iterator = kafkaStream.iterator();
    		while (iterator.hasNext()) {
    			String message = new String(iterator.next().message());
    			System.out.println("接收到："+message);
    		}
    	}
    	
    	public ConsumerConnector createConsumer(){
    		Properties properties = new Properties();
    		properties.setProperty("zookeeper.connect", "192.168.1.200:2181");
    		properties.setProperty("group.id", "group1");
    		ConsumerConnector createJavaConsumerConnector = Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));
    		return createJavaConsumerConnector;
    	}
    	
    	public static void main(String[] args) {
    		new KafkaConsumer("test").start();
    	}
    }

先启动消费者，再启动生产者类：  
生产者console输出内容：

消费者console输出内容：

log4j:WARN No appenders could be found for logger (kafka.utils.VerifiableProperties).
    log4j:WARN Please initialize the log4j system properly.
    log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
    接收到：message1
    接收到：message2
    接收到：message3
    接收到：message4
    接收到：message5
    接收到：message6
    接收到：message7
    接收到：message8
    接收到：message9

消费者终端同样会收到并输出：

message1
    message2
    message3
    message4
    message5
    message6
    message7
    message8
    message9