easyfk-mq-kafka Kafka

<dependency>
    <groupId>com.mcst</groupId>
    <artifactId>mq-kafka</artifactId>
</dependency>

dependencies {
    implementation 'com.mcst:mq-kafka'
}

easyfk:
  config:
    mq:
      kafka:
        enable-kafka: true

spring:
  kafka:
    bootstrap-servers: 192.168.1.100:9092,192.168.1.101:9092
    consumer:
      group-id: my-app-group
      auto-offset-reset: earliest
      enable-auto-commit: true
      max-poll-records: 500
      session-timeout-ms: 30000
      heartbeat-interval-ms: 10000
      concurrency: 3
      ack-mode: batch
      isolation-level: read_committed
    producer:
      retries: 3
      batch-size: 16384
      linger-ms: 5
      buffer-memory: 33554432
      acks: all
      enable-idempotence: true
      compression-type: snappy

easyfk:
  config:
    mq:
      kafka:
        enable-kafka: true
        enable-streams: true
        streams:
          application-id: my-streams-app
          state-dir: /tmp/kafka-streams
          processing-guarantee: at_least_once
          commit-interval-ms: 30000
          num-stream-threads: 2
          default-deserialization-exception-handler: logAndContinue
          default-production-exception-handler: fail
          retry-backoff-ms: 100
          reconnect-backoff-ms: 50

@Service
public class OrderService {

    @Resource
    private KafkaProducer kafkaProducer;
}

CommonMessage<OrderDTO> message = new CommonMessage<>();
message.setTopic("order-topic");
message.setData(orderDTO);
message.setMessageKey(orderDTO.getOrderId());
message.setTags("ORDER_CREATED");

kafkaProducer.syncSendMessage(message);

CommonMessage<String> message = new CommonMessage<>();
message.setTopic("notification-topic");
message.setData("用户注册成功");
message.setMessageKey(userId);

kafkaProducer.asyncSendMessage(message);

// 假设已定义 OrderProto.Order Protobuf 消息
OrderProto.Order order = OrderProto.Order.newBuilder()
    .setOrderId("ORD_001")
    .setAmount(9999)
    .build();

ProtobufMessage<OrderProto.Order> message = new ProtobufMessage<>();
message.setTopic("order-protobuf-topic");
message.setData(order);
message.setMessageKey("ORD_001");

kafkaProducer.syncSendProtobufMessage(message);

CommonMessage<String> message = new CommonMessage<>();
message.setTopic("event-topic");
message.setData("event data");
message.setTags("USER_EVENT");

Map<String, String> props = new HashMap<>();
props.put("source", "order-service");
props.put("traceId", "trace_12345");
message.setProperties(props);

kafkaProducer.syncSendMessage(message);

@Component
public class OrderConsumer {

    @KafkaListener(topics = "order-topic", containerFactory = "listenerContainerFactory")
    public void onMessage(ConsumerRecord<String, String> record) {
        OrderDTO order = JSONObject.parseObject(record.value(), OrderDTO.class);
        // 处理订单
    }
}

@Component
public class OrderProtobufConsumer {

    @KafkaListener(topics = "order-protobuf-topic", containerFactory = "protobufKafkaListenerContainerFactory")
    public void onMessage(ConsumerRecord<String, byte[]> record) {
        OrderProto.Order order = ProtobufUtils.deserialize(record.value(), OrderProto.Order.class);
        // 处理订单
    }
}

@Resource
private KafkaMessageUtil kafkaMessageUtil;

@KafkaListener(topics = "order-topic", containerFactory = "listenerContainerFactory")
public void onMessage(ConsumerRecord<String, String> record) {
    // 提取消息体
    OrderDTO order = kafkaMessageUtil.getMessageBodyFromStringRecord(record, OrderDTO.class);

    // 提取 Header 信息
    Map<String, String> headers = kafkaMessageUtil.extractHeaders(record);
    String messageId = headers.get("messageId");
    String tags = headers.get("tags");
}

@Component
public class OrderStreamProcessor extends BaseStreamProcessor {

    @Override
    protected String getProcessorName() {
        return "order-stream-processor";
    }

    @Override
    protected String[] getInputTopics() {
        return new String[]{"order-topic"};
    }

    @Override
    protected String[] getOutputTopics() {
        return new String[]{"order-result-topic"};
    }

    @Override
    protected void buildTopology(StreamsBuilder builder) {
        builder.<String, String>stream("order-topic")
            .filter((key, value) -> value != null && !value.isEmpty())
            .mapValues(value -> {
                // 处理逻辑
                return processOrder(value);
            })
            .to("order-result-topic");
    }
}

@Override
protected void buildTopology(StreamsBuilder builder) {
    builder.stream("input-topic", Consumed.with(SerdeFactory.string(), SerdeFactory.byteArray()))
        .mapValues(value -> {
            // 处理逻辑
            return transform(value);
        })
        .to("output-topic", Produced.with(SerdeFactory.string(), SerdeFactory.byteArray()));
}

@Resource
private StreamProcessorRegistry registry;

// 获取所有处理器名称
List<String> names = registry.getAllProcessorNames();

// 获取处理器数量
int count = registry.getProcessorCount();

// 获取所有处理器状态
List<ProcessorStatus> statuses = registry.getAllProcessorStatus();

// 获取统计信息
ProcessorStatistics stats = registry.getStatistics();

com.mcst.easyfk.mq.kafka
├── config
│   ├── KafkaMqConfig.java                 # 核心自动配置（生产者/消费者工厂）
│   └── KafkaStreamsConfig.java            # Kafka Streams 自动配置
├── producer
│   └── KafkaProducer.java                 # 消息生产者（同步/异步 + JSON/Protobuf）
├── properties
│   └── KafkaMqProperties.java             # 配置属性类
├── streams
│   ├── processor
│   │   ├── BaseStreamProcessor.java       # 流处理器抽象基类
│   │   └── StreamProcessorRegistry.java   # 处理器注册管理器
│   └── serde
│       ├── BaseMessageSerde.java          # BaseMessage 序列化器
│       ├── ProtobufMessageSerde.java      # Protobuf 消息序列化器
│       └── SerdeFactory.java              # Serde 工厂类
└── util
    └── KafkaMessageUtil.java              # 消息处理工具类