kafka的java api示例

不念不忘少年蓝@ 2022-05-08 14:26 112阅读 0赞

kafka的javaAPI

生产者示例：  
创建配置：

1、new Properties()
    	2、添加配置 metadata.broker.list
    				serializer.class = kafka.serializer.StringEncoder
    	3、ProducerConfig()
    	
    	4、创建Producer
    	
    	5、发送

示例代码：

Properties proper = new Properties();
        proper.put("metadata.broker.list","node1:9092,node2:9092,node3:9092");
        proper.put("serializer.class","kafka.serializer.StringEncoder");
    
        ProducerConfig producerConfig = new ProducerConfig(proper);
    
        Producer<String, String> producer = new Producer<String, String>(producerConfig);
    
        for (int i=0;i<100;i++)
        {
            producer.send(new KeyedMessage<String, String>("yuan","test"+i));
        }

消费者示例：

1、指定topic   指定线程数
    	2、new Properties()
    	3、添加配置：zookeeper   group.id    
    				 auto.offset.reset=smallest  相当于--from-beginning
    	4、ConsumerConfig
    	
    	5、创建Consumer连接
    	
    	6、创建消息流并，将1中的值添加
    	
    	7、获取主题中的流数据并输出

示例代码

final static String topic="yuan";
    	final static Integer threadNum=2;
    	public static void main(String[] args) {
    	Properties prop = new Properties();
    	prop.put("zookeeper.connect", "node1:2181,node2:2181,node3:2181");
    	prop.put("group.id", "yuan");
    	prop.put("auto.offset.reset", "smallest");
    	
    	ConsumerConfig consumerConfig = new ConsumerConfig(prop);
    	
    	ConsumerConnector consumerConn = Consumer.createJavaConsumerConnector(consumerConfig);
    	HashMap<String, Integer> map = new HashMap<String, Integer>();
    	map.put(topic,threadNum );
    	Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = consumerConn.createMessageStreams(map);
    	
    	List<KafkaStream<byte[], byte[]>> messageStream = messageStreams.get(topic);
    	
    	for (KafkaStream<byte[], byte[]> kafkaStream : messageStream) {
    		new Thread(new Runnable() {
    			
    			@Override
    			public void run() {
    				// TODO Auto-generated method stub
    				for (MessageAndMetadata<byte[], byte[]> messageAndMetadata : kafkaStream) {
    					String mess = new String(messageAndMetadata.message());
    					System.out.println(mess);
    				}
    			}
    		}).start();
    	}
    	
    }