kafka0.8生产者实例

本文简单介绍下kafka0.8的client的producer的实例。

maven

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.10</artifactId>
            <version>0.8.2.2</version>
        </dependency>

初始化

        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,brokerAddr);
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "DemoProducer");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer<String, String> producer = new KafkaProducer<>(props)

准备消息

                String dataKey = UUID.randomUUID().toString();
                String dataValue = UUID.randomUUID().toString();

                ProducerRecord<String, String> producerRecord = new ProducerRecord<>(
                        topic,
                        dataKey,
                        dataValue
                );

发送

同步发送

producer.send(producerRecord).get();

异步发送(默认)

producer.send(producerRecord);

使用callback

producer.send(producerRecord, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        if(exception != null){
                            LOGGER.error("send msg to {},error:{}",metadata.topic(),exception);
                        }
                    }
                });

doc

  • kafka08-documentation
  • Kafka Clients (At-Most-Once, At-Least-Once, Exactly-Once, and Avro Client)

原文发布于微信公众号 - 码匠的流水账(geek_luandun)

原文发表时间:2017-09-28

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Albert陈凯

3.6 Shuffle机制

3.6 Shuffle机制 在MapReduce框架中,Shuffle是连接Map和Reduce之间的桥梁,Map的输出要用到Reduce中必须经过Shuff...

2794
来自专栏个人分享

Shuffle相关分析

 Shuffle描述是一个过程,表现出的是多对多的依赖关系。Shuffle是连接map阶段和Reduce阶段的纽带,每个Reduce Task都会从Map Ta...

953
来自专栏about云

spark零基础学习线路指导

问题导读 1.你认为spark该如何入门? 2.你认为spark入门编程需要哪些步骤? 3.本文介绍了spark哪些编程知识? ? spark...

3835
来自专栏个人分享

Hive架构及Hive On Spark

(1)Table:每个表都对应在HDFS中的目录下,数据是经过序列化后存储在该目录中。同时Hive也支持表中的数据存储在其他类型的文件系统中,如NFS或本地文件...

3452
来自专栏大数据架构

Adaptive Execution 让 Spark SQL 更高效更智能

前面《Spark SQL / Catalyst 内部原理 与 RBO》与《Spark SQL 性能优化再进一步 CBO 基于代价的优化》介绍的优化,从查询本身与...

991
来自专栏悦思悦读

Spark为什么比Hadoop快那么多?

在2014年11月5日举行的Daytona Gray Sort 100TB Benchmark竞赛中,Databricks 用构建于206个运算节点之上的spa...

48411
来自专栏Hadoop实操

SparkStreaming读Kafka数据写Kudu

1.4K2
来自专栏Albert陈凯

Spark系列课程-00xxSpark任务调度疑问,生成有向无环图的这个东西叫什么名字?

下面我们一起来看一下Spark的任务调度 ? Spark任务调度.png 首先最左边的叫做RDD Object就是一个一个的RDD对象 一个一个的RDD对象,...

43514
来自专栏Hadoop实操

Spark2Streaming读非Kerberos环境的Kafka并写数据到Kudu

在前面的文章Fayson介绍了在Kerberos环境下《Spark2Streaming读Kerberos环境的Kafka并写数据到Kudu》,本篇文章Fayso...

3071
来自专栏Java技术栈

Tomcat集群session复制与Oracle的坑。。

问题描述 公司某个系统使用了tomcat自带的集群session复制功能,然后后报了一个oracle驱动包里面的连接不能被序列化的异常。 01-Nov-2017...

3719

扫码关注云+社区