ai-security-xdr/haobang-security-dm/syslog-serve/src/main/java/com/kafka/kafkaProducer.java

package com.kafka;
import com.common.entity.DeviceDevice;
import com.haobang.util.Sm4Util;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.LinkedHashMap;
import java.util.Properties;

import com.haobang.config.AppConfig;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.haobang.util.DeviceInfoUtil;


public class kafkaProducer {

    private static final Logger logger = LoggerFactory.getLogger(kafkaProducer.class);
    private static String strhexKey=AppConfig.getSM4Key();
    public static void main(String[] args) {
       // System.out.println(getFullLogString("syslogmessage"));



        // 配置生产者属性
        Properties props = new Properties();
        //props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.222.130:9092");
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, AppConfig.getKafkaProducerBootstrap());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 可选配置：提高可靠性
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

        // 创建生产者实例
        Producer<String, String> producer = new KafkaProducer<>(props);

        try {
            // 发送10条消息
            //for (int i = 0; i < 1; i++) {
                String key = "key-" + "1";
                String value = "message-" +  "1" + " at " + System.currentTimeMillis();

                // 创建生产者记录
                ProducerRecord<String, String> record =
                        new ProducerRecord<>(AppConfig.getKafkaProducerTopic(), key, value);

                // 发送消息（异步方式）
                producer.send(record, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        if (exception == null) {
                            System.out.println("消息发送成功: " +
                                    "主题=" + metadata.topic() +
                                    ", 分区=" + metadata.partition() +
                                    ", 偏移量=" + metadata.offset());
                                                    } else {
                            System.err.println("消息发送失败: " + exception.getMessage());
                        }
                    }
                });
                // 同步发送方式（如果需要）
                // RecordMetadata metadata = producer.send(record).get();
                // System.out.println("同步发送成功: " + metadata.offset());
           // }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭生产者
            producer.close();
        }
    }

    /**
     *      * syslog 文本消息
     *      * @param strSyslog
     */

    public static void  messagePush(DeviceDevice deviceDevice, String  strSyslog,String strReceiveTime)
    {
        // 配置生产者属性
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, AppConfig.getKafkaProducerBootstrap());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 可选配置：提高可靠性
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

        // 创建生产者实例
        Producer<String, String> producer = new KafkaProducer<>(props);
        try {

                String key = "key-" + System.currentTimeMillis();
                //String value = DeviceInfoUtil.getFullLogString(strSyslog);
                //采用动态获取Syslog 请求的设备信息
                String value = DeviceInfoUtil.getFullLogString(deviceDevice,strSyslog,strReceiveTime);

                //Message进行SM4加密写入kafka
                String Sm4message= Sm4Util.encryptCbc(value, strhexKey);
                System.out.println("Sm4message:"+Sm4message);

                // 创建生产者记录
                ProducerRecord<String, String> record =
                        new ProducerRecord<>(AppConfig.getKafkaProducerTopic(), key, Sm4message);

                // 发送消息（异步方式）
                producer.send(record, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        if (exception == null) {
                            System.out.println("消息发送成功: " +
                                    "主题=" + metadata.topic() +
                                    ", 分区=" + metadata.partition() +
                                    ", 偏移量=" + metadata.offset());
                        } else {
                            System.err.println("消息发送失败: " + exception.getMessage());
                            logger.error("消息发送失败: " + exception.getMessage());
                        }
                    }
                });

                // 同步发送方式（如果需要）
                // RecordMetadata metadata = producer.send(record).get();
                // System.out.println("同步发送成功: " + metadata.offset());

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭生产者
            producer.close();
        }
    }

    /**
     * 生成json字符串，补充设备配置信息
     * @param strSyslog
     * @return
     */
    private  static  String  getLogJsonString(String strSyslog)
    {
        ObjectMapper objectMapper = new ObjectMapper();
        // 创建设备配置信息MAP
        Map<String, Object> map = new LinkedHashMap<>();
        map.put("device_id",AppConfig.getAppServieDeviceId() );
        map.put("device_name", AppConfig.getAppServieDeviceName());
        map.put("vendor", AppConfig.getAppServieVendor());
        map.put("data_type", AppConfig.getAppServieDataType());
        // 添加syslogMessage
        map.put("syslogMessage", strSyslog);
        try {
            //生成json消息字符串
            String json = objectMapper.writeValueAsString(map);
            return json;
        }
        catch(Exception ex)
        {
            logger.error("kafkaProducer getLogJsonString(）生成json字符串 异常: syslog: "+ strSyslog);
            logger.error("Exception: : "+ ex.getMessage());
        }
        return null;
    }




}