kafuka生产者和消费者及配置

kafuka生产者和消费者及配置#kafka生产者配置#kafka集群kafka.bootstrap.servers=ip:端口#发送端确认模式kafka.acks=all#发送失败重试次数kafka.retries=10#批处理条数kafka.batch.size=16384#延迟统一收集,产生聚合,然后批量发送kafka.linger.ms=100#批处理缓冲区kafka.buffer.memo…

大家好,又见面了,我是你们的朋友全栈君。

#kafka 生产者配置
#kafka 集群
kafka.bootstrap.servers=ip:端口
#发送端确认模式
kafka.acks=all
#发送失败重试次数
kafka.retries =10
#批处理条数
kafka.batch.size=16384
#延迟统一收集,产生聚合,然后批量发送
kafka.linger.ms=100
#批处理缓冲区
kafka.buffer.memory=33554432
#key 序列化
kafka.key.serializer=org.apache.kafka.common.serialization.StringSerializer
#value序列化
kafka.value.serializer=org.apache.kafka.common.serialization.StringSerializer
#消费端 集群
kafka.bootstrap.servers=IP:端口
#一个用于跟踪调查的ID ,最好同group.id相同
kafka.client.id=MesSystem
#Consumer归属的组ID
kafka.group.id=debtorInfo
#限制每回返回的最大数据条数
kafka.max.poll.records=1000
#是否自动提交
kafka.enable.auto.commit=false
#自动提交的频率
kafka.auto.commit.interval.ms=1000
#会话的超时限制
kafka.session.timeout.ms=15000
kafka.key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
kafka.value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

//生产者
KafkaProducerUtils.send("topics", json.toString());//往kafka中存入消息
//KafkaProducerUtils 工具类
package com.tera.util;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.log4j.Logger;

import java.util.List;
import java.util.Properties;

