kafka安装使用

时间：2024-11-10 09:36:38

版本：kafka_2.11-0.10.1.0  (之前安装2.10-0.10.0.0，一直出问题)

安装

  1. 下载并解压代码 
    wget http://mirrors.cnnic.cn/apache/kafka/0.10.0.0/kafka_2.10-0.10.0.0.tgz
    
#http://kafka.apache.org/downloads

tar -zxvf kafka_2.10-0.10.0.0.tgz
    cd kafka_2.-0.10.0.0
  2. 修改每个broker安装目录下的配置文件 
    # $targetID默认是0，每个broker的broker.id必须要唯一
    
broker.id=$targetID

#默认是注释的,$IP改成当前节点的IP即可。如果不改该配置项，在节点通过命令行可以收发消息，而在其他机器是无法通过IP去访问队列的
    
#在之前的版本不叫listeners，而是advertised.host.name和host.name
    
listeners=PLAINTEXT://$IP:9092
  3. 启动服务 
    #kafka依赖于zookeeper
    
#如果没有的话，可以通过kafka提供的脚本快速创建一个单节点zookeeper实例：
    
#bin/zookeeper-server-start.sh config/zookeeper.properties

#确认zookeeper服务已经启动后，启动kafka服务
    
nohup bin/kafka-server-start.sh config/server.properties &
  4. 创建一个名为test，有一份备份，一个分区的topic 
    bin/kafka-topics.sh --create --zookeeper localhost: --replication-factor  --partitions  --topic test
    
#查看所有topic

bin/kafka-topics.sh --list --zookeeper localhost:
  5. 发送消息 
    bin/kafka-console-producer.sh --broker-list localhost: --topic test
  6. 开启一个消费者接收消息 
    bin/kafka-console-consumer.sh --bootstrap-server localhost: --topic test --from-beginning
  7. 查看topic信息 
    bin/kafka-topics.sh --describe --zookeeper localhost: --topic test

Springboot结合Kafka的使用

1.在pom文件添加依赖

<dependency>

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

    <artifactId>spring-kafka</artifactId>

</dependency>

2.在application.properties中添加配置

# APACHE KAFKA (KafkaProperties)

spring.kafka.bootstrap-servers=192.168.0.155:,192.168.0.156:

spring.kafka.client-id=K1
spring.kafka.consumer.auto-offset-reset= earliest

spring.kafka.consumer.enable-auto-commit= true

spring.kafka.consumer.group-id= test-consumer-group
spring.kafka.producer.batch-size=

spring.kafka.producer.bootstrap-servers= 192.168.0.155:,192.168.0.156:

spring.kafka.producer.client-id= P1

spring.kafka.producer.retries=

spring.kafka.template.default-topic= test

3.创建消费者类(订阅消息的对象)

import org.springframework.kafka.annotation.KafkaListener;

import org.springframework.stereotype.Component;

@Component

public class ListenerBean {

    @KafkaListener(topics = "myTopic")

    public void processMessage(String content) {

        System.out.println("you have a new message:" + content);

        // ...

    }

}

4.创建生产者类(发布消息的对象)

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.boot.autoconfigure.EnableAutoConfiguration;

import org.springframework.kafka.core.KafkaTemplate;

import org.springframework.stereotype.Component;

import org.springframework.web.bind.annotation.PathVariable;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RequestMethod;

import org.springframework.web.bind.annotation.RestController;

@Component

@RestController

@RequestMapping("/send")

@EnableAutoConfiguration

public class SendMsgBean {

    private final KafkaTemplate<String,String> kafkaTemplate;

    @Autowired

    public SendMsgBean(KafkaTemplate<String,String> kafkaTemplate) {

        this.kafkaTemplate = kafkaTemplate;

    }

    @RequestMapping(path="/{msg}",method=RequestMethod.GET)

    public String send(@PathVariable("msg") String msg) {

        System.out.println("==sending msg:" + msg);

        kafkaTemplate.send("test","test-"+msg);

        return "message has been sent!";

    }

}

只需这4步，就可以在springboot中使用kafka了，现在我们访问 http://localhost:8080/send/mymessage  就可以在控制台看到信息了。

源码下载

参考：

相关文章