Java调用Kafka

矫情吗；* 2022-05-25 07:58 50阅读 0赞

### 依赖 ###

<dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.11</artifactId>
            <version>0.10.2.1</version>
        </dependency>

#### 调试代码 ####

package top.yuyufeng.learn.quene.kafka;
    
    import kafka.admin.AdminUtils;
    import kafka.admin.RackAwareMode;
    import kafka.admin.TopicCommand;
    import kafka.server.ConfigType;
    import kafka.utils.ZkUtils;
    import org.apache.kafka.common.security.JaasUtils;
    import org.junit.Test;
    
    import java.util.Iterator;
    import java.util.Map;
    import java.util.Properties;
    import java.util.Set;
    import java.util.concurrent.ExecutionException;
    
    /**
     * @author yuyufeng
     * @date 2018/5/10.
     */
    public class TopicOperation {
    
        ZkUtils zkUtils = ZkUtils.apply("127.0.0.1:2181", 30000, 30000, JaasUtils.isZkSecurityEnabled());
    
        /**
         * 创建topic
         */
        @Test
        public void createTopic() {
            // 创建一个单分区单副本名为topic-1的topic
            AdminUtils.createTopic(zkUtils, "topic3", 1, 1, new Properties(), RackAwareMode.Enforced$.MODULE$);
            zkUtils.close();
        }
    
        /**
         * 删除topic
         */
        @Test
        public void deleteTopic() {
            // 删除topic 'topic1'
            AdminUtils.deleteTopic(zkUtils, "topic1");
            zkUtils.close();
        }
    
        /**
         * 查询Topic
         */
        @Test
        public void queryTopic() {
            // 获取topic 'test'的topic属性属性
            Properties props = AdminUtils.fetchEntityConfig(zkUtils, ConfigType.Topic(), "topic1");
            // 查询topic-level属性
            Iterator it = props.entrySet().iterator();
            while (it.hasNext()) {
                Map.Entry entry = (Map.Entry) it.next();
                Object key = entry.getKey();
                Object value = entry.getValue();
                System.out.println(key + " = " + value);
            }
            zkUtils.close();
        }
    
    
        @Test
        public void listTopics() throws ExecutionException, InterruptedException {
            String[] options = new String[]{
                    " --list",
                    "--zookeeper",
                    "127.0.0.1:2181"
            };
            TopicCommand.TopicCommandOptions topicCommandOptions = new TopicCommand.TopicCommandOptions(options);
            TopicCommand.listTopics(zkUtils, topicCommandOptions);
        }
    
        /**
         * 修改Topic
         */
        public void modifyTopic() {
            Properties props = AdminUtils.fetchEntityConfig(zkUtils, ConfigType.Topic(), "test");
            // 增加topic级别属性
            props.put("min.cleanable.dirty.ratio", "0.3");
            // 删除topic级别属性
            props.remove("max.message.bytes");
            // 修改topic 'test'的属性
            AdminUtils.changeTopicConfig(zkUtils, "test", props);
            zkUtils.close();
        }
    
    }

### 生产者和消费者 ###

#### 消费者 ####

package top.yuyufeng.learn.quene.kafka;
    
    import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import org.apache.kafka.common.TopicPartition;
    
    import java.util.Arrays;
    import java.util.Collection;
    import java.util.Properties;
    
    /**
     * @author yuyufeng
     * @date 2018/5/10.
     */
    public class MyConsumer {
        public static void main(String[] args) {
            Properties props = new Properties();
            props.put("bootstrap.servers", "192.168.176.129:9092");
            props.put("group.id", "test-consumer-group");
            props.put("enable.auto.commit", "true");
            props.put("auto.commit.interval.ms", "1000");
            props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            final KafkaConsumer<String, String> consumer = new KafkaConsumer<String,String>(props);
            consumer.subscribe(Arrays.asList("topic1"),new ConsumerRebalanceListener() {
                @Override
                public void onPartitionsRevoked(Collection<TopicPartition> collection) {
                }
                @Override
                public void onPartitionsAssigned(Collection<TopicPartition> collection) {
                    //将偏移设置到最开始
                    consumer.seekToBeginning(collection);
                }
            });
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records){
                    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                }
            }
        }
    }

#### 生产者 ####

package top.yuyufeng.learn.quene.kafka;
    
    import kafka.serializer.StringEncoder;
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.Producer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    
    import java.util.Properties;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @author yuyufeng
     * @date 2018/5/10.
     */
    public class MyProducer {
        public static void main(String[] args) {
            Properties props = new Properties();
            props.put("bootstrap.servers", "192.168.176.129:9092");
    
            props.put("acks", "all");
            props.put("retries", 0);
            props.put("batch.size", 16384);
            props.put("linger.ms", 1);
            props.put("buffer.memory", 33554432);
    
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            Producer producer = new KafkaProducer<Integer, String>(props);
            int i = 0;
            while (true) {
                producer.send(new ProducerRecord<String, String>("topic1", "key"+i,"message: " + i++));
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("MyProducer.main");
            }
        }
    
    }