public class KafkaProducerUtils {
    //把KafkaProducer对象放到本地线程中
    private static ThreadLocal<KafkaProducer> local = new ThreadLocal<KafkaProducer>();
    private static Properties props;
    private static KafkaProducer<String, String> producer;
    static {

        props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, PropertyUtil.getProperty("kafka.bootstrap.servers"));
        props.put(ProducerConfig.ACKS_CONFIG, PropertyUtil.getProperty("kafka.acks"));
        props.put(ProducerConfig.RETRIES_CONFIG, Integer.parseInt(PropertyUtil.getProperty("kafka.retries")));
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, Integer.parseInt(PropertyUtil.getProperty("kafka.batch.size")));
        props.put(ProducerConfig.LINGER_MS_CONFIG, Integer.parseInt(PropertyUtil.getProperty("kafka.linger.ms")));
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, Integer.parseInt(PropertyUtil.getProperty("kafka.buffer.memory")));
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, PropertyUtil.getProperty("kafka.key.serializer"));
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, PropertyUtil.getProperty("kafka.value.serializer"));
        producer = new KafkaProducer<String, String>(props);
        
    }
    
    static class SendCallback implements Callback {
        ProducerRecord<String, String> record;
        int sendSeq = 0;

        public SendCallback(ProducerRecord record, int sendSeq) {
            this.record = record;
            this.sendSeq = sendSeq;
        }
        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
            //send success
            if (null == e) {
                String meta = "send----topic:" + recordMetadata.topic() + ", partition:"
                        + recordMetadata.topic() + ", offset:" + recordMetadata.offset();
                System.out.println("send message success, record:" + record.toString() + ", meta:" + meta);
                
                System.out.println("value==========="+record.value());
                return;
            }
            //send failed
            System.out.println("send message failed, seq:" + sendSeq + ", record:" + record.toString() + ", errmsg:" + e.getMessage());
           
        }
    }

    /**
     * 发送消息到kafka
     * @param topicName
     * @param key
     * @param value
     */
    public static void send(String topicName,String value) throws Exception {
        if(StringUtils.isNullOrEmpty(topicName)){
            throw new Exception("参数错误,topicName不能为空");
        }
//        RecordMetadata recordMetadata =  producer.send(new ProducerRecord<String, String>(topicName,null,value)).get();
//        System.out.println("topic---"+recordMetadata.topic()+"--hasTimestamp---"+recordMetadata.hasTimestamp()+"--hasOffset"+
//        		recordMetadata.hasOffset()+"--partition--"+recordMetadata.partition()+"---"+recordMetadata.serializedKeySize()+"--"+recordMetadata.serializedValueSize()
//        		+"-----all--"+recordMetadata.toString()
//        		);
        ProducerRecord record= new ProducerRecord<String, String>(topicName,null,value);
       producer.send(record,new SendCallback(record,0));
        producer.flush();
    }
    /**
     * 发送消息到kafka
     * @param topicName
     * @param key
     * @param value
     */
    public static void sendBatch(String topicName,List<String> list) throws Exception {
        if(StringUtils.isNullOrEmpty(topicName)){
            throw new Exception("参数错误,topicName不能为空");
        }
        if(list==null || list.size() ==0){
            throw new Exception("参数错误,list不能为空");
        }
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
            for (String value : list){
                producer.send(new ProducerRecord<String, String>(topicName,null,value));
            }
            producer.close();


    }


    public static void main(String[] args) {
        KafkaProducerUtils kafkaProducerUtils = new KafkaProducerUtils();
        try {
            kafkaProducerUtils.send("withdrawaldev","123");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
 
//消费者
@Autowired
private DefaultKafkaConsumerFactory consumerFactory; 
Consumer consumer = consumerFactory.createConsumer();
consumer.subscribe(Arrays.asList("t_message_log"));
ConsumerRecords<Integer, String> records = null;
records = consumer.poll(100);
 for (ConsumerRecord<Integer, String> record : records) {	
		value = record.value();//数据
			        	
 }
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/136688.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • 深入理解Java类型信息(Class对象)与反射机制

    深入理解Java类型信息(Class对象)与反射机制【版权申明】未经博主同意,谢绝转载!(请尊重原创,博主保留追究权)http://blog.csdn.net/javazejian/article/details/70768369出自【zejian的博客】关联文章:深入理解Java类型信息(Class对象)与反射机制深入理解Java枚举类型(enum)深入理解Java注解类型(@Annotation)深入理解

  • Matlab中meshgrid的用法简介

    Matlab中meshgrid的用法简介meshgrid:网格1、主要使用的函数为[X,Y]=meshgrid(xgv,ygv);meshgrid函数生成的X,Y是大小相等的矩阵,xgv,ygv是两个网格矢量,xgv,ygv都是行向量。X:通过将xgv复制length(ygv)行(严格意义上是length(ygv)-1行)得到Y:首先对ygv进行转置得到ygv’,将ygv’复制(length(xgv)-1)次得到。…

  • aix dump_coredump文件分析

    aix dump_coredump文件分析Usingthedumpextractor,jextractTousethefullfacilitiesofthedumpviewer,youmustfirstrunthe jextract toolonthesystemdump.The jextract toolobtainsplatform-specificinformatio…

  • Hadoop切分纯文本时对某一行跨两个分片这种情况的处理

    Hadoop切分纯文本时对某一行跨两个分片这种情况的处理Hadoop切分纯文本时对某一行跨两个分片这种情况的处理

  • 逻辑斯蒂回归(Logistic Regression)

    logistic回归logistic回归经常被人译为“逻辑回归“,虽然我个人认为貌似并没有什么关联,但下面就姑且这么叫吧。逻辑回归虽然是名字里带着回归,但其实是一种解决分类问题的算法,说到分类就有分几类的区别,本篇我们只讨论用于二分类问题的逻辑回归。基本的线性回归的形式为:y=ωTx+by=ωTx+by=\omega^{T}x+b线性回归模型产生的预测值是一系列实值。为了使得输…

  • jdbc访问mysql(百度app下载)

     MySQL下载地址:http://dev.mysql.com/downloads/mysql/5.1.html#downloads 选择你对应的操作系统,然后选择这个系统下的第一个或者第二个,点击Pickamirror后即可下载 MySQL的JDBC驱动下载地址:http://dev.mysql.com/downloads/connector/在页面中选择你对应的开

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号