kafka的数据发送和接收java_API

小咪咪 2022-07-14 03:00 226阅读 0赞

**往消息队列里面发送数据**

import kafka.javaapi.producer.Producer;
    import kafka.producer.KeyedMessage;
    import kafka.producer.ProducerConfig;
    
    import java.io.IOException;
    import java.util.Properties;
    
    import scala.math.Numeric.IntIsIntegral;
    
    /** *kafka 消息发送 * */
    public class SendDatatToKafka { 
        public static void main(String[] args) {
            SendDatatToKafka s = new SendDatatToKafka();
            try {
                s.send("test", "jack", "rose");
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    
        public void send(String topic, String key, String data) throws IOException {
            Properties props = new Properties();
            props.put("metadata.broker.list", "slave1:9092");
            props.put("serializer.class", "kafka.serializer.StringEncoder");
            // key.serializer.class默认为serializer.class
            props.put("key.serializer.class", "kafka.serializer.StringEncoder");
            props.put("request.required.acks", "1");
            ProducerConfig config = new ProducerConfig(props);
    
            Producer<String, String> producer = new Producer<String, String>(config);
            for (int i = 0; i < 1000; i++) {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
                producer.send(new KeyedMessage<String, String>(topic, key, data + i));
    
            }
    
            producer.close();
        }
    }

**消息队列接收消息**

/** *接收消息队列 */
    import java.io.IOException;
    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 GetDataFromKafka implements Runnable { 
    
        private String topic;
        private String path;
        public GetDataFromKafka(String topic,String path) {
            this.path=path;
            this.topic = topic;
        }
        public static void main(String[] args) {
            GetDataFromKafka gdkast=new GetDataFromKafka("test", "d:\\clusterMonitor.rrd");
            new Thread(gdkast).start();
        }   
        @Override
        public void run() {
            System.out.println("start runing consumer");
    
            Properties properties = new Properties();
            properties.put("zookeeper.connect",
                    "192.168.10.219:2181");// 声明zk
            properties.put("group.id", "CMMtest");// 必须要使用别的组名称， // 如果生产者和消费者都在同一组，则不能访问同一组内的topic数据
            properties.put("auto.offset.reset", "largest"); 
            ConsumerConnector consumer = Consumer
                    .createJavaConsumerConnector(new ConsumerConfig(properties));
            // TODO Auto-generated method stub
    
            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[]> stream = messageStreams.get(topic).get(0);// 获取每次接收到的这个数据
            ConsumerIterator<byte[], byte[]> iterator = stream.iterator();
            while (iterator.hasNext()) {
                String message = new String(iterator.next().message());
                //hostName+";"+ip+";"+commandName+";"+res+";"+System.currentTimeMillis();
                //这里指的注意，如果没有下面这个语句的执行很有可能回从头来读消息的
                consumer.commitOffsets();
                System.out.println(message);
            }
        }
    }