Kafka java实例代码

旧城等待， 2022-06-10 08:12 154阅读 0赞

工程准备

1、首先搭建Kafka集群环境，这里我们kafka的集群环境已经搭建好的，详情参考：

[http://blog.csdn.net/liulihui1988/article/details/77099022][http_blog.csdn.net_liulihui1988_article_details_77099022]

2、搭建Maven工程，普通工程需要导入相关的依赖包，想必Maven工程相对简单，如下图：

![这里写图片描述][SouthEast]

*  kafka Maven项目 pom.xml 文件，执行添加一下依赖代码

<dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka_2.10</artifactId>
                <version>0.8.2.0</version>
            </dependency>

## Kafka java实例代码 ##

*  消息生产者代码示例

package com.jxl.cloub.kafka;
    
    import java.util.Date;
    import java.util.Properties;
    import kafka.javaapi.producer.Producer;
    import kafka.producer.KeyedMessage;
    import kafka.producer.ProducerConfig;
    
    /** * 消息生产者代码示例 * */
    public class ProducerDemo { 
    
        public static void main(String[] args) {
            int events=100;
            // 设置配置属性
            Properties props = new Properties();
            props.put("metadata.broker.list","centos-node6:19092,centos-node7:19092,centos-node8:19092");
            props.put("serializer.class", "kafka.serializer.StringEncoder");
            // key.serializer.class默认为serializer.class
            props.put("key.serializer.class", "kafka.serializer.StringEncoder");
            // 可选配置，如果不配置，则使用默认的partitioner
            props.put("partitioner.class", "com.jxl.cloub.kafka.PartitionerDemo");
            // 触发acknowledgement机制，否则是fire and forget，可能会引起数据丢失
            // 值为0,1,-1,可以参考
            props.put("request.required.acks", "1");
            ProducerConfig config = new ProducerConfig(props);
    
            // 创建producer
            Producer<String, String> producer = new Producer<String, String>(config);
            // 产生并发送消息
            long start=System.currentTimeMillis();
            for (long i = 0; i < events; i++) {
                long runtime = new Date().getTime();
                String ip = "10.1.6." + i;//rnd.nextInt(255);
                String msg = runtime + ",www.example.com," + ip;
                //如果topic不存在，则会自动创建，默认replication-factor为1，partitions为0
                KeyedMessage<String, String> data = new KeyedMessage<String, String>(
                        "test", ip, msg);
                producer.send(data);
            }
            System.out.println("耗时:" + (System.currentTimeMillis() - start));
            // 关闭producer
            producer.close();
        }
    
    }

*  消息消费者代码示例

package com.jxl.cloub.kafka;
    
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    import java.util.Properties;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    import kafka.consumer.Consumer;
    import kafka.consumer.ConsumerConfig;
    import kafka.consumer.KafkaStream;
    import kafka.javaapi.consumer.ConsumerConnector;
    
    /** * 消息消费者代码示例 * */
    public class ConsumerDemo { 
    
        private final ConsumerConnector consumer;
        private final String topic;
        private ExecutorService executor;
    
        public ConsumerDemo(String a_zookeeper, String a_groupId, String a_topic) {
            consumer = Consumer.createJavaConsumerConnector(createConsumerConfig(a_zookeeper,a_groupId));
            this.topic = a_topic;
        }
    
        public void shutdown() {
            if (consumer != null)
                consumer.shutdown();
            if (executor != null)
                executor.shutdown();
        }
    
        public void run(int numThreads) {
            Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
            topicCountMap.put(topic, new Integer(numThreads));
            Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer
                    .createMessageStreams(topicCountMap);
            List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);
    
            // now launch all the threads
            executor = Executors.newFixedThreadPool(numThreads);
    
            // now create an object to consume the messages
            //
            int threadNumber = 0;
            for (final KafkaStream stream : streams) {
                executor.submit(new ConsumerMsgTask(stream, threadNumber));
                threadNumber++;
            }
        }
    
        private static ConsumerConfig createConsumerConfig(String a_zookeeper,
                String a_groupId) {
            Properties props = new Properties();
            props.put("zookeeper.connect", a_zookeeper);
            props.put("group.id", a_groupId);
            props.put("zookeeper.session.timeout.ms", "400");
            props.put("zookeeper.sync.time.ms", "200");
            props.put("auto.commit.interval.ms", "1000");
    
            return new ConsumerConfig(props);
        }
    
        public static void main(String[] arg) {
            String[] args = { "centos-node7:2181", "0", "test", "12" };
            String zooKeeper = args[0];
            String groupId = args[1];
            String topic = args[2];
            int threads = Integer.parseInt(args[3]);
    
            ConsumerDemo demo = new ConsumerDemo(zooKeeper, groupId, topic);
            demo.run(threads);
    
            try {
                Thread.sleep(10000);
            } catch (InterruptedException ie) {
    
            }
            demo.shutdown();
        }
    
    }

*  分区Partitioner类示例

package com.jxl.cloub.kafka;
    
    import kafka.producer.Partitioner;
    import kafka.utils.VerifiableProperties;
    
    /** * Partitioner类示例 * */
    public class PartitionerDemo implements Partitioner{ 
    
        public PartitionerDemo(VerifiableProperties props) {}
    
        public int partition(Object obj, int numPartitions) {
            int partition = 0;
            if (obj instanceof String) {
                String key=(String)obj;
                int offset = key.lastIndexOf('.');
                if (offset > 0) {
                    partition = Integer.parseInt(key.substring(offset + 1)) % numPartitions;
                }
            }else{
                partition = obj.toString().length() % numPartitions;
            }
    
            return partition;
        }
    
    }

*  消息处理类

package com.jxl.cloub.kafka;
    
    import kafka.consumer.ConsumerIterator;
    import kafka.consumer.KafkaStream;
    
    /** * *消息处理类 */
    public class ConsumerMsgTask implements Runnable { 
    
        private KafkaStream m_stream;
        private int m_threadNumber;
    
        public ConsumerMsgTask(KafkaStream stream, int threadNumber) {
            m_threadNumber = threadNumber;
            m_stream = stream;
        }
    
        public void run() {
            ConsumerIterator<byte[], byte[]> it = m_stream.iterator();
            while (it.hasNext())
                System.out.println("Thread " + m_threadNumber + ": "
                        + new String(it.next().message()));
            System.out.println("Shutting down Thread: " + m_threadNumber);
        }
    
    }

[http_blog.csdn.net_liulihui1988_article_details_77099022]: http://blog.csdn.net/liulihui1988/article/details/77099022
[SouthEast]: /images/20220610/59d3f8d2dc7b4389ae6032ee2c5f7624.png