安装kafka和zookeeper以及使用

1.安装zookeeper

zookeeper下载：http://zookeeper.apache.org/releases.html
从3.5.5开始，带有bin名称的包才是要下载的包可以直接使用

解压压缩包

进入conf目录

拷贝一份zoo_sample.cfg文件改名为zoo.cfg

更改zoo.cfg配置信息 ：

data和dataLog 数据和数据日志存储路径

dataDir=/apache-zookeeper-3.7.1-bin/data

dataLogDir=/apache-zookeeper-3.7.1-bin/dataLog

进入\bin目录

window: 双击zkServer.cmd启动

linux: zkServer.sh

一个zookeeper图形化工具ZooInspector

下载地址：https://issues.apache.org/jira/secure/attachment/12436620/ZooInspector.zip

直接解压zip进入\zooInspector\build目录jar包启动

执行成功，会弹出client

然后进行连接 zoo默认端口为 2181

可查看zoo节点信息

2.安装kafka
下载：https://kafka.apache.org/downloads.html下载binary版本的

解压安装包

进入config目录下，找到server.properties

编辑 log.dirs为 path+\logs   (path为kafka路径)

Zookeeper地址和端口，Kafka支持内置的Zookeeper和引用外部的Zookeeper

内置zookeeper配置在zookeeper.properties里面默认配置是单机版

zookeeper.connect=localhost:2181,localhost:2182,localhost:2183（默认就是localhost:2181）

在windows系统启动 脚本都在\bin\windows路径下面

启动

.\bin\windows\kafka-server-start.bat .\config\server.properties

创建主题

.\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic nowTest

查看主题命令

.\bin\windows\kafka-topics.bat --list --zookeeper localhost:2181

创建生产者

.\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic nowTest

创建消费者

.\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic nowTest --from-beginning

kafka可视化工具 https://www.kafkatool.com/download.html 下载安装使用即可

3. springboot配置kafka
依赖

  <dependency>

       <groupId>org.springframework.kafka</groupId>

       <artifactId>spring-kafka</artifactId>

   </dependency>

    <!--内嵌的zookeeper和kafka 开发期间，不用准备相关环境了-->

    <dependency>

	    <groupId>org.springframework.kafka</groupId>

	    <artifactId>spring-kafka-test</artifactId>

	    <scope>test</scope>

	</dependency>

kafka配置文件KafkaConfig

    /**

     * @return

     * @Description 主要业务的Kafka生产者配置

     */

    public Map<String, Object> producerConfigs(){

        Map<String, Object> props = new HashMap<>();

        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");

        props.put(ProducerConfig.RETRIES_CONFIG, 0);

        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);

        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);

        props.put(ProducerConfig.ACKS_CONFIG, "1");

        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        return props;

    }

    /**

     * @return

     * @Description 主要业务的Kafka生产者配置工厂

     */

    public ProducerFactory<String, Object> producerFactory(){

        return new DefaultKafkaProducerFactory<>(producerConfigs());

    }

 /**

     * @return

     * @Description 主要业务的Kafka消费者配置

     */

    public Map<String, Object> consumerConfigs(){

        Map<String, Object> propsMap = new HashMap<>();

        propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");

        propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);

        propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);

        propsMap.put(ConsumerConfig.GROUP_ID_CONFIG,"FLOWAC_0_DEV_WXD");

        propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"FALSE");

        propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

        propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        return propsMap;

    }

    /**

     * @return

     * @Description 主要业务的Kafka消费者工厂

     */

    public ConsumerFactory<String, String> consumerFactory(){

        return new DefaultKafkaConsumerFactory<>(consumerConfigs());

    }

    @Bean(name = "kafkaOneListenerFactory")

    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerFactory(){

        ConcurrentKafkaListenerContainerFactory<String, String> factory =

                new ConcurrentKafkaListenerContainerFactory<>();

        factory.setConsumerFactory(consumerFactory());

        factory.getContainerProperties().setPollTimeout(1500);

        return factory;

    }

    @Bean(name = "kafkaOneTemplate")

    public KafkaTemplate<String, Object> kafkaTemplate(){

        return new KafkaTemplate<>(producerFactory());

    }

发送消息

  private KafkaTemplate<String, String> kafkaTemplate;

  String message = String.format("%s","Test");

  kafkaTemplate.send("nowTest", message);

接收消息

@KafkaListener(topics="nowTest",containerFactory = "kafkaOneListenerFactory")

public void kafkaCustomer(String msg){

    log.info("收到消息通知: {}",msg);

}