阿坝州网站建设公司

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://azwzsj.com/article/pgsjeh.html

其他资讯