消息队列-Kafka(1)

- 日志文件：*.log
- 位移索引文件：*.index
- 时间索引文件：*.timeindex

# 创建Topic 
# --topic 主题名称 避免使用[_]及[.]号
# --replication-factor 副本数量(不能超过broker节点数)
# --partitions 分区数量(并发)
./bin/kafka-topics.sh --create \
--topic UserDataQueue \
--replication-factor 3 \
--partitions 5 \
--bootstrap-server localhost:9092,localhost:9093,localhost:9094

# 查看Topic
./bin/kafka-topics.sh --list \
--bootstrap-server localhost:9092,localhost:9093,localhost:9094

# 修改Topic
# 删除Topic

# 发送消息
# --topic 指定目标Topic
./bin/kafka-console-producer.sh \
--topic UserDataQueue \
--bootstrap-server localhost:9092,localhost:9093,localhost:9094

# 拉取消息
# --from-beginning 从头开始(获取现有的全量数据)
./bin/kafka-console-consumer.sh \
--topic UserDataQueue \
--bootstrap-server localhost:9092,localhost:9093,localhost:9094 \
--from-beginning

# 需要修改的参数

# the directory where the snapshot is stored.
dataDir=/kafka/zkdata
# the port at which the clients will connect
clientPort=2182

# 启动
./bin/zookeeper-server-start.sh -daemon /kafka/zookeeper.properties

# 需修改参数

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=1  # 同一集群内ID必须唯一

# The address the socket server listens on. It will get the value returned from 
# java.net.InetAddress.getCanonicalHostName() if not configured.
#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://localhost:9092  # 同一主机的话，端口号不能相同

# A comma separated list of directories under which to store log files
log.dirs=/kafka/data01  # 日志存储目录，需做隔离

# Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=localhost:2182  # Zookeeper连接地址，参见2.1 zk配置

# Kafka启动

# broker-1
./bin/kafka-server-start.sh -daemon /kafka/server01.properties

# broker-2
./bin/kafka-server-start.sh -daemon /kafka/server02.properties

# broker-3
./bin/kafka-server-start.sh -daemon /kafka/server03.properties

implementation 'org.springframework.kafka:spring-kafka'

spring:
  kafka:
    bootstrap-servers: localhost:9092,localhost:9093,localhost:9094
    producer:
      client-id: kfk-demo
      retries: 3

@RestController
public class IndexController {

    @Autowired
    KafkaTemplate<Object, Object> kafkaTemplate;

    @GetMapping
    public String index() {
        int rdm = new Random().nextInt(1000);
        kafkaTemplate.send("UserDataQueue", new UserData("", rdm));
        return "hello world";
    }

    @GetMapping("index2")
    public String index2() {
		// 发送字符串方式
        kafkaTemplate.send("UserDataTopic", new Gson().toJson(new UserData("apple", 23)));
        return "ok";
    }
}

@Component
@KafkaListener(
        id = "kfk-demo-userdata",
        topics = {"UserDataQueue"},
        groupId = "kfk-demo-group",
        clientIdPrefix = "kfk-demo-client")
public class KfkListener {

    @KafkaHandler
    public void process(@Payload UserData ud,
                        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
        System.out.println(String.format("topic: %s, partition: %d, userData: %s", topic, partition, ud));
    }

    @KafkaHandler(isDefault = true)
    public void process(Object obj) {
        System.out.println(obj);
    }
}

// 接收字符串方式
@Slf4j
@Component
@KafkaListener(id = "kfk-demo2", topics = {"UserDataTopic"})
public class KfkUserDataTopicListener {

    @KafkaHandler
    public void process(String userDataStr) {
        UserData userData = new Gson().fromJson(userDataStr, UserData.class);
        log.info("username: {}, age: {}", userData.getUsername(), userData.getAge());
    }
}

@Configuration
public class KafkaConfig {

    @Bean
    public NewTopic userDataTopic() {
        return new NewTopic("UserDataTopic", 3, (short) 1);
    }
}