kafaka

@AComponentScan
public class HelloApp {
  public static void main(String[] args) {
    long start = System.currentTimeMillis();
    TioApplication.run(HelloApp.class, args);
    long end = System.currentTimeMillis();
    System.out.println((end - start) + "ms");
  }
}

package com.litongjava.mq.kafaka;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class KafkaProducerUtils {

  private static KafkaProducer<String, String> producer;
  private static String topic;

  public static void init(String servers, String topicName) {
    topic = topicName;

    Properties props = new Properties();
    props.put("bootstrap.servers", servers);
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

    producer = new KafkaProducer<>(props);
  }

  public static void send(String message) {
    ProducerRecord<String, String> record = new ProducerRecord<>(topic, message);
    producer.send(record);
  }

  public static KafkaProducer<String, String> getProducer() {
    return producer;
  }
}

package com.litongjava.mq.kafaka;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class KafkaConsumerHandler implements Runnable {

  private KafkaConsumer<String, String> consumer;

  public KafkaConsumerHandler(String servers, String groupId, String topic) {
    Properties props = new Properties();
    props.put("bootstrap.servers", servers);
    props.put("group.id", groupId);
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("auto.offset.reset", "earliest");

    consumer = new KafkaConsumer<>(props);
    consumer.subscribe(Collections.singletonList(topic));
  }

  @Override
  public void run() {
    while (true) {
      ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

      for (ConsumerRecord<String, String> record : records) {
        System.out.println("Received message: " + record.value());
      }
    }
  }
}

package com.litongjava.mq.kafaka.config;

import com.litongjava.annotation.AConfiguration;
import com.litongjava.annotation.Initialization;
import com.litongjava.hook.HookCan;
import com.litongjava.mq.kafaka.KafkaConsumerHandler;
import com.litongjava.mq.kafaka.KafkaProducerUtils;
import com.litongjava.tio.utils.environment.EnvUtils;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@AConfiguration
public class KafkaClientConfig {

  @Initialization
  public void init() {
    String servers = EnvUtils.get("kafka.bootstrap.servers");
    String topic = EnvUtils.get("kafka.topic");
    String groupId = EnvUtils.get("kafka.group.id");

    // 初始化生产者
    KafkaProducerUtils.init(servers, topic);
    log.info("Kafka Producer initialized");

    // 启动消费者线程
    KafkaConsumerHandler handler = new KafkaConsumerHandler(servers, groupId, topic);
    Thread consumerThread = new Thread(handler);
    consumerThread.start();

    log.info("Kafka Consumer started");

    // 优雅关闭
    HookCan.me().addDestroyMethod(() -> {
      log.info("Kafka shutting down...");
    });
  }
}

package com.litongjava.mq.kafaka.controller;

import com.litongjava.annotation.RequestPath;
import com.litongjava.mq.kafaka.KafkaProducerUtils;

@RequestPath("/kafaka/test")
public class KafakaTestController {

  @RequestPath()
  public String index() {
    String msg = "Hello Kafka";
    KafkaProducerUtils.send(msg);
    return "Message sent to Kafka";
  }
}

KafkaProducer<String, String> producer = KafkaProducerUtils.getProducer();
ProducerRecord<String, String> record =new ProducerRecord<>("test-topic", "Hello Kafka");
producer.send(record);
System.out.println("Message sent");