资讯

精准传达 • 有效沟通

从品牌网站建设到网络营销策划，从策略到执行的一站式服务

网站建设 >

查看其它板块

kafkajava生产消费程序demo示例

kafka是吞吐量巨大的一个消息系统，它是用scala写的，和普通的消息的生产消费还有所不同，写了个demo程序供大家参考。kafka的安装请参考官方文档。

成都创新互联公司是一家以网络技术公司，为中小企业提供网站维护、做网站、成都网站建设、网站备案、服务器租用、空间域名、软件开发、小程序制作等企业互联网相关业务，是一家有着丰富的互联网运营推广经验的科技公司，有着多年的网站建站经验，致力于帮助中小企业在互联网让打出自已的品牌和口碑，让企业在互联网上打开一个面向全国乃至全球的业务窗口：建站服务热线：13518219792

首先我们需要新建一个maven项目，然后在pom中引用kafka jar包，引用依赖如下：

1
2
3
4
5
    org.apache.kafka
    kafka_2.10
    0.8.0

我们用的版本是0.8， 下面我们看下生产消息的代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
package cn.outofmemory.kafka;
 
import java.util.Properties;
 
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
 
/**
 * Hello world!
 *
 */
public class KafkaProducer
{
    private final Producer producer;
    public final static String TOPIC = "TEST-TOPIC";
 
    private KafkaProducer(){
        Properties props = new Properties();
        //此处配置的是kafka的端口
        props.put("metadata.broker.list", "192.168.193.148:9092");
 
        //配置value的序列化类
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        //配置key的序列化类
        props.put("key.serializer.class", "kafka.serializer.StringEncoder");
 
        //request.required.acks
        //0, which means that the producer never waits for an acknowledgement from the broker (the same behavior as 0.7). This option provides the lowest latency but the weakest durability guarantees (some data will be lost when a server fails).
        //1, which means that the producer gets an acknowledgement after the leader replica has received the data. This option provides better durability as the client waits until the server acknowledges the request as successful (only messages that were written to the now-dead leader but not yet replicated will be lost).
        //-1, which means that the producer gets an acknowledgement after all in-sync replicas have received the data. This option provides the best durability, we guarantee that no messages will be lost as long as at least one in sync replica remains.
        props.put("request.required.acks","-1");
 
        producer = new Producer(new ProducerConfig(props));
    }
 
    void produce() {
        int messageNo = 1000;
        final int COUNT = 10000;
 
        while (messageNo < COUNT) {
            String key = String.valueOf(messageNo);
            String data = "hello kafka message " + key;
            producer.send(new KeyedMessage(TOPIC, key ,data));
            System.out.println(data);
            messageNo ++;
        }
    }
 
    public static void main( String[] args )
    {
        new KafkaProducer().produce();
    }
}

下面是消费端的代码实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
package cn.outofmemory.kafka;
 
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
 
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;
 
public class KafkaConsumer {
 
    private final ConsumerConnector consumer;
 
    private KafkaConsumer() {
        Properties props = new Properties();
        //zookeeper 配置
        props.put("zookeeper.connect", "192.168.193.148:2181");
 
        //group 代表一个消费组
        props.put("group.id", "jd-group");
 
        //zk连接超时
        props.put("zookeeper.session.timeout.ms", "4000");
        props.put("zookeeper.sync.time.ms", "200");
        props.put("auto.commit.interval.ms", "1000");
        props.put("auto.offset.reset", "smallest");
        //序列化类
        props.put("serializer.class", "kafka.serializer.StringEncoder");
 
        ConsumerConfig config = new ConsumerConfig(props);
 
        consumer = kafka.consumer.Consumer.createJavaConsumerConnector(config);
    }
 
    void consume() {
        Map topicCountMap = new HashMap();
        topicCountMap.put(KafkaProducer.TOPIC, new Integer(1));
 
        StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
        StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());
 
        Map>> consumerMap =
                consumer.createMessageStreams(topicCountMap,keyDecoder,valueDecoder);
        KafkaStream stream = consumerMap.get(KafkaProducer.TOPIC).get(0);
        ConsumerIterator it = stream.iterator();
        while (it.hasNext())
            System.out.println(it.next().message());
    }
 
    public static void main(String[] args) {
        new KafkaConsumer().consume();
    }
}

注意消费端需要配置成zk的地址，而生产端配置的是kafka的ip和端口。

源码地址获取：mingli

有兴趣的朋友们可以前往球球哦~一起分享学习技术：2042849237


当前名称：kafkajava生产消费程序demo示例
网页地址：http://cdkjz.cn/article/pgsjeh.html
返回首页 了解更多建站资讯

相关资讯

多年建站经验

多一份参考，总有益处

联系快上网，免费获得专属《策划方案》及报价

咨询相关问题或预约面谈，可以通过以下方式与我们联系

业务热线：400-028-6601 / 大客户专线   成都：13518219792   座机：028-86922220

在线咨询 提交需求
友情链接 交换友情链接
app开发广汉锦华橡塑成都app软件开发成都棕数机房成都响应式网站建设云南护栏打桩机成都免费建网站java虚拟主机域名注册时嘉做网站

成都网站建设公司地址：成都市青羊区太升南路288号锦天国际A座10层 建设咨询400-028-6601

成都快上网科技有限公司-四川网站建设设计公司 | 蜀ICP备19037934号 Copyright 2020,ALL Rights Reserved cdkjz.cn | 成都网站建设 | © Copyright 2020版权所有.

专家团队为您提供成都网站建设,成都网站设计,成都品牌网站设计,成都营销型网站制作等服务,成都建网站就找快上网！ | 成都网站建设哪家好？ | 网站建设地图