Kafka入门

Kafka 安装及快速入门

1. 安装JDK

• 安装JAVA环境,这里不赘述,注意配置JAVA_HOME并确保安装成功

2.下载kafka

• 下载地址：http://kafka.apache.org/downloads
  

选择红框中的版本，Kafka包名组成： Scala版本 - Kafka自身版本

下官方推荐的版本

解压之后

bin是可执行文件

config是配置文件

3. 修改配置文件

新建logs文件夹

修改server.properties

kafka2.7.0/config/

log.dirs=C:/kafka/kafka2.7.0/logs

4. 启动 zookeeper

新建cmd窗口，启动zookeeper server

bin\windows\zookeeper-server-start.bat config\zookeeper.properties
1

运行成功,2181端口,不要关闭窗口

5. 启动kafka

新建cmd窗口，启动kafka server

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

运行成功,不要关闭窗口

6. 创建主题

新建cmd窗口，创建主题

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

运行成功,不要关闭窗口

查看主题列表

kafka-topics.bat --list --zookeeper localhost:2181
1

7. 启动kafka producer

新建cmd窗口，启动kafka producer

kafka-console-producer.bat --broker-list localhost:9092 --topic  kafka
1

运行成功,不要关闭窗口

8. 启动kafka consumer

新建cmd窗口，启动kafka consumer

kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic kafka
1

运行成功,不要关闭窗口

9. 测试

kafka producer 发送消息

kafka consumer接收消息

SpringBoot 整合Kafka

创建项目

项目整体架构：

1. pom 文件代码如下：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.1.9.RELEASE</version>
		<relativePath/> <!-- lookup parent from repository -->
	</parent>
	<groupId>com.example</groupId>
	<artifactId>kafka-demo</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<name>kafka-demo</name>
	<description>Demo project for Spring Boot</description>
	<properties>
		<java.version>1.8</java.version>
	</properties>
	<dependencies>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-web</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter</artifactId>
		</dependency>

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>org.springframework.kafka</groupId>
			<artifactId>spring-kafka</artifactId>
		</dependency>
		<dependency>
			<groupId>org.projectlombok</groupId>
			<artifactId>lombok</artifactId>
			<optional>true</optional>
		</dependency>

		<!-- 添加 gson 依赖 -->
		<dependency>
			<groupId>com.google.code.gson</groupId>
			<artifactId>gson</artifactId>
			<version>2.8.5</version>
		</dependency>

		<dependency>
			<groupId>org.slf4j</groupId>
			<artifactId>slf4j-log4j12</artifactId>
			<version>1.7.25</version>
		</dependency>
	</dependencies>

	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
		</plugins>
	</build>

</project>
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667

主要引入了 spring-kafka 、lombok 、 gson 依赖。

2. 用户实体类

@Data
@NoArgsConstructor
@AllArgsConstructor
public class User {
    private int id;
    private String nickName;
    private String telNumber;
    private String password;
    private String sex;
}
12345678910

3. application.yml

 server:
   port: 8888
 spring:
   main:
     banner-mode: 'OFF'
   kafka:
     bootstrap-servers: 127.0.0.1:9092 #指定kafka server的地址，集群配多个，中间，逗号隔开
     producer:
       key-serializer: org.apache.kafka.common.serialization.StringSerializer
       value-serializer: org.apache.kafka.common.serialization.StringSerializer
     consumer:
       group-id: default_consumer_group #群组ID
       enable-auto-commit: true
       auto-commit-interval: 1000
       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
12345678910111213141516

4. KafkaController 类

@RestController
public class KafkaController {
    @Resource
    KafkaTemplate<String,Object> kafkaTemplate;

    @PostMapping("/postMessage")
    public ResponseEntity sendMessage(){
      User user=new User();
      user.setId(1);
      user.setNickName("深蓝梦境");
      user.setTelNumber("12929019282");
      user.setPassword("123456");
      user.setSex("女");
      GsonBuilder builder=new GsonBuilder();
      String users=builder.create().toJson(user);
      kafkaTemplate.send("kafka", users);
      return ResponseEntity.ok("发送成功");
    }
}
12345678910111213141516171819

5. FirstConsumer类

@Component
public class FirstConsumer {
    /**
     * 定义此消费者接收topics = "kafka"的消息，与controller中的topic对应上即可
     * @param record 变量代表消息本身，可以通过ConsumerRecord类型的record变量来打印接收的消息的各种信息
     */
    @KafkaListener(topics="kafka")
    public void userListener(ConsumerRecord<?,?> record){
        System.out.println(record.topic());
        System.out.println(record.offset());
        System.out.println(record.value());
    }
}
12345678910111213

6. 测试

启动SpringBootApplication

使用postman访问接口并发送消息

consumer接收消息