producer consumer java代码

小鱼儿 2022-06-14 00:36 153阅读 0赞

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 ConsumerTest {
    
    	public ConsumerConnector createConsumer(){
    		Properties properties = new Properties();
    		//zookeeper配置
    		properties.setProperty("zookeeper.connect", "192.168.10.250:2181");
    		properties.setProperty("zookeeper.session.timeout.ms", "4000");  
    		properties.setProperty("zookeeper.sync.time.ms", "200");  
    		//consumer所在的组
    		properties.setProperty("group.id", "test_consumer_group");
            properties.setProperty("auto.commit.interval.ms", "1000");  
            properties.setProperty("auto.offset.reset", "smallest");    
            properties.setProperty("serializer.class", "kafka.serializer.StringEncoder"); 
            
    		return Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));
    	}
    	
    	public void consumer(){
    		//获取连接器
    		ConsumerConnector connector = createConsumer();
    		
    		HashMap<String, Integer> topics = new HashMap<String,Integer>();
    		//设置topic和partition
    		topics.put("test", 1);
    		//获取消息
    		Map<String, List<KafkaStream<byte[], byte[]>>>  messages = connector.createMessageStreams(topics);
    		//获取topic中某个partition中的数据
    		KafkaStream<byte[], byte[]> kafkaStream = messages.get("test").get(0);
    		//遍历获取
    		ConsumerIterator<byte[], byte[]> iterator =  kafkaStream.iterator();
    		while(iterator.hasNext()){
    			byte[] bytes = iterator.next().message();
    			System.out.println("consumer message is "+new String(bytes));
    		}
    		
    	}
    	
    	public static void main(String[] args) {
    		new ConsumerTest().consumer();
    	}
    }

package kafka;
    
    import java.util.Properties;
    
    import kafka.javaapi.producer.Producer;
    import kafka.producer.KeyedMessage;
    import kafka.producer.ProducerConfig;
    
    
    
    public class ProducerTest {
    	
    	public Producer<String,String> createProducer(){
    		Properties properties = new Properties();
    		properties.setProperty("zookeeper.connect", "192.168.10.250:2181");
    		properties.setProperty("serializer.class", "kafka.serializer.StringEncoder"); 
    		properties.setProperty("metadata.broker.list", "192.168.10.250:9092");
    		
    		return new Producer<String,String>(new ProducerConfig(properties));
    	}
    	
    	public void producer(){
    		Producer<String,String> producer = createProducer();
    		for(int i=0;i<100;i++){
    			producer.send(new KeyedMessage<String, String>("test", "producer message "+i));
    		}
    	}
    	
    	public static void main(String[] args) {
    		new ProducerTest().producer();
    	}
    }