kafkajavaAPI入库程序的实现方法

这篇文章主要讲解了“kafka javaAPI入库程序的实现方法”，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习“kafka javaAPI入库程序的实现方法”吧！

十年的金堂县网站建设经验，针对设计、前端、开发、售后、文案、推广等六对一服务，响应快，48小时及时工作处理。网络营销推广的优势是能够根据用户设备显示端的尺寸不同，自动调整金堂县建站的显示方式，使网站能够适用不同显示终端，在浏览器中调整网站的宽度，无论在任何一种浏览器上浏览网站，都能展现优雅布局与设计，从而大程度地提升浏览体验。创新互联公司从事“金堂县网站设计”,“金堂县网站推广”以来，每个客户项目都认真落实执行。

讲解

maven导包

  
    org.apache.kafka  
    kafka-clients  
    2.3.0  

连接kafka

Properties props = new Properties();
props.put("acks", "all");  //保证所有副本接受到消息
props.put("bootstrap.servers", Config.ipList);  //可设置多个
props.put("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
props.put("retries", "2");
KafkaProducer produce= new KafkaProducer(props);

kerberos认证

  • kerberos是大数据平台的安全认证策略，可在项目启动时先一步完成。这里介绍两种实现方式。

方式一

  • 指定认证文件

//加载keberos配置文件
System.setProperty("java.security.krb5.conf", "/etc/krb5.conf"); 
//加载kerberos用户文件
System.setProperty("java.security.auth.login.config", "/etc/kafka/conf/kafka_jaas.conf");

方式二

  • 某些时候，考虑到用户切换，不同机器，有不同的用户信息，每个都要通过配置文件设置，比较麻烦，考虑使用java的启动的临时文件功能（主要是炫技——微笑）。

//加载keberos配置文件
System.setProperty("java.security.krb5.conf", "/etc/krb5.conf");
KafkaUtil.configureJAAS(Config.tabFile, Config.principal);  //用户和认证文件

/**
 * 生成jaas.conf临时文件
 * @param keyTab  tab认证文件位置
 * @param principal 认证用户
 */
public static void configureJAAS(String keyTab, String principal) {
  String JAAS_TEMPLATE =
            "KafkaClient {\n"
            + "com.sun.security.auth.module.Krb5LoginModule required\n" +
              "useKeyTab=true\n" +
              "keyTab=\"%1$s\"\n" +
              "principal=\"%2$s\";\n"
            + "};";
      String content = String.format(JAAS_TEMPLATE, keyTab, principal);
    
      File jaasConf = null;
      PrintWriter writer = null;
    try {
        jaasConf  = File.createTempFile("jaas", ".conf");
        writer = new PrintWriter(jaasConf);
        writer.println(content);
        
    } catch (IOException e) {
        e.printStackTrace();
    } finally {
        
          if (writer != null) {
              writer.close();
          }
          jaasConf.deleteOnExit();
    }
    System.setProperty("java.security.auth.login.config", jaasConf.getAbsolutePath());
    
}

应用

  • 实际线上使用时，考虑到数据传输效率和稳定性，要做以下优化。

    • 传输类为线程类，线程池管理，增加传输效率。

    • 批量上传数据。

    • 添加Callback处理机制，避免数据丢失。

  • 上传线程类如下。

public class Performance extends Thread{
	private final static Logger log = LoggerFactory.getLogger(Performance.class);
	
	
	private List> recordList;

	public Performance(List> recordList) {
		this.recordList=recordList;
	}
	
	/**
     *入库测试方法
     */
    public static void test() {
		 log.info("Kafka Tool Test");

	        try {
	            /* parse args */
	            String topicName ="test40";
	            /*总发包数*/
	            long numRecords = 10000000000L;
	            /*包大小*/
	            int recordSize = 1500;
	            /*每次最多发送包数*/
	            int throughput = 10000000;
	            

	            Properties props = new Properties();
	            props.put("acks", "1");
	    	    props.put("bootstrap.servers","ip:6667,ip:6667");
	    	    props.put("sasl.kerberos.service.name", "kafka");
	    	    props.put("security.protocol", "SASL_PLAINTEXT");
	    	    
	            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
	            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");

	            
	            KafkaProducer producer = new KafkaProducer(props);
	            

	            /* 创建测试数据 */
	            byte[] payload = new byte[recordSize];
	            Random random = new Random(0);
	            for (int i = 0; i < payload.length; ++i)
	                payload[i] = (byte) (random.nextInt(26) + 65);
	            
	            
	            /*创建测试数据发送对象*/
	            ProducerRecord record = new ProducerRecord(topicName, payload);
	            
	            /*测试数据模型  包总数*/
	            Stats stats = new Stats(numRecords, 5000);
	            
	            /*启动时间*/
	            long startMs = System.currentTimeMillis();
	            
	            /*帮助生成者发送流量类  每次最多发送包数  时间*/
	            ThroughputThrottler throttler = new ThroughputThrottler(throughput, startMs);
	            
	            for (int i = 0; i < numRecords; i++) {
	                long sendStartMs = System.currentTimeMillis();
	                Callback cb = stats.nextCompletion(sendStartMs, payload.length, stats,record.topic(),record.value());
	                producer.send(record, cb);

	                if (throttler.shouldThrottle(i, sendStartMs)) {
	                    throttler.throttle();
	                }
	            }

	            /* 结束任务 */
	            producer.close();
	            stats.printTotal();
	        } catch (Exception e) {
	            log.info("Test Error:"+e);
	        }

	}
	
    /**
     * 实际入库方法
     */
	@Override
	public void run() {
//	   log.info("Start To Send:");
	   super.run();
	   KafkaUtil kafkaUtil=new KafkaUtil();
	   KafkaProducer produce=kafkaUtil.create();
	   
	   //总包数
	   long size=recordList.size();
//	   		size=10000000000L;
	   
	   /*每次最多发送包数*/
	   int throughput = 900000;
//	       throughput = 10000000;
	       
	       
		 /*测试数据模型  包总数*/
       Stats stats = new Stats(size, 5000);
       /*启动时间*/
       long startMs = System.currentTimeMillis();
       
       /*帮助生成者发送流量类  每次最多发送包数  时间*/
       ThroughputThrottler throttler = new ThroughputThrottler(throughput, startMs);
       
       int i=0;
       for (ProducerRecord record:recordList) {
       	
           long sendStartMs = System.currentTimeMillis();
           //参数说明：发送数据时间  数据长度 数据模型类
           Callback cb = stats.nextCompletion(sendStartMs, record.value().length, stats,record.topic(),record.value());
           
           produce.send(record,cb);
           if (throttler.shouldThrottle(i, sendStartMs)) {
               throttler.throttle();
           }
           i++;
       }
       produce.close();
//       stats.printTotal();
//       log.info("End to Send");
       log.info("Finish Data To Send");
       LogModel.sendNum++;
	}
	
    private static class Stats {
        private long start;
        private long windowStart;
        private int[] latencies;
        private int sampling;
        private int iteration;
        private int index;
        private long count;
        private long bytes;
        private int maxLatency;
        private long totalLatency;
        private long windowCount;
        private int windowMaxLatency;
        private long windowTotalLatency;
        private long windowBytes;
        private long reportingInterval;

        public Stats(long numRecords, int reportingInterval) {
            this.start = System.currentTimeMillis();
            this.windowStart = System.currentTimeMillis();
            this.index = 0;
            this.iteration = 0;
            this.sampling = (int) (numRecords / Math.min(numRecords, 500000));
            this.latencies = new int[(int) (numRecords / this.sampling) + 1];
            this.index = 0;
            this.maxLatency = 0;
            this.totalLatency = 0;
            this.windowCount = 0;
            this.windowMaxLatency = 0;
            this.windowTotalLatency = 0;
            this.windowBytes = 0;
            this.totalLatency = 0;
            this.reportingInterval = reportingInterval;
        }

        public void record(int iter, int latency, int bytes, long time) {
            this.count++;
            this.bytes += bytes;
            this.totalLatency += latency;
            this.maxLatency = Math.max(this.maxLatency, latency);
            this.windowCount++;
            this.windowBytes += bytes;
            this.windowTotalLatency += latency;
            this.windowMaxLatency = Math.max(windowMaxLatency, latency);
            if (iter % this.sampling == 0) {
                this.latencies[index] = latency;
                this.index++;
            }
            /* maybe report the recent perf */
            if (time - windowStart >= reportingInterval) {
                printWindow();
                newWindow();
            }
        }

        public Callback nextCompletion(long start, int bytes, Stats stats,String topic,byte[] data) {
            Callback cb = new PerfCallback(this.iteration, start, bytes, stats,topic,data);
            this.iteration++;
            return cb;
        }

        /**
         * 传输效率反馈
         */
        public void printWindow() {
            long ellapsed = System.currentTimeMillis() - windowStart;
            double recsPerSec = 1000.0 * windowCount / (double) ellapsed;
            double mbPerSec = 1000.0 * this.windowBytes / (double) ellapsed / (1024.0 * 1024.0);
            System.out.printf("%d spend time,%d records sent, %.1f records/sec (%.2f MB/sec), %.1f ms avg latency, %.1f max latency.\n",
            		          ellapsed,
                              windowCount,
                              recsPerSec,
                              mbPerSec,
                              windowTotalLatency / (double) windowCount,
                              (double) windowMaxLatency);
        }

        public void newWindow() {
            this.windowStart = System.currentTimeMillis();
            this.windowCount = 0;
            this.windowMaxLatency = 0;
            this.windowTotalLatency = 0;
            this.windowBytes = 0;
        }

        /**
         * 传输效率
         */
        public void printTotal() {
            long elapsed = System.currentTimeMillis() - start;
            double recsPerSec = 1000.0 * count / (double) elapsed;
            double mbPerSec = 1000.0 * this.bytes / (double) elapsed / (1024.0 * 1024.0);
            int[] percs = percentiles(this.latencies, index, 0.5, 0.95, 0.99, 0.999);
            System.out.printf("%d spend time,%d records sent, %f records/sec (%.2f MB/sec), %.2f ms avg latency, %.2f ms max latency, %d ms 50th, %d ms 95th, %d ms 99th, %d ms 99.9th.\n",
                              elapsed,
            				  count,
                              recsPerSec,
                              mbPerSec,
                              totalLatency / (double) count,
                              (double) maxLatency,
                              percs[0],
                              percs[1],
                              percs[2],
                              percs[3]);
        }

        private static int[] percentiles(int[] latencies, int count, double... percentiles) {
            int size = Math.min(count, latencies.length);
            Arrays.sort(latencies, 0, size);
            int[] values = new int[percentiles.length];
            for (int i = 0; i < percentiles.length; i++) {
                int index = (int) (percentiles[i] * size);
                values[i] = latencies[index];
            }
            return values;
        }
    }

    private static final class PerfCallback implements Callback {
        private final long start;
        private final int iteration;
        private final int bytes;
        private final Stats stats;
        private final String topic;
        private final byte[] data;

        public PerfCallback(int iter, long start, int bytes, Stats stats,String topic,byte[] data) {
            this.start = start;
            this.stats = stats;
            this.iteration = iter;
            this.bytes = bytes;
            this.topic=topic;
            this.data=data;
        }

        public void onCompletion(RecordMetadata metadata, Exception exception) {
            long now = System.currentTimeMillis();
            int latency = (int) (now - start);
            this.stats.record(iteration, latency, bytes, now);
            if (exception != null){
            	ProducerRecord record=new ProducerRecord(topic,data);

                //将数据重新添加入数据队列，二次上传
            	ControlTask.recordList.add(record);
            	log.error("Send Error And Second To Send",exception);
            }
           
        }
    }

}

  • KafkaUtil.java

public class KafkaUtil {
//	private final static Logger log = LoggerFactory.getLogger(KafkaUtil.class);
	
	private KafkaProducer produce;

	/**
	 * 创建连接
	 * @return
	 */
	public KafkaProducer create(){
		Properties props = new Properties();
	    props.put("acks", "all");
	    props.put("bootstrap.servers", Config.ipList);
	    props.put("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
	    props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
//	    props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 120000);  //增加等待时间
	    props.put("retries", "2");
	    
	    //kerbores安全认证
	    if(Config.kerberos==0){
	    	
	    	props.put("security.protocol", "SASL_PLAINTEXT");
	    	props.put("sasl.mechanism", "GSSAPI");
	    	props.put("sasl.kerberos.service.name", "kafka");
	    	
	    }
	    
	    produce = new KafkaProducer(props);
	    
	    return produce;
	}
	
	/**
	 * 发送数据
	 * @param record
	 * @param cb
	 */
	public void send(ProducerRecord record,Callback cb){
		produce.send(record,cb);
	}
	
	/**
	 * 关闭连接
	 * @param produce
	 */
	public void close(){
		produce.flush();
		produce.close();
	}
	
	/**
	 * 生成jaas.conf临时文件
	 * @param keyTab  tab认证文件位置
	 * @param principal 认证用户
	 */
    public static void configureJAAS(String keyTab, String principal) {
    	String JAAS_TEMPLATE =
                "KafkaClient {\n"
                + "com.sun.security.auth.module.Krb5LoginModule required\n" +
                  "useKeyTab=true\n" +
                  "keyTab=\"%1$s\"\n" +
                  "principal=\"%2$s\";\n"
                + "};";
         String content = String.format(JAAS_TEMPLATE, keyTab, principal);
        
         File jaasConf = null;
         PrintWriter writer = null;
        try {
            jaasConf  = File.createTempFile("jaas", ".conf");
            writer = new PrintWriter(jaasConf);
            writer.println(content);
            
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            
             if (writer != null) {
                  writer.close();
             }
             jaasConf.deleteOnExit();
        }
        System.setProperty("java.security.auth.login.config", jaasConf.getAbsolutePath());
    }
}

感谢各位的阅读，以上就是“kafka javaAPI入库程序的实现方法”的内容了，经过本文的学习后，相信大家对kafka javaAPI入库程序的实现方法这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是创新互联，小编将为大家推送更多相关知识点的文章，欢迎关注！


名称栏目：kafkajavaAPI入库程序的实现方法
链接地址：http://cdiso.cn/article/jdeijc.html

其他资讯

联系小谭

企业微信号
友情链接： 乐山做网站   成都400电话   seo推广公司   成都模板网站   腾讯云免备案空间   成都搅拌罐车   南充网站建设   成都标牌标识牌制作公司   重庆移动网站建设   成都网站建设   

123ABC 成都网站建设，成都定制网站建设——全心全意建网站公司