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)


相关推荐

  • c中getline的用法_enum用法

    c中getline的用法_enum用法getline()用法getline是C++标准库函数;它有两种形式,一种是头文件<istream>中输入流成员函数;一种在头文件<string>中普通函数;它遇到以下情况发生会导致生成的本字符串结束:(1)到文件结束,(2)遇到函数的定界符,(3)输入达到最大限度。输入流成员函数getline()函数语法结构:在<istream>中的g…

    2022年10月26日
  • PHP 二维数组根据某个字段排序

    PHP 二维数组根据某个字段排序PHP 二维数组根据某个字段排序

  • hdu 1203 I NEED A OFFER!

    hdu 1203 I NEED A OFFER!

  • 均方误差(mean-square error, MSE)「建议收藏」

    均方误差(mean-square error, MSE)「建议收藏」全参考图像质量评价的方法有:PSNR峰值信噪比;SSIM结构相似性;MSE均方误差;我们接下来介绍一下均方误差。SSE在统计学里,该参数计算的是拟合数据与原始数据对应点的误差的平方和,计算公式为:其中是真实数据,是拟合数据,&gt;0,从这里可以看出SSE越接近于0,说明模型选择和拟合更好。MSE(均方方差)该统计参数是预测数据和原始数据对应点误差的平方和的均值…

  • 工业相机:传感器尺寸与像元尺寸的关系

    工业相机:传感器尺寸与像元尺寸的关系相同分辨率的工业相机,传感器面积越大,则其单位像素的面积也越大,成像质量也会越好。同样的500万像素的工业相机,2/3”的传感器成像质量就要优于1/2”的。一般来说,工业相机的靶面大小,如果要求不是太严格,可以直接按照公式计算。通常,像工业相机厂家会告诉你多少寸的传感器,那么就可以直接使用16mm*X”的传感器得到工业相机的图像传感器的靶面对角线长度。**工业相机使用16mm代表1寸,而不是使用25

  • 谈谈我对服务熔断、服务降级的理解

    谈谈我对服务熔断、服务降级的理解伴随着微服务架构被宣传得如火如荼,一些概念也被推到了我们面前(管你接受不接受),其实大多数概念以前就有,但很少被提的这么频繁(现在好像不提及都不好意思交流了)。想起有人总结的一句话,微服务架构的特点就是:“一解释就懂,一问就不知,一讨论就吵架”。其实对老外的总结能力一直特别崇拜,KevinKelly、MartinFowler、WernerVogels……,都是著名的“

发表回复

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

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