欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

Kafka入门使用

程序员文章站 2022-06-28 16:57:01
1. 阻塞队列BlockingQueue(接口)解决线程通信的问题;阻塞方法:put、take。生产者消费者模式生产者:产生数据的线程。消费者:使用数据的线程。实现类ArrayBlockingQueue;LinkedBlockingQueue;PriorityBlockingQueue、SynchronousQueue、DelayQueue等。代码实现import java.util.Random;import java.util.concurrent.Array...

1. 阻塞队列

  1. BlockingQueue(接口)

    解决线程通信的问题;

    阻塞方法:put、take。

  2. 生产者消费者模式

    生产者:产生数据的线程。

    消费者:使用数据的线程。

    Kafka入门使用

  3. 实现类

    ArrayBlockingQueue;

    LinkedBlockingQueue;

    PriorityBlockingQueue、SynchronousQueue、DelayQueue等。

  4. 代码实现

    import java.util.Random;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    
    public class BlockingQueueTests {
        public static void main(String[] args) {
            BlockingQueue queue = new ArrayBlockingQueue(10);
            new Thread(new Producer(queue)).start();
            new Thread(new Consumer(queue)).start();
            new Thread(new Consumer(queue)).start();
            new Thread(new Consumer(queue)).start();
        }
    }
    
    class Producer implements Runnable {
    
        private BlockingQueue<Integer> queue;
    
        public Producer(BlockingQueue<Integer> queue) {
            this.queue = queue;
        }
    
        @Override
        public void run() {
            try {
                for(int i = 0; i < 100; i++){
                    Thread.sleep(20);
                    queue.put(i);
                    System.out.println(Thread.currentThread().getName() + "生产:" + queue.size() );
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    
    class Consumer implements Runnable {
    
        private BlockingQueue<Integer> queue;
    
        public Consumer(BlockingQueue<Integer> queue) {
            this.queue = queue;
        }
    
        @Override
        public void run() {
            try {
                while (true) {
                    Thread.sleep(new Random().nextInt(1000));
                    queue.take();
                    System.out.println(Thread.currentThread().getName() + "消费:" + queue.size());
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    

2. Kafka入门

  1. Kafka简介

    Kafka是一个分布式的流媒体平台。

    应用:消息系统、日志收集、用户行为追踪、流式处理。

  2. Kafka特点

    高吞吐量、消息持久化、高可靠性、高扩展性。

  3. Kafka术语

    Broker、Zookeeper

    Topic、Partition、Offset

    Leader Replica、Follower Replica

  4. 下载并配置Kafka

    打开Kafka下载目录下的config的zookeeper.properties,修改里面的dataDir为你想要的位置。

    再打开同为Kafka下载目录下的config的server.properties,修改里面的log,dir(日志存放位置)为你想要的位置。

  5. Windows系统下启动Kafka

    先启动zookeeper,打开命令行,cd到kafka的下载目录,bin\windows\zookeeper-server-start.bat config\zookeeper.properties

    再打开一个命令行,cd到kafka的下载目录,bin\windows\kafka-server-start.bat config\server.properties

    再次打开一个命令行,cd到kafka的下载目录里的bin目录下的windows目录下,kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test即可创建一个名为test的主题,kafka-topics.bat --list --bootstrap-server localhost:9092即可查看所有主题。

    利用生产者发消息:kafka-console-producer.bat --broker-list localhost:9092 --topic test,回车之后即可发消息。

    再次启动一个命令行窗口来查看消费者读消息,依旧cd到kafka的下载目录里的bin目录下的windows目录下,kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic tset --from-beginning,回车之后即可看到生产者发的消息。

3. Spring整合Kafka

  1. 引入依赖

    <dependency>
    			<groupId>org.springframework.kafka</groupId>
    			<artifactId>spring-kafka</artifactId>
    </dependency>
    
  2. 配置Kafka

    配置server、consumer。

    # KafkaProperties
    spring.kafka.bootstrap-servers=localhost:9092
    # 消费者的分组id
    spring.kafka.consumer.group-id=community-consumer-group
    # 是否自动提交
    spring.kafka.consumer.enable-auto-commit=true
    # 自动提交的频率
    spring.kafka.consumer.auto-commit-interval=3000
    
  3. 访问Kafka

    生产者:

    kafkaTemplate.send(topic,data);
    

    消费者:

    @KafkaListener(topics = {"test})
    public void handleMessage(ConsumerRecord record){}
    
  4. 代码实现

    public class KafkaTests {
    
        @Autowired
        private KafkaProducer kafkaProducer;
    
        @Test
        public void testKafka() {
            kafkaProducer.sendMessage("test","你好");
            kafkaProducer.sendMessage("test","在吗");
    
            try {
                Thread.sleep(1000 * 10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    
    @Component
    class KafkaProducer {
    
        @Autowired
        private KafkaTemplate kafkaTemplate;
    
        public void sendMessage(String topic,String content) {
            kafkaTemplate.send(topic,content);
        }
    }
    
    @Component
    class KafkaConsumer {
    
        @KafkaListener(topics = {"test"})
        public void handleMessage(ConsumerRecord record) {
            System.out.println(record.value());
        }
    }
    

本文地址:https://blog.csdn.net/GoodBoyKaiii/article/details/110292614

相关标签: kafka